code: def open(self, callback):
          
          if self.is_active:
              raise ValueError("This manager is already open.")
  
          if self._closed:
              raise ValueError("This manager has been closed and can not be re-used.")
  
          self._callback = functools.partial(_wrap_callback_errors, callback)
  
          # Create the RPC
          self._rpc = bidi.ResumableBidiRpc(
              start_rpc=self._client.api.streaming_pull,
              initial_request=self._get_initial_request,
              should_recover=self._should_recover,
          )
          self._rpc.add_done_callback(self._on_rpc_done)
  
          # Create references to threads
          self._dispatcher = dispatcher.Dispatcher(self, self._scheduler.queue)
          self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response)
          self._leaser = leaser.Leaser(self)
          self._heartbeater = heartbeater.Heartbeater(self)
  
          # Start the thread to pass the requests.
          self._dispatcher.start()
  
          # Start consuming messages.
          self._consumer.start()
  
          # Start the lease maintainer thread.
          self._leaser.start()
  
          # Start the stream heartbeater thread.
          self._heartbeater.start()
docs: Begin consuming messages.
  
          Args:
              callback (Callable[None, google.cloud.pubsub_v1.message.Messages]):
                  A callback that will be called for each message received on the
                  stream.