code: func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogPublisher, func(), error) {
  	// TODO(stevvooe): The level of coordination here is WAY too much for logs.
  	// These should only be best effort and really just buffer until a session is
  	// ready. Ideally, they would use a separate connection completely.
  
  	var (
  		err       error
  		publisher api.LogBroker_PublishLogsClient
  	)
  
  	err = a.withSession(ctx, func(session *session) error {
  		publisher, err = api.NewLogBrokerClient(session.conn.ClientConn).PublishLogs(ctx)
  		return err
  	})
  	if err != nil {
  		return nil, nil, err
  	}
  
  	// make little closure for ending the log stream
  	sendCloseMsg := func() {
  		// send a close message, to tell the manager our logs are done
  		publisher.Send(&api.PublishLogsMessage{
  			SubscriptionID: subscriptionID,
  			Close:          true,
  		})
  		// close the stream forreal
  		publisher.CloseSend()
  	}
  
  	return exec.LogPublisherFunc(func(ctx context.Context, message api.LogMessage) error {
  			select {
  			case <-ctx.Done():
  				sendCloseMsg()
  				return ctx.Err()
  			default:
  			}
  
  			return publisher.Send(&api.PublishLogsMessage{
  				SubscriptionID: subscriptionID,
  				Messages:       []api.LogMessage{message},
  			})
  		}), func() {
  			sendCloseMsg()
  		}, nil
  }
docs: // Publisher returns a LogPublisher for the given subscription
  // as well as a cancel function that should be called when the log stream
  // is completed.