Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrBroadcasterClosed = errors.New("broadcaster closed")
ErrBroadcasterClosed is returned when Subscribe is called on a closed broadcaster.
Functions ¶
This section is empty.
Types ¶
type Broadcaster ¶
type Broadcaster[T any] struct { // contains filtered or unexported fields }
Broadcaster distributes events to multiple subscribers with bounded buffers.
func NewBroadcaster ¶
func NewBroadcaster[T any](metrics BroadcasterMetrics, opts ...BroadcasterOption) *Broadcaster[T]
NewBroadcaster creates a new bounded broadcaster.
func (*Broadcaster[T]) Close ¶ added in v0.7.0
func (b *Broadcaster[T]) Close()
Close cleans up all active subscribers and closes their channels. Safe to call multiple times. After Close, Subscribe returns ErrBroadcasterClosed.
func (*Broadcaster[T]) IsClosed ¶ added in v0.7.0
func (b *Broadcaster[T]) IsClosed() bool
IsClosed reports whether the broadcaster has been closed (server shutdown). Watch handlers use this to distinguish shutdown from slow-consumer drops.
func (*Broadcaster[T]) Publish ¶
func (b *Broadcaster[T]) Publish(event Event[T])
Publish sends an event to all subscribers. If a subscriber's buffer is full, it is dropped. No-op if the broadcaster is closed.
func (*Broadcaster[T]) Subscribe ¶
func (b *Broadcaster[T]) Subscribe(ctx context.Context) (*Subscriber[T], error)
Subscribe creates a new subscriber with a bounded channel. When the buffer is full, the subscriber is dropped (closed). Returns ErrBroadcasterClosed if the broadcaster has been closed.
func (*Broadcaster[T]) SubscriberCount ¶
func (b *Broadcaster[T]) SubscriberCount() int
SubscriberCount returns number of active subscribers.
func (*Broadcaster[T]) Unsubscribe ¶
func (b *Broadcaster[T]) Unsubscribe(id string)
Unsubscribe removes a subscriber.
type BroadcasterMetrics ¶ added in v0.5.0
type BroadcasterMetrics struct {
IncSubscribers func()
DecSubscribers func()
IncEvents func()
IncDrops func()
}
BroadcasterMetrics defines optional callbacks for observability.
type BroadcasterOption ¶ added in v0.7.0
type BroadcasterOption func(*broadcasterConfig)
BroadcasterOption configures a Broadcaster.
func WithBufferSize ¶ added in v0.7.0
func WithBufferSize(n int) BroadcasterOption
WithBufferSize sets the subscriber channel buffer size. Defaults to 1024 if not specified.
type Event ¶
type Event[T any] struct { Type string // ADDED, MODIFIED, DELETED Object T Version string // resourceVersion }
Event represents a watch event.
type InformerManager ¶
type InformerManager struct {
EnvBroadcaster *Broadcaster[*divergev1alpha1.Environment]
PgBroadcaster *Broadcaster[*divergev1alpha1.PreviewGroup]
// contains filtered or unexported fields
}
InformerManager manages shared informers and broadcasts events.
func NewInformerManager ¶
func NewInformerManager(logger *slog.Logger, metrics BroadcasterMetrics) *InformerManager
NewInformerManager creates informer event handlers and connects them to broadcasters.
func (*InformerManager) Close ¶ added in v0.7.0
func (m *InformerManager) Close()
Close cleanly shuts down all broadcasters, unblocking active watch streams.
func (*InformerManager) HandleEnvironmentEvent ¶
func (m *InformerManager) HandleEnvironmentEvent(eventType string, obj client.Object)
HandleEnvironmentEvent handles an environment event from an informer.
func (*InformerManager) HandlePreviewGroupEvent ¶
func (m *InformerManager) HandlePreviewGroupEvent(eventType string, obj client.Object)
HandlePreviewGroupEvent handles a preview group event from an informer.
type LogStreamer ¶
type LogStreamer struct {
// contains filtered or unexported fields
}
LogStreamer streams logs from pods.
func NewLogStreamer ¶
func NewLogStreamer(clientset kubernetes.Interface) *LogStreamer
func (*LogStreamer) StreamPodLogs ¶
func (l *LogStreamer) StreamPodLogs(ctx context.Context, namespace, podName, container string, follow bool, tailLines int64, since *time.Time) (io.ReadCloser, error)
StreamPodLogs streams logs from a specific pod/container.
type Subscriber ¶
type Subscriber[T any] struct { // contains filtered or unexported fields }
Subscriber is a registered watch subscriber.
func (*Subscriber[T]) Events ¶
func (s *Subscriber[T]) Events() <-chan Event[T]
Events returns the subscriber's event channel.
Source Files