Documentation
¶
Overview ¶
Package live provides realtime dashboard event fan-out: every event this process publishes, plus, in a multi-replica deployment, the audit and usage rows its peers persist (see StartRemote).
Index ¶
- Constants
- type Broker
- func (b *Broker) Close()
- func (b *Broker) Enabled() bool
- func (b *Broker) HasLiveSubscribers() bool
- func (b *Broker) Heartbeat() time.Duration
- func (b *Broker) LatestSeq() uint64
- func (b *Broker) PublishAuditEvent(eventType string, entry *auditlog.LogEntry)
- func (b *Broker) PublishRemoteAuditEntry(entry *auditlog.EntryPreview, writtenAt int64)
- func (b *Broker) PublishRemoteUsageEntry(entry usage.UsageLogEntry, writtenAt int64)
- func (b *Broker) PublishUsageEvent(eventType string, entry *usage.UsageEntry)
- func (b *Broker) StartRemote(cfg RemoteConfig)
- func (b *Broker) Subscribe(cursor uint64) *Subscription
- type Config
- type Event
- type RemoteConfig
- type Subscription
Constants ¶
const ( EventAuditStarted = auditlog.LiveEventAuditStarted EventAuditUpdated = auditlog.LiveEventAuditUpdated EventAuditStream = auditlog.LiveEventAuditStream EventAuditCompleted = auditlog.LiveEventAuditCompleted EventAuditFailed = auditlog.LiveEventAuditFailed EventAuditFlushed = auditlog.LiveEventAuditFlushed EventAuditRemoved = auditlog.LiveEventAuditRemoved EventUsageCompleted = usage.LiveEventUsageCompleted EventUsageFailed = usage.LiveEventUsageFailed EventUsageFlushed = usage.LiveEventUsageFlushed EventHeartbeat = "heartbeat" EventReset = "reset" )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker stores a bounded replay window and fans live events out to subscribers.
func (*Broker) Close ¶
func (b *Broker) Close()
Close terminates all active subscribers and prevents future live events.
func (*Broker) HasLiveSubscribers ¶
HasLiveSubscribers reports whether any dashboard stream is currently connected. Publishers of high-frequency preview events use it to skip building payloads nobody would receive.
func (*Broker) PublishAuditEvent ¶
PublishAuditEvent publishes a compact audit log preview event. Connected subscribers receive the full preview, including any captured bodies; the copy retained for replay strips bodies (flagging them as captured so the dashboard hydrates them from the persisted entry) and is size-capped, so broker retention stays bounded regardless of request size.
The full preview is encoded only while someone is subscribed. Bodies are the bulk of the payload, and with no dashboard open (the common case) nobody would read them. A subscriber that connects between that check and the publish receives the retained copy, which is what replay gives it anyway.
func (*Broker) PublishRemoteAuditEntry ¶ added in v0.2.2
func (b *Broker) PublishRemoteAuditEntry(entry *auditlog.EntryPreview, writtenAt int64)
PublishRemoteAuditEntry publishes an audit entry another replica persisted, as the audit.flushed event that replica's own dashboards received, marked with its write stamp. The preview carries no bodies, only whether each was captured, so remote traffic costs the stream only the compact preview; the dashboard loads them from the stored entry.
func (*Broker) PublishRemoteUsageEntry ¶ added in v0.2.2
func (b *Broker) PublishRemoteUsageEntry(entry usage.UsageLogEntry, writtenAt int64)
PublishRemoteUsageEntry publishes a usage entry another replica persisted, as the usage.flushed event that replica's own dashboards received, marked with its write stamp.
func (*Broker) PublishUsageEvent ¶
func (b *Broker) PublishUsageEvent(eventType string, entry *usage.UsageEntry)
PublishUsageEvent publishes a compact usage log event. Cached usage entries are broadcast like any other so the dashboard can choose to surface or hide them via the "Hide cached requests" toggle on the Usage page.
func (*Broker) StartRemote ¶ added in v0.2.2
func (b *Broker) StartRemote(cfg RemoteConfig)
StartRemote starts publishing the audit and usage entries other replicas persist, polling while at least one subscriber is connected. Close stops it. It does nothing on a disabled or closed broker, without a reader, or when already started.
func (*Broker) Subscribe ¶
func (b *Broker) Subscribe(cursor uint64) *Subscription
Subscribe registers a client and returns replay events after cursor.
type Config ¶
type Config struct {
Enabled bool
BufferSize int
ReplayLimit int
SubscriberBuffer int
Heartbeat time.Duration
}
Config controls the in-process live event broker.
type Event ¶
type Event struct {
Seq uint64 `json:"seq"`
Type string `json:"type"`
RequestID string `json:"request_id,omitempty"`
Timestamp time.Time `json:"timestamp"`
Data json.RawMessage `json:"data,omitempty"`
}
Event is the stable envelope sent over the live dashboard stream.
type RemoteConfig ¶ added in v0.2.2
type RemoteConfig struct {
// Writer is this process's writer id (storage.WriterID): entries it
// stamped were already published live and are skipped.
Writer string
// Audit and Usage read the stamped entries. Either may be nil.
Audit auditlog.RecentWriteReader
Usage usage.RecentWriteReader
}
RemoteConfig names the persisted traffic StartRemote publishes.
type Subscription ¶
type Subscription struct {
Replay []Event
Events <-chan Event
Reset bool
// Latest is the newest sequence assigned when the subscription was
// created, captured atomically with Replay: replay never goes past it,
// and anything newer arrives on Events. Streams start their heartbeat
// cursor from it — the broker-global counter read later can already
// cover events still queued on Events, and advertising those would let
// a reconnecting client skip them.
Latest uint64
// contains filtered or unexported fields
}
Subscription is one live stream consumer.
Source Files