Documentation
¶
Overview ¶
Package streams implements the V1Streams gRPC service: durable, topic-based message streams.
Index ¶
- type Service
- type ServiceImpl
- func (s *ServiceImpl) CancelStreamSessions()
- func (s *ServiceImpl) Cleanup() error
- func (s *ServiceImpl) GetTopicMetadata(ctx context.Context, req *contracts.GetStreamTopicMetadataRequest) (*contracts.StreamTopicMetadata, error)
- func (s *ServiceImpl) Publish(ctx context.Context, req *contracts.PublishStreamMessageRequest) (*contracts.PublishStreamMessageResponse, error)
- func (s *ServiceImpl) PublishBatch(ctx context.Context, req *contracts.PublishStreamMessagesRequest) (*contracts.PublishStreamMessagesResponse, error)
- func (s *ServiceImpl) RegisterSubscribe(ctx context.Context, req *contracts.SubscribeStreamRequest) (*rpcstream.ChannelStream, error)
- func (s *ServiceImpl) Reset(ctx context.Context, req *contracts.ResetStreamTopicRequest) (*contracts.ResetStreamTopicResponse, error)
- func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStreamRequest, ...) error
- type ServiceOptFunc
- type ServiceOpts
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Service ¶
type Service interface {
v1connect.V1StreamsHandler
RegisterSubscribe(ctx context.Context, req *contracts.SubscribeStreamRequest) (*rpcstream.ChannelStream, error)
CancelStreamSessions()
Cleanup() error
}
Service is the durable streams gRPC service. CancelStreamSessions lets the engine hang up Subscribe RPCs on shutdown; RegisterSubscribe is Subscribe for a caller in the same process.
func NewService ¶
func NewService(fs ...ServiceOptFunc) (Service, error)
type ServiceImpl ¶
type ServiceImpl struct {
v1connect.UnimplementedV1StreamsHandler
// contains filtered or unexported fields
}
func (*ServiceImpl) CancelStreamSessions ¶
func (s *ServiceImpl) CancelStreamSessions()
CancelStreamSessions hangs up every Subscribe RPC. Safe to call repeatedly.
func (*ServiceImpl) Cleanup ¶
func (s *ServiceImpl) Cleanup() error
Cleanup stops the publisher's workers and the wake subscription.
func (*ServiceImpl) GetTopicMetadata ¶ added in v0.110.13
func (s *ServiceImpl) GetTopicMetadata(ctx context.Context, req *contracts.GetStreamTopicMetadataRequest) (*contracts.StreamTopicMetadata, error)
func (*ServiceImpl) Publish ¶
func (s *ServiceImpl) Publish(ctx context.Context, req *contracts.PublishStreamMessageRequest) (*contracts.PublishStreamMessageResponse, error)
func (*ServiceImpl) PublishBatch ¶ added in v0.114.1
func (s *ServiceImpl) PublishBatch(ctx context.Context, req *contracts.PublishStreamMessagesRequest) (*contracts.PublishStreamMessagesResponse, error)
func (*ServiceImpl) RegisterSubscribe ¶ added in v0.114.1
func (s *ServiceImpl) RegisterSubscribe(ctx context.Context, req *contracts.SubscribeStreamRequest) (*rpcstream.ChannelStream, error)
RegisterSubscribe serves Subscribe over channels to a caller in the same process (the in-process operator host), with every check the RPC makes: ctx must carry the tenant, as the auth middleware puts it, and a tenant without the durable streams entitlement is refused with PermissionDenied, which the stream's Err reports once it ends.
func (*ServiceImpl) Reset ¶ added in v0.112.0
func (s *ServiceImpl) Reset(ctx context.Context, req *contracts.ResetStreamTopicRequest) (*contracts.ResetStreamTopicResponse, error)
Reset makes subscribers without a cursor start after req.Cursor, or the topic's latest message. Subscriptions already open, and any with a cursor, are unaffected.
func (*ServiceImpl) Subscribe ¶
func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStreamRequest, connectStream *connect.ServerStream[contracts.StreamMessage]) error
type ServiceOptFunc ¶
type ServiceOptFunc func(*ServiceOpts)
func WithLogger ¶
func WithLogger(l *zerolog.Logger) ServiceOptFunc
func WithPubSub ¶
func WithPubSub(pubsub msgqueue.PubSub) ServiceOptFunc
func WithRepositoryV1 ¶
func WithRepositoryV1(r v1.Repository) ServiceOptFunc
type ServiceOpts ¶
type ServiceOpts struct {
// contains filtered or unexported fields
}
Source Files