Help us shape the future of Go! Take the 2026 Go Developer Survey

stream

package
v1.9.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 26, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package stream provides a real-time event broker for Dispatch lifecycle events. It bridges the ext.Extension system to connected clients via topic-based pub/sub.

Index

Constants

View Source 
const (
	TopicJobs      = "jobs"
	TopicWorkflows = "workflows"
	TopicFirehose  = "firehose"
)
View Source 
const DefaultBufferSize = 256

DefaultBufferSize is the default per-subscriber event buffer.

View Source 
const DefaultCredits int64 = 1000

DefaultCredits is the default initial credits for new subscribers.

Variables

This section is empty.

Functions

func JobTopic 

func JobTopic(jobID string) string

JobTopic returns the topic name for a specific job.

func ParseTopicEntity 

func ParseTopicEntity(topic string) (entityType, entityID string)

ParseTopicEntity extracts the entity type and ID from a topic string. For example, "job:job_abc123" returns ("job", "job_abc123"). Returns ("", "") for global topics like "jobs" or "firehose".

func QueueTopic 

func QueueTopic(queue string) string

QueueTopic returns the topic name for a queue.

func ValidateTopic 

func ValidateTopic(topic string) error

ValidateTopic checks whether a topic string is valid.

func WorkflowTopic 

func WorkflowTopic(runID string) string

WorkflowTopic returns the topic name for a specific workflow run.

Types

type Broker 

type Broker struct {
	// contains filtered or unexported fields
}

Broker is the real-time stream broker. It implements the ext.Extension interface to receive lifecycle events and fans them out to subscribers via topic-based pub/sub.

func NewBroker 

func NewBroker(logger log.Logger, opts ...BrokerOption) *Broker

NewBroker creates a new stream broker.

func (*Broker) GetSubscriber 

func (b *Broker) GetSubscriber(subscriberID string) (*Subscriber, bool)

GetSubscriber returns a subscriber by ID.

func (*Broker) Name 

func (b *Broker) Name() string

Name implements ext.Extension.

func (*Broker) OnCronFired 

func (b *Broker) OnCronFired(_ context.Context, entryName string, jobID id.JobID) error

func (*Broker) OnJobCancelled 

func (b *Broker) OnJobCancelled(_ context.Context, j *job.Job) error

OnJobCancelled implements ext.JobCancelled. It goes to the same topics as job.failed: the job's own topic, jobs, and the firehose.

func (*Broker) OnJobCompleted 

func (b *Broker) OnJobCompleted(_ context.Context, j *job.Job, elapsed time.Duration) error

func (*Broker) OnJobDLQ 

func (b *Broker) OnJobDLQ(_ context.Context, j *job.Job, jobErr error) error

func (*Broker) OnJobEnqueued 

func (b *Broker) OnJobEnqueued(_ context.Context, j *job.Job) error

func (*Broker) OnJobFailed 

func (b *Broker) OnJobFailed(_ context.Context, j *job.Job, jobErr error) error

func (*Broker) OnJobRetrying 

func (b *Broker) OnJobRetrying(_ context.Context, j *job.Job, attempt int, nextRunAt time.Time) error

func (*Broker) OnJobStarted 

func (b *Broker) OnJobStarted(_ context.Context, j *job.Job) error

func (*Broker) OnShutdown 

func (b *Broker) OnShutdown(_ context.Context) error

func (*Broker) OnWorkflowCompleted 

func (b *Broker) OnWorkflowCompleted(_ context.Context, r *workflow.Run, elapsed time.Duration) error

func (*Broker) OnWorkflowFailed 

func (b *Broker) OnWorkflowFailed(_ context.Context, r *workflow.Run, runErr error) error

func (*Broker) OnWorkflowStarted 

func (b *Broker) OnWorkflowStarted(_ context.Context, r *workflow.Run) error

func (*Broker) OnWorkflowStepCompleted 

func (b *Broker) OnWorkflowStepCompleted(_ context.Context, r *workflow.Run, stepName string, elapsed time.Duration) error

func (*Broker) OnWorkflowStepFailed 

func (b *Broker) OnWorkflowStepFailed(_ context.Context, r *workflow.Run, stepName string, stepErr error) error

func (*Broker) RemoveSubscriber 

func (b *Broker) RemoveSubscriber(subscriberID string)

RemoveSubscriber removes a subscriber from all topics and closes it.

func (*Broker) Stats 

func (b *Broker) Stats() BrokerStats

Stats returns broker statistics.

func (*Broker) Subscribe 

func (b *Broker) Subscribe(subscriberID string, topics ...string) *Subscriber

Subscribe creates a new subscriber on the given topics.

func (*Broker) SubscribeTo 

func (b *Broker) SubscribeTo(subscriberID string, topics ...string)

SubscribeTo adds an existing subscriber to additional topics.

func (*Broker) Topics 

func (b *Broker) Topics() *TopicRegistry

Topics returns the topic registry for external use (e.g., DWP server).

func (*Broker) Unsubscribe 

func (b *Broker) Unsubscribe(subscriberID string, topics ...string)

Unsubscribe removes a subscriber from specific topics.

type BrokerOption 

type BrokerOption func(*Broker)

BrokerOption configures a Broker.

func WithBufferSize 

func WithBufferSize(size int) BrokerOption

WithBufferSize sets the per-subscriber event buffer size.

func WithDefaultCredits 

func WithDefaultCredits(credits int64) BrokerOption

WithDefaultCredits sets the initial credits for new subscribers.

type BrokerStats 

type BrokerStats struct {
	TopicCount      int   `json:"topic_count"`
	SubscriberCount int   `json:"subscriber_count"`
	TotalPublished  int64 `json:"total_published"`
	TotalDropped    int64 `json:"total_dropped"`
}

BrokerStats contains broker metrics.

type CronEventData 

type CronEventData struct {
	EntryName string `json:"entry_name"`
	JobID     string `json:"job_id"`
}

CronEventData is the payload for cron lifecycle events.

type Event 

type Event struct {
	// Type identifies the lifecycle event.
	Type EventType `json:"type"`

	// Timestamp is when the event was emitted.
	Timestamp time.Time `json:"ts"`

	// Topic is the channel this event was published on.
	Topic string `json:"topic"`

	// Data is the event-specific payload.
	Data json.RawMessage `json:"data"`
}

Event is the envelope sent to subscribers on a topic channel.

type EventType 

type EventType string

EventType identifies the kind of lifecycle event. 

const (
	// Job events.
	EventJobEnqueued  EventType = "job.enqueued"
	EventJobStarted   EventType = "job.started"
	EventJobCompleted EventType = "job.completed"
	EventJobFailed    EventType = "job.failed"
	EventJobRetrying  EventType = "job.retrying"
	EventJobDLQ       EventType = "job.dlq"
	EventJobCancelled EventType = "job.cancelled"

	// Workflow events.
	EventWorkflowStarted       EventType = "workflow.started"
	EventWorkflowStepCompleted EventType = "workflow.step_completed"
	EventWorkflowStepFailed    EventType = "workflow.step_failed"
	EventWorkflowCompleted     EventType = "workflow.completed"
	EventWorkflowFailed        EventType = "workflow.failed"

	// Cron events.
	EventCronFired EventType = "cron.fired"
)

type JobEventData 

type JobEventData struct {
	JobID      string `json:"job_id"`
	JobName    string `json:"job_name"`
	Queue      string `json:"queue"`
	ScopeAppID string `json:"scope_app_id,omitempty"`
	ScopeOrgID string `json:"scope_org_id,omitempty"`
	ElapsedMs  int64  `json:"elapsed_ms,omitempty"`
	Error      string `json:"error,omitempty"`
	Attempt    int    `json:"attempt,omitempty"`
	NextRunAt  string `json:"next_run_at,omitempty"`
}

JobEventData is the payload for job lifecycle events.

type Subscriber 

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber receives events from topics it is subscribed to. It uses credit-based flow control: the subscriber grants credits indicating how many events it can receive. The broker stops sending when credits reach zero.

func NewSubscriber 

func NewSubscriber(id string, bufferSize int, initialCredits int64) *Subscriber

NewSubscriber creates a subscriber with the given buffer size and initial credits.

func (*Subscriber) AddCredits 

func (s *Subscriber) AddCredits(n int64)

AddCredits replenishes flow-control credits.

func (*Subscriber) C 

func (s *Subscriber) C() <-chan *Event

C returns the read-only event channel.

func (*Subscriber) Close 

func (s *Subscriber) Close()

Close closes the subscriber channel. Safe to call multiple times.

func (*Subscriber) Credits 

func (s *Subscriber) Credits() int64

Credits returns the current credit count.

func (*Subscriber) ID 

func (s *Subscriber) ID() string

ID returns the subscriber identifier.

func (*Subscriber) SetFilter 

func (s *Subscriber) SetFilter(fn func(*Event) bool)

SetFilter sets an optional event filter predicate.

func (*Subscriber) Topics 

func (s *Subscriber) Topics() []string

Topics returns a copy of all subscribed topic names.

type TopicRegistry 

type TopicRegistry struct {
	// contains filtered or unexported fields
}

TopicRegistry manages subscriber sets per topic. It is safe for concurrent use.

func NewTopicRegistry 

func NewTopicRegistry() *TopicRegistry

NewTopicRegistry creates an empty topic registry.

func (*TopicRegistry) Broadcast 

func (tr *TopicRegistry) Broadcast(topics []string, evt *Event) int

Broadcast sends an event to all subscribers on multiple topics. Deduplicates subscribers that are on more than one of the listed topics.

func (*TopicRegistry) Publish 

func (tr *TopicRegistry) Publish(topic string, evt *Event) int

Publish sends an event to all subscribers on the given topic. Returns the number of subscribers that received the event.

func (*TopicRegistry) Subscribe 

func (tr *TopicRegistry) Subscribe(topic string, sub *Subscriber)

Subscribe adds a subscriber to a topic. Creates the topic if it doesn't exist.

func (*TopicRegistry) SubscriberCount 

func (tr *TopicRegistry) SubscriberCount(topic string) int

SubscriberCount returns the number of subscribers on a topic.

func (*TopicRegistry) TopicCount 

func (tr *TopicRegistry) TopicCount() int

TopicCount returns the number of active topics.

func (*TopicRegistry) Unsubscribe 

func (tr *TopicRegistry) Unsubscribe(topic, subscriberID string)

Unsubscribe removes a subscriber from a topic. Cleans up empty topics.

func (*TopicRegistry) UnsubscribeAll 

func (tr *TopicRegistry) UnsubscribeAll(subscriberID string)

UnsubscribeAll removes a subscriber from all topics.

type WorkflowEventData 

type WorkflowEventData struct {
	RunID      string `json:"run_id"`
	Name       string `json:"name"`
	StepName   string `json:"step_name,omitempty"`
	ScopeAppID string `json:"scope_app_id,omitempty"`
	ScopeOrgID string `json:"scope_org_id,omitempty"`
	ElapsedMs  int64  `json:"elapsed_ms,omitempty"`
	Error      string `json:"error,omitempty"`
}

WorkflowEventData is the payload for workflow lifecycle events.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.