brokerapi

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 2, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package brokerapi defines the local realtime registry protocol and Go client.

Index

Constants

View Source 
const (
	// ProtocolVersion is the current newline-delimited JSON protocol version.
	ProtocolVersion = 1

	MethodPing         = "ping"
	MethodObserve      = "observe"
	MethodObserveBatch = "observe_batch"
	MethodList         = "list"
	MethodGet          = "get"
	MethodSummary      = "summary"
	MethodGC           = "gc"
	MethodSubscribe    = "subscribe"
)
View Source 
const SocketPathEnv = "AHT_SOCKET"

SocketPathEnv overrides the broker socket used by clients and integrations.

Variables

View Source 
var (
	// ErrUnavailable means no realtime broker accepted the local connection.
	ErrUnavailable = errors.New("aht broker unavailable")
	// ErrProtocol means the broker returned an invalid or incompatible frame.
	ErrProtocol = errors.New("aht broker protocol error")
)

Functions

func IsUnavailable 

func IsUnavailable(err error) bool

IsUnavailable reports whether err means no broker accepted the connection.

func SocketPath 

func SocketPath(storePath string) string

SocketPath returns the local socket for one registry snapshot path.

Types

type Client 

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

Client sends registry operations to one local broker.

func NewClient 

func NewClient(storePath string) *Client

NewClient returns a client for the broker associated with storePath.

func NewClientForSocket 

func NewClientForSocket(socketPath string) *Client

func (*Client) GC 

func (c *Client) GC(ctx context.Context, deleteAfter time.Duration) (registry.GCResult, error)

GC removes expired gone-session tombstones through the broker.

func (*Client) Get 

func (c *Client) Get(ctx context.Context, id string) (registry.Session, error)

Get returns one broker-owned session.

func (*Client) List 

func (c *Client) List(ctx context.Context, filter registry.Filter) ([]registry.Session, error)

List returns the broker's current filtered sessions.

func (*Client) Observe 

func (c *Client) Observe(ctx context.Context, observation registry.Observation) (registry.Session, error)

Observe records one observation in the broker.

func (*Client) ObserveBatch 

func (c *Client) ObserveBatch(
	ctx context.Context,
	observations []registry.Observation,
) ([]registry.Session, error)

ObserveBatch records a group of observations atomically in the broker.

func (*Client) Ping 

func (c *Client) Ping(ctx context.Context) error

Ping verifies that the broker is accepting requests.

func (*Client) SocketPath 

func (c *Client) SocketPath() string

SocketPath returns the exact endpoint used by the client.

func (*Client) Subscribe 

func (c *Client) Subscribe(ctx context.Context, filter registry.Filter) (*Subscription, error)

Subscribe returns the initial snapshot followed by strictly newer revisions.

func (*Client) Summary 

func (c *Client) Summary(ctx context.Context, filter registry.Filter) ([]registry.Summary, error)

Summary returns filtered multiplexer-session summaries.

type Error 

type Error struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

Error is a machine-readable broker failure.

type RemoteError 

type RemoteError struct {
	Code    string
	Message string
}

RemoteError is an operation error returned by the broker.

func (*RemoteError) Error 

func (e *RemoteError) Error() string

type Request 

type Request struct {
	Version      int                    `json:"version"`
	ID           string                 `json:"id,omitempty"`
	Method       string                 `json:"method"`
	Observation  *registry.Observation  `json:"observation,omitempty"`
	Observations []registry.Observation `json:"observations,omitempty"`
	Filter       registry.Filter        `json:"filter,omitzero"`
	SessionID    string                 `json:"session_id,omitempty"`
	DeleteAfter  time.Duration          `json:"delete_after,omitempty"`
}

Request is one broker RPC. Subscribe keeps the connection open after the initial response and streams Snapshot frames.

type Response 

type Response struct {
	Version   int                     `json:"version"`
	ID        string                  `json:"id,omitempty"`
	Type      string                  `json:"type"`
	Error     *Error                  `json:"error,omitempty"`
	Session   *registry.Session       `json:"session,omitempty"`
	Sessions  []registry.Session      `json:"sessions,omitempty"`
	Summaries []registry.Summary      `json:"summaries,omitempty"`
	Snapshot  *registry.StateSnapshot `json:"snapshot,omitempty"`
	GC        *registry.GCResult      `json:"gc,omitempty"`
	Now       time.Time               `json:"now,omitzero"`
}

Response is one broker result or subscription frame.

type Store 

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

Store routes operations through the realtime broker and falls back to the durable snapshot when the broker is offline. The fallback keeps one-shot CLI use functional; a running broker remains the authoritative hot path.

func NewStore 

func NewStore(storePath string) *Store

NewStore returns a broker-backed registry store for storePath.

func NewStoreForSocket 

func NewStoreForSocket(storePath string, socketPath string) *Store

func (*Store) Client 

func (s *Store) Client() *Client

Client returns the realtime client used by the store.

func (*Store) GC 

func (s *Store) GC(ctx context.Context, deleteAfter time.Duration) (registry.GCResult, error)

func (*Store) Get 

func (s *Store) Get(ctx context.Context, id string) (registry.Session, error)

func (*Store) List 

func (s *Store) List(ctx context.Context, filter registry.Filter) ([]registry.Session, error)

func (*Store) Observe 

func (s *Store) Observe(ctx context.Context, observation registry.Observation) (registry.Session, error)

func (*Store) ObserveBatch 

func (s *Store) ObserveBatch(
	ctx context.Context,
	observations []registry.Observation,
) ([]registry.Session, error)

func (*Store) SummaryByTmuxSession 

func (s *Store) SummaryByTmuxSession(
	ctx context.Context,
	filter registry.Filter,
) ([]registry.Summary, error)

func (*Store) SummaryByTmuxSessionWithOptions 

func (s *Store) SummaryByTmuxSessionWithOptions(
	ctx context.Context,
	options registry.SummaryOptions,
) ([]registry.Summary, error)

type Subscription 

type Subscription struct {
	Snapshots <-chan registry.StateSnapshot
	Errors    <-chan error
	// contains filtered or unexported fields
}

Subscription streams immutable effective-state snapshots until its context is canceled.

func (*Subscription) Close 

func (s *Subscription) Close()

Close cancels the subscription. It is safe to call more than once.

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.