code: func (lb *LogBroker) PublishLogs(stream api.LogBroker_PublishLogsServer) (err error) {
  	remote, err := ca.RemoteNode(stream.Context())
  	if err != nil {
  		return err
  	}
  
  	var currentSubscription *subscription
  	defer func() {
  		if currentSubscription != nil {
  			lb.markDone(currentSubscription, remote.NodeID, err)
  		}
  	}()
  
  	for {
  		logMsg, err := stream.Recv()
  		if err == io.EOF {
  			return stream.SendAndClose(&api.PublishLogsResponse{})
  		}
  		if err != nil {
  			return err
  		}
  
  		if logMsg.SubscriptionID == "" {
  			return status.Errorf(codes.InvalidArgument, "missing subscription ID")
  		}
  
  		if currentSubscription == nil {
  			currentSubscription = lb.getSubscription(logMsg.SubscriptionID)
  			if currentSubscription == nil {
  				return status.Errorf(codes.NotFound, "unknown subscription ID")
  			}
  		} else {
  			if logMsg.SubscriptionID != currentSubscription.message.ID {
  				return status.Errorf(codes.InvalidArgument, "different subscription IDs in the same session")
  			}
  		}
  
  		// if we have a close message, close out the subscription
  		if logMsg.Close {
  			// Mark done and then set to nil so if we error after this point,
  			// we don't try to close again in the defer
  			lb.markDone(currentSubscription, remote.NodeID, err)
  			currentSubscription = nil
  			return nil
  		}
  
  		// Make sure logs are emitted using the right Node ID to avoid impersonation.
  		for _, msg := range logMsg.Messages {
  			if msg.Context.NodeID != remote.NodeID {
  				return status.Errorf(codes.PermissionDenied, "invalid NodeID: expected=%s;received=%s", remote.NodeID, msg.Context.NodeID)
  			}
  		}
  
  		lb.publish(logMsg)
  	}
  }
docs: // PublishLogs publishes log messages for a given subscription