remote

remote

import "github.com/agentstation/starmap/remote"

Package remote provides a reactive Starmap catalog consumer.

Index

• Constants
• type Config
• type Health
• type HealthError
• type PollingFallbackPolicy
• type PollingFallbackStatus
• type StreamState
• type Subscriber
  • func New(config Config) (*Subscriber, error)
  • func (s *Subscriber) Catalog() *catalogs.Catalog
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Health() Health
  • func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus
  • func (s *Subscriber) Start(ctx context.Context) error

Constants

const (
    // DefaultReconnectMinDelay is the first reconnect delay.
    DefaultReconnectMinDelay = 100 * time.Millisecond
    // DefaultReconnectMaxDelay bounds reconnect delay growth.
    DefaultReconnectMaxDelay = 5 * time.Second
    // DefaultExpectedHeartbeatInterval matches the server's default heartbeat.
    DefaultExpectedHeartbeatInterval = 20 * time.Second
    // DefaultLivenessTimeout bounds a stream with no heartbeat or event.
    DefaultLivenessTimeout = 60 * time.Second
    // DefaultShutdownTimeout bounds Close while joining owned loops.
    DefaultShutdownTimeout = 5 * time.Second
)

type Config

Config defines one remote Starmap catalog source. BaseURL is the versioned API root, for example https://starmap.example.com/api/v1.

type Config struct {
    // BaseURL is the trusted absolute HTTPS versioned Starmap API root.
    // Plain HTTP is accepted only for a loopback publisher.
    BaseURL string
    // HTTPClient supplies transport, TLS, authentication, and fetch timeout
    // policy. A private bounded client is used when nil.
    HTTPClient *http.Client
    // ReconnectMinDelay is the first reconnect delay. Zero selects the default.
    ReconnectMinDelay time.Duration
    // ReconnectMaxDelay bounds exponential reconnect delay. Zero selects the
    // default.
    ReconnectMaxDelay time.Duration
    // ExpectedHeartbeatInterval is the configured server heartbeat interval.
    // Zero selects the server's default.
    ExpectedHeartbeatInterval time.Duration
    // LivenessTimeout is the maximum time without a comment or publication
    // frame. Zero selects the default.
    LivenessTimeout time.Duration
    // ShutdownTimeout bounds Close while it joins subscriber-owned loops. Zero
    // selects the default.
    ShutdownTimeout time.Duration
    // PollingFallback explicitly enables bounded conditional polling after
    // repeated streaming failures. Nil keeps polling disabled.
    PollingFallback *PollingFallbackPolicy
}

type Health

Health is an immutable snapshot of subscriber transport and catalog health. Stream activity and catalog freshness are independent: heartbeats never change CatalogGeneratedAt or CatalogAgeSeconds.

type Health struct {
    StreamState             StreamState           `json:"stream_state"`
    ActiveGenerationID      string                `json:"active_generation_id,omitempty"`
    CatalogGeneratedAt      time.Time             `json:"catalog_generated_at"`
    CatalogAgeSeconds       int64                 `json:"catalog_age_seconds"`
    LastHeartbeatAt         time.Time             `json:"last_heartbeat_at"`
    LastEventAt             time.Time             `json:"last_event_at"`
    LastSuccessfulCatchUpAt time.Time             `json:"last_successful_catch_up_at"`
    Retries                 uint64                `json:"retries"`
    LastError               *HealthError          `json:"last_error,omitempty"`
    PollingFallback         PollingFallbackStatus `json:"polling_fallback"`
}

type HealthError

HealthError is a secret-free classification of the latest subscriber error. It deliberately excludes endpoint URLs, response bodies, and wrapped error text because those values can contain credentials or publisher details.

type HealthError struct {
    Operation  string    `json:"operation"`
    Kind       string    `json:"kind"`
    StatusCode int       `json:"status_code,omitempty"`
    Terminal   bool      `json:"terminal"`
    OccurredAt time.Time `json:"occurred_at"`
}

type PollingFallbackPolicy

PollingFallbackPolicy explicitly enables bounded conditional polling after repeated streaming failures. Polling remains disabled when this policy is nil.

type PollingFallbackPolicy struct {
    // AfterFailures is the number of consecutive stream open, read, or catch-up
    // failures required before fallback polling begins.
    AfterFailures int
    // Interval is the minimum time between fallback manifest polls.
    Interval time.Duration
}

type PollingFallbackStatus

PollingFallbackStatus is an immutable snapshot of the subscriber's bounded polling fallback. Counters are cumulative for the subscriber lifetime.

type PollingFallbackStatus struct {
    // Enabled reports whether construction configured a polling fallback.
    Enabled bool
    // Active reports that the stream failure threshold has been reached and
    // streaming has not yet recovered.
    Active bool
    // Entries counts transitions into fallback mode.
    Entries uint64
    // Polls counts conditional current-manifest requests.
    Polls uint64
    // Modified counts verified non-304 responses handled by fallback polling.
    Modified uint64
}

type StreamState

StreamState is the subscriber's current reactive transport state.

type StreamState string

const (
    // StreamStateIdle means Start has not established a lifecycle.
    StreamStateIdle StreamState = "idle"
    // StreamStateStarting means initial verification or stream setup is active.
    StreamStateStarting StreamState = "starting"
    // StreamStateStreaming means an SSE stream is established and caught up.
    StreamStateStreaming StreamState = "streaming"
    // StreamStateRetrying means the subscriber is recovering a failed stream.
    StreamStateRetrying StreamState = "retrying"
    // StreamStatePolling means explicit conditional fallback polling is active.
    StreamStatePolling StreamState = "polling"
    // StreamStateStopped means the one-shot lifecycle has ended.
    StreamStateStopped StreamState = "stopped"
)

type Subscriber

Subscriber owns one explicitly started remote catalog lifecycle.

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

func New

func New(config Config) (*Subscriber, error)

New validates config and constructs an idle subscriber. It starts no goroutine and performs no remote request.

func (*Subscriber) Catalog

func (s *Subscriber) Catalog() *catalogs.Catalog

Catalog returns the current immutable catalog. Before Start succeeds it is the verified embedded bootstrap; afterward it is the latest activated remote generation.

func (*Subscriber) Close

func (s *Subscriber) Close() error

Close cancels and joins the subscriber lifecycle within ShutdownTimeout. It is idempotent.

func (*Subscriber) Health

func (s *Subscriber) Health() Health

Health returns the current subscriber health without performing I/O.

func (*Subscriber) PollingFallbackStatus

func (s *Subscriber) PollingFallbackStatus() PollingFallbackStatus

PollingFallbackStatus returns the current bounded polling fallback state.

func (*Subscriber) Start

func (s *Subscriber) Start(ctx context.Context) error

Start performs a verified initial fetch and starts the caller-context-owned lifecycle. Normally it establishes the event stream and closes the fetch-to-subscribe gap with a mandatory current-state catch-up before returning. When an explicit PollingFallbackPolicy is configured, an initial stream failure starts bounded fallback/reconnect recovery instead. HTTP 401 and 403 responses are terminal for the active lifecycle and never retry or enter polling fallback.

Generated by gomarkdoc