channel

type EventHandler ¶

type EventHandler func(ctx context.Context, event *goclientnew.EventLog)

EventHandler reacts to one event the Channel subscribes to. Events are handled in delivery order on the poll goroutine, so a slow handler delays the next poll, which is the backpressure a bookmark that advances on delivery wants.

Delivery is best effort: the bookmark advances as events are delivered, so a handler can miss an event (a lost response, a process that dies mid-batch, a server that died before emitting, a reader behind the retention window), and nothing marks the gap. A reaction is a way to act sooner, never the source of truth: reconcile from state on start and on a schedule, and let a missed event cost latency rather than correctness.

type Handler ¶

type Handler func(ctx context.Context, run *Run) (json.RawMessage, error)

Handler executes one run of a command. It returns the result to report, a JSON value in the shape the command's signature declares, or nil for none; or an error, which reports the run Failed, or Rejected when it is a Rejection. The context ends at the run's delivered timeout, or at shutdown; a handler that is still working then has its result dropped and the run redelivered later, so handlers are idempotent or refuse a re-delivery.

type Listener ¶

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

Listener serves one Channel. Build it with NewListener, register handlers, then Run it.

func NewListener(client *cubapi.Client, ref Ref, opts ...Options) *Listener

func (l *Listener) Channel() *goclientnew.Channel

func (l *Listener) Commands() []string

func (l *Listener) Handle(command string, handler Handler) *Listener

func (l *Listener) OnEvent(handler EventHandler) *Listener

func (l *Listener) Run(ctx context.Context) error

func (l *Listener) Status() ListenerStatus

type ListenerStatus ¶

type ListenerStatus struct {
	Channel     string
	LeaseHeld   bool
	LastPoll    time.Time
	LastWork    time.Time
	InFlight    int
	Polls       int64
	Runs        int64
	Events      int64
	Concurrency int
}

ListenerStatus is what a probe reads: whether the lease is held, when the listener last polled and last had work, and what is in flight.

type Options ¶

type Options struct {
	// Concurrency is how many runs execute at once; each poll asks for exactly the free slots.
	// One is a serialising listener. Zero means DefaultConcurrency.
	Concurrency int
	// DrainTimeout is how long shutdown waits for in-flight handlers before cancelling them.
	// Zero means DefaultDrainTimeout.
	DrainTimeout time.Duration
	// Logger receives the listener's log lines. Nil means the standard logger.
	Logger *log.Logger
}

Options tune a Listener.

type Ref ¶

type Ref = cubapi.Ref

Ref names the Channel to listen on: a slug, a space/slug, or a UUID, as cubapi resolves it.

type Rejection ¶

type Rejection struct {
	Reason string
}

Rejection is the error a Handler returns to report a run Rejected rather than Failed: the handler did not try, because the command or its params are not ones it accepts.

func (r *Rejection) Error() string

type Run ¶

type Run struct {
	*goclientnew.CommandRun
	// Client is the listener's client, so that what a handler does back against ConfigHub
	// goes over the same session.
	Client *cubapi.Client
	// contains filtered or unexported fields
}

Run is one delivered run, with its params decoded for the handler.

func (r *Run) Bool(name string) bool

func (r *Run) Has(name string) bool

func (r *Run) Int(name string) int64

func (r *Run) Param(name string, target any) (bool, error)

func (r *Run) String(name string) string

type Status ¶

type Status string

Status is where a run is in its life, as the server reports it.

const (
	StatusPending   Status = "Pending"
	StatusDelivered Status = "Delivered"
	StatusCompleted Status = "Completed"
	StatusFailed    Status = "Failed"
	StatusRejected  Status = "Rejected"
	StatusCanceled  Status = "Canceled"
	StatusExpired   Status = "Expired"
	StatusTimedOut  Status = "TimedOut"
)

func (s Status) IsTerminal() bool