Method WorkspaceUpdates

tailnet/service.go:221–251

(req *proto.WorkspaceUpdatesRequest, stream proto.DRPCTailnet_WorkspaceUpdatesStream)

}

func (s *DRPCService) WorkspaceUpdates(req *proto.WorkspaceUpdatesRequest, stream proto.DRPCTailnet_WorkspaceUpdatesStream) error {
	defer stream.Close()

	ctx := stream.Context()

	ownerID, err := uuid.FromBytes(req.WorkspaceOwnerId)
	if err != nil {
		return xerrors.Errorf("parse workspace owner ID: %w", err)
	}

	sub, err := s.WorkspaceUpdatesProvider.Subscribe(ctx, ownerID)
	if err != nil {
		return xerrors.Errorf("subscribe to workspace updates: %w", err)
	}
	defer sub.Close()

	for {
		select {
		case updates, ok := <-sub.Updates():
			if !ok {
				return nil
			}
			err := stream.Send(updates)
			if err != nil {
				return xerrors.Errorf("send workspace update: %w", err)
			}
		case <-stream.Context().Done():
			return nil
		}
	}
}

type communicator struct {
	logger slog.Logger