stream

package
v1.9.4 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 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.