code: func (lb *LogBroker) SubscribeLogs(request *api.SubscribeLogsRequest, stream api.Logs_SubscribeLogsServer) error {
  	ctx := stream.Context()
  
  	if err := validateSelector(request.Selector); err != nil {
  		return err
  	}
  
  	lb.mu.Lock()
  	pctx := lb.pctx
  	lb.mu.Unlock()
  	if pctx == nil {
  		return errNotRunning
  	}
  
  	subscription := lb.newSubscription(request.Selector, request.Options)
  	subscription.Run(pctx)
  	defer subscription.Stop()
  
  	log := log.G(ctx).WithFields(
  		logrus.Fields{
  			"method":          "(*LogBroker).SubscribeLogs",
  			"subscription.id": subscription.message.ID,
  		},
  	)
  	log.Debug("subscribed")
  
  	publishCh, publishCancel := lb.subscribe(subscription.message.ID)
  	defer publishCancel()
  
  	lb.registerSubscription(subscription)
  	defer lb.unregisterSubscription(subscription)
  
  	completed := subscription.Wait(ctx)
  	for {
  		select {
  		case <-ctx.Done():
  			return ctx.Err()
  		case <-pctx.Done():
  			return pctx.Err()
  		case event := <-publishCh:
  			publish := event.(*logMessage)
  			if publish.completed {
  				return publish.err
  			}
  			if err := stream.Send(&api.SubscribeLogsMessage{
  				Messages: publish.Messages,
  			}); err != nil {
  				return err
  			}
  		case <-completed:
  			completed = nil
  			lb.logQueue.Publish(&logMessage{
  				PublishLogsMessage: &api.PublishLogsMessage{
  					SubscriptionID: subscription.message.ID,
  				},
  				completed: true,
  				err:       subscription.Err(),
  			})
  		}
  	}
  }
docs: // SubscribeLogs creates a log subscription and streams back logs