broker

package
v1.3.0 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package broker provides a lightweight client and wire protocol for communicating with the local AHT realtime broker over a Unix domain socket.

Use NewClientForSocket or NewClient to connect directly to the broker daemon. Unlike github.com/zigai/aht/pkg/client, operations against a broker Client never fall back to durable disk files or take filesystem locks; if the broker is stopped, operations fail immediately with ErrUnavailable. Context cancellation interrupts both request and subscription-handshake I/O. Responses are newline-delimited JSON frames capped at MaxResponseBytes; oversized frames fail with ErrProtocol. Close subscriptions to cancel and join their reader goroutines.

Index

Examples

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 (

	// MaxResponseBytes bounds each newline-delimited broker response, including its newline.
	MaxResponseBytes = 64 << 20
)
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 DefaultSocketPath 

func DefaultSocketPath() string

DefaultSocketPath returns the default endpoint for the current user's AHT broker.

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
Example 
package main

import (
	"fmt"

	"github.com/zigai/aht/pkg/broker"
)

func main() {
	client := broker.NewClientForSocket("/path/to/aht.sock")
	fmt.Printf("Socket: %s\n", client.SocketPath())
}

Output:
Socket: /path/to/aht.sock

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.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/zigai/aht/pkg/broker"
	"github.com/zigai/aht/pkg/registry"
)

func main() {
	client := broker.NewClientForSocket("/path/to/offline.sock")
	_, err := client.List(context.Background(), registry.Filter{})
	if broker.IsUnavailable(err) {
		fmt.Println("broker is offline")
	}
}

Output:
broker is offline

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.

func (*Client) SummaryByTmuxSession 

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

SummaryByTmuxSession implements registry.Store.

func (*Client) SummaryByTmuxSessionWithOptions 

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

SummaryByTmuxSessionWithOptions implements registry.Store.

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 independently owned effective-state snapshots until its context is canceled. Callers may modify a received snapshot without affecting the broker or subsequent snapshots.

func NewSubscription 

func NewSubscription(snapshots <-chan registry.StateSnapshot, errors <-chan error, cancel context.CancelFunc) *Subscription

NewSubscription returns a Subscription wrapping the given channels and optional cancel function.

func (*Subscription) Close 

func (s *Subscription) Close()

Close cancels the subscription and waits for its broker reader to finish. For subscriptions created with NewSubscription, the supplied cancel function owns any external worker cleanup. It is safe to call Close 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.