Documentation
¶
Overview ¶
Package eventstream provides event streaming abstractions for server-to-client and bidirectional communication over HTTP, with implementations for SSE and WebSocket.
Index ¶
- Variables
- type BidirectionalEventStream
- type BidirectionalEventStreamUpgrader
- type Event
- type EventStream
- type EventStreamUpgrader
- type Option
- type StreamManager
- func (m *StreamManager[S]) Add(ctx context.Context, groupID, memberID string, stream S)
- func (m *StreamManager[S]) BroadcastToGroup(ctx context.Context, groupID string, event *Event) error
- func (m *StreamManager[S]) BroadcastToGroupFiltered(ctx context.Context, groupID string, event *Event, ...) error
- func (m *StreamManager[S]) Get(ctx context.Context, groupID, memberID string) S
- func (m *StreamManager[S]) GetGroupStreams(ctx context.Context, groupID string) []S
- func (m *StreamManager[S]) GetStreamCount(ctx context.Context, groupID string) int
- func (m *StreamManager[S]) GroupHasStreams(ctx context.Context, groupID string) bool
- func (m *StreamManager[S]) Remove(ctx context.Context, groupID, memberID string)
- func (m *StreamManager[S]) SendToMember(ctx context.Context, groupID, memberID string, event *Event) error
Constants ¶
This section is empty.
Variables ¶
var ErrNoSuchMember = platformerrors.New("no stream for that member")
ErrNoSuchMember is returned by SendToMember when the named member has no stream in the named group.
It exists because the alternative was reporting success. A caller sending to a member that has disconnected, or that it named wrongly, got a nil error and no delivery — and the two cases a caller most wants to tell apart, "delivered" and "there was nobody there", were the same answer.
Functions ¶
This section is empty.
Types ¶
type BidirectionalEventStream ¶
type BidirectionalEventStream interface {
EventStream
// Receive returns a channel of inbound events from the client.
Receive() <-chan *Event
}
BidirectionalEventStream extends EventStream with client-to-server receiving.
type BidirectionalEventStreamUpgrader ¶
type BidirectionalEventStreamUpgrader interface {
UpgradeToBidirectionalStream(w http.ResponseWriter, r *http.Request) (BidirectionalEventStream, error)
}
BidirectionalEventStreamUpgrader upgrades an HTTP connection to a BidirectionalEventStream.
type Event ¶
type Event struct {
Type string `json:"type"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Event represents a typed event with a JSON payload.
type EventStream ¶
type EventStream interface {
// Send pushes an event to the client.
Send(ctx context.Context, event *Event) error
// Done returns a channel that closes when the stream terminates.
Done() <-chan struct{}
// Close terminates the stream.
Close() error
}
EventStream is a unidirectional server-to-client event stream.
type EventStreamUpgrader ¶
type EventStreamUpgrader interface {
UpgradeToEventStream(w http.ResponseWriter, r *http.Request) (EventStream, error)
}
EventStreamUpgrader upgrades an HTTP connection to a unidirectional EventStream.
type Option ¶
type Option func(*options)
Option configures the StreamManager this package constructs. The zero configuration works: absent observability deps are normalized downstream.
func WithTracerProvider ¶
WithTracerProvider attaches a tracer provider.
type StreamManager ¶
type StreamManager[S EventStream] struct { // contains filtered or unexported fields }
StreamManager manages active event streams grouped by group ID and member ID.
func NewStreamManager ¶
func NewStreamManager[S EventStream](opts ...Option) *StreamManager[S]
NewStreamManager creates a new StreamManager.
func (*StreamManager[S]) Add ¶
func (m *StreamManager[S]) Add(ctx context.Context, groupID, memberID string, stream S)
Add registers a stream for a group and member.
func (*StreamManager[S]) BroadcastToGroup ¶
func (m *StreamManager[S]) BroadcastToGroup(ctx context.Context, groupID string, event *Event) error
BroadcastToGroup sends an event to all streams in a group, and returns the joined Send failures of the streams that did not take it.
A single stream's failure does not halt the broadcast: every stream in the snapshot is attempted, and the failures are joined at the end. A caller that gets a non-nil error therefore knows the event reached a subset, and can read which sends failed off the joined error; a group with no streams, or one where every send succeeded, returns nil.
func (*StreamManager[S]) BroadcastToGroupFiltered ¶
func (m *StreamManager[S]) BroadcastToGroupFiltered(ctx context.Context, groupID string, event *Event, includeFunc func(memberID string) bool) error
BroadcastToGroupFiltered sends an event to streams in a group for which includeFunc returns true, and returns the joined Send failures of the included streams that did not take it.
As with BroadcastToGroup, a single stream's failure does not halt the broadcast: every included stream in the snapshot is attempted, and the failures are joined at the end. Streams includeFunc excluded contribute nothing, so a filter that matches nobody returns nil, as does one where every included send succeeded.
func (*StreamManager[S]) Get ¶
func (m *StreamManager[S]) Get(ctx context.Context, groupID, memberID string) S
Get returns a specific stream, or the zero value if not found.
func (*StreamManager[S]) GetGroupStreams ¶
func (m *StreamManager[S]) GetGroupStreams(ctx context.Context, groupID string) []S
GetGroupStreams returns all streams for a group.
func (*StreamManager[S]) GetStreamCount ¶
func (m *StreamManager[S]) GetStreamCount(ctx context.Context, groupID string) int
GetStreamCount returns the number of streams for a group.
func (*StreamManager[S]) GroupHasStreams ¶
func (m *StreamManager[S]) GroupHasStreams(ctx context.Context, groupID string) bool
GroupHasStreams returns whether a group has any active streams.
func (*StreamManager[S]) Remove ¶
func (m *StreamManager[S]) Remove(ctx context.Context, groupID, memberID string)
Remove removes a stream.
func (*StreamManager[S]) SendToMember ¶
func (m *StreamManager[S]) SendToMember(ctx context.Context, groupID, memberID string, event *Event) error
SendToMember sends an event to a specific member in a group.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package eventstreamcfg selects and builds an eventstream upgrader from configuration: SSE or WebSocket.
|
Package eventstreamcfg selects and builds an eventstream upgrader from configuration: SSE or WebSocket. |
|
Package noop is the eventstream implementation for a caller with no transport to stream over.
|
Package noop is the eventstream implementation for a caller with no transport to stream over. |
|
Package sse upgrades an HTTP request to a Server-Sent Events stream.
|
Package sse upgrades an HTTP request to a Server-Sent Events stream. |
|
Package websocket upgrades an HTTP request to a WebSocket event stream, over gorilla/websocket.
|
Package websocket upgrades an HTTP request to a WebSocket event stream, over gorilla/websocket. |