code: func (s *gRPCBrokerClientImpl) StartStream() error {
  	ctx, cancelFunc := context.WithCancel(context.Background())
  	defer cancelFunc()
  	defer s.Close()
  
  	stream, err := s.client.StartStream(ctx)
  	if err != nil {
  		return err
  	}
  	doneCh := stream.Context().Done()
  
  	go func() {
  		for {
  			select {
  			case <-doneCh:
  				return
  			case <-s.quit:
  				return
  			case se := <-s.send:
  				err := stream.Send(se.i)
  				se.ch <- err
  			}
  		}
  	}()
  
  	for {
  		i, err := stream.Recv()
  		if err != nil {
  			return err
  		}
  		select {
  		case <-doneCh:
  			return nil
  		case <-s.quit:
  			return nil
  		case s.recv <- i:
  		}
  	}
  
  	return nil
  }
docs: // StartStream implements the GRPCBrokerClient interface and will block until
  // the quit channel is closed or the context reports Done. The stream will pass
  // connection information to/from the plugin.