node

Documentation

Overview

Package node assembles a running jetwayd from a configuration.

It exists so that there is exactly one wiring. jetwayd builds a node and serves it; the scenario suite builds a node and drives it. An integration test standing on a second, parallel copy of this assembly would be testing the copy -- and this repository has twice shipped tests that passed because they encoded the same assumption as the code they were checking.

Index

• Constants
• Variables
• type Node
  • func Build(ctx context.Context, cfg *config.Config, log *slog.Logger, opts Options) (*Node, error)
  • func (n *Node) Addr(listener string) string
  • func (n *Node) Close()
  • func (n *Node) Drain(ctx context.Context, hs *http.Server)
  • func (n *Node) Handler() ingress.Handler
  • func (n *Node) Holding() bool
  • func (n *Node) Listeners() []ingress.Ingress
  • func (n *Node) LivePeers() []string
  • func (n *Node) ReloadPeers(peers []config.Peer) (added int, err error)
  • func (n *Node) Serve(ctx context.Context, drainTimeout time.Duration) error
  • func (n *Node) Start(ctx context.Context) error
• type Options

Constants

const RelayOutboxDepth = 4096

RelayOutboxDepth is how many frames a relaying node queues per link before refusing a send: eight times a subscriber's, because a switch's links -- a trunk above all -- carry many subscribers' traffic at once.

Variables

var SpoolReadyAge = 30 * time.Second

SpoolReadyAge is how old the spool's oldest unflushed entry may be before the node reports not ready: the store is not keeping up, and a load balancer should send partners' new sessions to a node whose is.

Types

type Node

type Node struct {
	Config  *config.Config
	Log     *slog.Logger
	Store   store.Store
	Gateway *gateway.Gateway
	Bus     *gateway.Bus
	Queues  *queue.Manager
	Sweeper *queue.Sweeper
	Router  *egress.Router
	Spool   *spool.Spool
	Fleet   *demo.RunningFleet
	API     *api.Server
	// contains filtered or unexported fields
}

Node is an assembled gateway: store, links, queues, sweeper and console.

Everything a caller might want to drive or assert on is exported, because the scenario suite is a legitimate consumer rather than a special case that needs back doors cut for it.

func Build

func Build(ctx context.Context, cfg *config.Config, log *slog.Logger, opts Options) (*Node, error)

Build assembles a node without starting anything that accepts work.

Listeners bind here, so a port conflict or an unreadable certificate fails at build time rather than later in a goroutine where it reads as silence.

func (*Node) Addr

func (n *Node) Addr(listener string) string

Addr returns the address of the named TCP listener, which is how a test that asked for port 0 finds out what it got.

func (*Node) Close

func (n *Node) Close()

func (*Node) Drain

func (n *Node) Drain(ctx context.Context, hs *http.Server)

Drain stops taking new work, lets what is in flight finish, then stops serving. Cutting links first would lose messages mid-pipeline.

func (*Node) Handler

func (n *Node) Handler() ingress.Handler

makeHandler builds the function every ingress calls. Handler is the ingress callback: it is what turns bytes off a link into a stored, parsed, applied message.

func (*Node) Holding

func (n *Node) Holding() bool

Holding reports whether this process currently holds the system: without a lease, always; with one, only while it is the writer.

func (*Node) Listeners

func (n *Node) Listeners() []ingress.Ingress

Listeners returns the bound ingress listeners.

func (*Node) LivePeers

func (n *Node) LivePeers() []string

LivePeers names the partners currently holding a session.

func (*Node) ReloadPeers

func (n *Node) ReloadPeers(peers []config.Peer) (added int, err error)

ReloadPeers adds the peers in the new list that this node does not have yet, without a restart: a partner onboarded while the links stay up. Peers already configured are left as they are; removing one is a restart, because a link that is open is a promise.

func (*Node) Serve

func (n *Node) Serve(ctx context.Context, drainTimeout time.Duration) error

Serve runs the console until the context is cancelled.

func (*Node) Start

func (n *Node) Start(ctx context.Context) error

Start brings up everything that accepts or generates work, and returns once the links are running. It does not serve the console; Serve does that.

type Options

type Options struct {
	// Store, when set, is used instead of the one the config names: a
	// harness sharing one store between nodes, which is how a lease is
	// tested.
	Store store.Store

	// LocatorSecret seeds record locator allocation. Required.
	LocatorSecret []byte
	// SkipConsole omits the HTTP server. A load run wants the pipeline, not
	// the console.
	SkipConsole bool
	// ExtendAPI is passed through to the console's mux, so an embedder can
	// serve its own pages from the node's one listener.
	ExtendAPI func(mux *http.ServeMux)
}

Options are the knobs the scenario suite needs and jetwayd does not.