code: def open(self):
          
          if self.is_active:
              raise ValueError("Can not open an already open stream.")
  
          request_generator = _RequestQueueGenerator(
              self._request_queue, initial_request=self._initial_request
          )
          call = self._start_rpc(iter(request_generator), metadata=self._rpc_metadata)
  
          request_generator.call = call
  
          # TODO: api_core should expose the future interface for wrapped
          # callables as well.
          if hasattr(call, "_wrapped"):  # pragma: NO COVER
              call._wrapped.add_done_callback(self._on_call_done)
          else:
              call.add_done_callback(self._on_call_done)
  
          self._request_generator = request_generator
          self.call = call
docs: Opens the stream.