connected

Documentation

Overview

Package connected provides a pastry transport implementation where all nodes live within the same process and can communicate with each other directly.

Index

• func RandomJitter() *rapid.Generator[uint32]
• func RandomUniverse(logger *slog.Logger) *rapid.Generator[*Universe]
• func RandomUniverseN(logger *slog.Logger, minSize, maxSize int) *rapid.Generator[*Universe]
• type JitterSource
• type LocalNode
  • func (n *LocalNode) Format(state fmt.State, verb rune)
  • func (n *LocalNode) GetRemoteNodeLink(other *LocalNode) *nodeLink
  • func (n *LocalNode) LocalNode() pastry.Node
  • func (n *LocalNode) Logger() *slog.Logger
  • func (n *LocalNode) NodeID() pastry.Identifier
  • func (n *LocalNode) RandomNode() *rapid.Generator[pastry.Node]
  • func (n *LocalNode) ReadNode(p *pastry.MessagePayload) (pastry.Node, error)
  • func (n *LocalNode) SaturateRouter()
  • func (n *LocalNode) SendLocal(ctx context.Context, msg pastry.Message) error
• type NodeCallback
• type Service
  • func NewService[S any](u *Universe, initializer func(*LocalNode) *S) Service[S]
• type Universe
  • func NewUniverse(logger *slog.Logger, jitterSource JitterSource) *Universe
  • func (u *Universe) AddEvent(delta uint32, action func(ctx context.Context) error)
  • func (u *Universe) AddEventWithJitter(delta uint32, action func(ctx context.Context) error)
  • func (u *Universe) AddNode(nodeID pastry.Identifier, location uint16) *LocalNode
  • func (u *Universe) Err() error
  • func (u *Universe) Format(state fmt.State, verb rune)
  • func (u *Universe) GetOwner(dest pastry.Identifier) *LocalNode
  • func (u *Universe) GetOwningIdentifier(dest pastry.Identifier) pastry.Identifier
  • func (u *Universe) Now() uint32
  • func (u *Universe) RandomNode() *rapid.Generator[*LocalNode]
  • func (u *Universe) SaturateAllRouters()
  • func (u *Universe) Tick(ctx context.Context) bool
  • func (u *Universe) WaitToSettle(ctx context.Context) error
  • func (u *Universe) WhenNodeAdded(callback NodeCallback)

Functions

func RandomJitter

func RandomJitter() *rapid.Generator[uint32]

RandomJitter returns a property test generator that produces jitter that adds to the delay of each message delivery.

func RandomUniverse

func RandomUniverse(logger *slog.Logger) *rapid.Generator[*Universe]

RandomUniverse returns a property test generator that produces a connected universe.

func RandomUniverseN

func RandomUniverseN(logger *slog.Logger, minSize, maxSize int) *rapid.Generator[*Universe]

RandomUniverseN returns a property test generator that produces a connected universe containing a particular number of nodes.

Types

type JitterSource

type JitterSource func() uint32

type LocalNode

type LocalNode struct {
	MessageLibrary   *pastry.MessageLibrary
	Transceiver      *pastry.Transceiver
	Router           *pastry.Router
	SentMessages     *pastrytest.SentMessages
	ReceivedMessages *pastrytest.SentMessages
	// contains filtered or unexported fields
}

LocalNode represents one of the nodes in a connected universe. A LocalNode is the pastry.Transport implementation for the node. It contains the node's router and transceiver. It does not contain any other Pastry applications; you must instantiate those yourself depending on your needs. Use GetRemoteNodeLink to get the pastry.Node implementation that represents this node's view of another node on the network.

func (*LocalNode) Format

func (n *LocalNode) Format(state fmt.State, verb rune)

func (*LocalNode) GetRemoteNodeLink

func (n *LocalNode) GetRemoteNodeLink(other *LocalNode) *nodeLink

GetRemoteNodeLink returns the pastry.Node implementation that represents this node's view of another node on the network.

func (*LocalNode) LocalNode

func (n *LocalNode) LocalNode() pastry.Node

func (*LocalNode) Logger

func (n *LocalNode) Logger() *slog.Logger

func (*LocalNode) NodeID

func (n *LocalNode) NodeID() pastry.Identifier

func (*LocalNode) RandomNode

func (n *LocalNode) RandomNode() *rapid.Generator[pastry.Node]

RandomNode returns a property test generator that produces a link to a random node in the universe.

func (*LocalNode) ReadNode

func (n *LocalNode) ReadNode(p *pastry.MessagePayload) (pastry.Node, error)

func (*LocalNode) SaturateRouter

func (n *LocalNode) SaturateRouter()

SaturateRouter fills the node's Router with every node in the universe.

func (*LocalNode) SendLocal

func (n *LocalNode) SendLocal(ctx context.Context, msg pastry.Message) error

SendLocal serializes and sends a message to the node's transceiver.

type NodeCallback

type NodeCallback func(node *LocalNode)

type Service

type Service[S any] map[*LocalNode]*S

Service contains an instance of some Pastry application type for each of the nodes in a connected universe. The underlying type is a map, keyed by LocalNode pointers.

func NewService

func NewService[S any](u *Universe, initializer func(*LocalNode) *S) Service[S]

type Universe

type Universe struct {
	Nodes       []*LocalNode
	Identifiers []pastry.Identifier
	// contains filtered or unexported fields
}

Universe represents a “connected” Pastry network, where all nodes live within the same process and can communicate with each other directly.

func NewUniverse

func NewUniverse(logger *slog.Logger, jitterSource JitterSource) *Universe

NewUniverse creates a new empty connected universe. You will usually not have to call this function directly, and will instead use RandomUniverse or RandomUniverseN to create a random universe in a Rapid property test.

func (*Universe) AddEvent

func (u *Universe) AddEvent(delta uint32, action func(ctx context.Context) error)

AddEvent enqueues a new event that will be performed delta units of time into the future.

func (*Universe) AddEventWithJitter

func (u *Universe) AddEventWithJitter(delta uint32, action func(ctx context.Context) error)

AddEventWithJitter enqueues a new event that will be performed delta units of time, plus a random amount of jitter, into the future.

func (*Universe) AddNode

func (u *Universe) AddNode(nodeID pastry.Identifier, location uint16) *LocalNode

AddNode adds a node to a connected universe. Panics if there is already a node with the same identifier.

The node's router starts out empty. You should use SaturateRouter to force the node to know about every other node in the universe, or use the join protocol to have it learn about other nodes organically.

func (*Universe) Err

func (u *Universe) Err() error

Err returns whether any errors occurred while perform an event via [Tick].

func (*Universe) Format

func (u *Universe) Format(state fmt.State, verb rune)

func (*Universe) GetOwner

func (u *Universe) GetOwner(dest pastry.Identifier) *LocalNode

GetOwner returns which node in the universe owns a particular destination identifier.

func (*Universe) GetOwningIdentifier

func (u *Universe) GetOwningIdentifier(dest pastry.Identifier) pastry.Identifier

GetOwningIdentifier returns the identifier of the node in the universe that owns a particular destination identifier.

func (*Universe) Now

func (u *Universe) Now() uint32

Now returns the current simulated time in the connected universe.

func (*Universe) RandomNode

func (u *Universe) RandomNode() *rapid.Generator[*LocalNode]

RandomNode returns a property test generator that produces a random node in the universe.

func (*Universe) SaturateAllRouters

func (u *Universe) SaturateAllRouters()

SaturateAllRouters fills every node's Router with information about every node in the universe. Use this if you want to ensure all nodes know about each other as part of your test case, rather than using the join protocol to have nodes find each other organically.

func (*Universe) Tick

func (u *Universe) Tick(ctx context.Context) bool

Tick performs the next enqueued event, if any. It returns false if there were no events to perform, or if the event (or any previous event) returned an error.

func (*Universe) WaitToSettle

func (u *Universe) WaitToSettle(ctx context.Context) error

WaitToSettle waits until the universe has “settled”, ensuring that all pending events have been performed, as well as (transitively) any events enqueued while performing any of those events.

func (*Universe) WhenNodeAdded

func (u *Universe) WhenNodeAdded(callback NodeCallback)

WhenNodeAdded registers a callback that will be invoked every time a new node is added to a connected universe. (It will be immediately called for any nodes that are already in the universe.)