swarm

Documentation

Index

• Variables
• func NewPeerPool[P PeerConn](concurrentConns uint64, fnGetPeers func(ctx context.Context) (<-chan P, error)) *peerPool[P]
• func NewPeerStore(db *state.DBTree) *peerStore
• type BaseProtocol
  • func (p BaseProtocol[T, P]) PeerInfosToPeerConns(ctx context.Context, peerInfos []PeerDevice) []P
  • func (t BaseProtocol[T, P]) TryEndpoints(ctx context.Context, parent *process.Process, peer PeerDevice, ...)
  • func (p BaseProtocol[T, P]) TryPeerDevices(ctx context.Context, parent *process.Process, peerDevices []PeerDevice, ...) (chDone <-chan struct{})
• type PeerConn
• type PeerDevice
• type PeerDialInfo
  • func (pdi PeerDialInfo) ID() process.PoolUniqueID
  • func (pdi PeerDialInfo) MapKey() (state.Keypath, error)
  • func (pdi PeerDialInfo) MarshalText() ([]byte, error)
  • func (pdi *PeerDialInfo) ScanMapKey(keypath state.Keypath) error
  • func (pdi PeerDialInfo) String() string
  • func (pdi *PeerDialInfo) UnmarshalText(bs []byte) error
• type PeerEndpoint
• type PeerPool
• type PeerStore
• type Transport

Variables

var (
	ErrProtocol    = errors.New("protocol error")
	ErrPeerIsSelf  = errors.New("peer is self")
	ErrUnreachable = errors.New("peer unreachable")
	ErrNotReady    = errors.New("not ready")
)

Functions

func NewPeerPool

func NewPeerPool[P PeerConn](concurrentConns uint64, fnGetPeers func(ctx context.Context) (<-chan P, error)) *peerPool[P]

func NewPeerStore

func NewPeerStore(db *state.DBTree) *peerStore

Types

type BaseProtocol

type BaseProtocol[T Transport, P PeerConn] struct {
	process.Process
	log.Logger
	Transports map[string]T
}

func (BaseProtocol[T, P]) PeerInfosToPeerConns

func (p BaseProtocol[T, P]) PeerInfosToPeerConns(ctx context.Context, peerInfos []PeerDevice) []P

func (BaseProtocol[T, P]) TryEndpoints

func (t BaseProtocol[T, P]) TryEndpoints(
	ctx context.Context,
	parent *process.Process,
	peer PeerDevice,
	fn func(ctx context.Context, peerConn P) error,
)

TryEndpoints accepts a list of peer endpoints for a single peer, attempts to establish connections to each of the endpoints concurrently, and runs the provided function on each one. As long as the function returns an error, it will continue attempting (while respecting the backoff for that endpoint). As soon as the function succeeds once, for a single endpoint, all connections are closed and TryEndpoints terminates. The returned channel closes when termination occurs.

func (BaseProtocol[T, P]) TryPeerDevices

func (p BaseProtocol[T, P]) TryPeerDevices(
	ctx context.Context,
	parent *process.Process,
	peerDevices []PeerDevice,
	fn func(ctx context.Context, peer P) error,
) (chDone <-chan struct{})

type PeerConn

type PeerConn interface {
	PeerEndpoint
	Transport() Transport
	EnsureConnected(ctx context.Context) error
	Close() error
}

type PeerDevice

type PeerDevice interface {
	DeviceUniqueID() string
	SetDeviceUniqueID(id string)
	Addresses() types.Set[types.Address]
	PublicKeys(addr types.Address) (*crypto.SigningPublicKey, *crypto.AsymEncPubkey)
	StateURIs() types.Set[string]
	AddStateURI(stateURI string)
	RemoveStateURI(stateURI string)
	LastContact() time.Time
	LastFailure() time.Time
	Failures() uint64
	Ready() bool
	RemainingBackoff() time.Duration

	Endpoints() map[PeerDialInfo]PeerEndpoint
	Endpoint(dialInfo PeerDialInfo) (PeerEndpoint, bool)
}

type PeerDialInfo

type PeerDialInfo struct {
	TransportName string
	DialAddr      string
}

func (PeerDialInfo) ID

func (pdi PeerDialInfo) ID() process.PoolUniqueID

func (PeerDialInfo) MapKey

func (pdi PeerDialInfo) MapKey() (state.Keypath, error)

func (PeerDialInfo) MarshalText

func (pdi PeerDialInfo) MarshalText() ([]byte, error)

func (*PeerDialInfo) ScanMapKey

func (pdi *PeerDialInfo) ScanMapKey(keypath state.Keypath) error

func (PeerDialInfo) String

func (pdi PeerDialInfo) String() string

func (*PeerDialInfo) UnmarshalText

func (pdi *PeerDialInfo) UnmarshalText(bs []byte) error

type PeerEndpoint

type PeerEndpoint interface {
	PeerDevice
	DialInfo() PeerDialInfo
	Dialable() bool
	UpdateConnStats(success bool)
	LastContact() time.Time
	LastFailure() time.Time
	Failures() uint64
	Ready() bool
	RemainingBackoff() time.Duration
}

type PeerPool

type PeerPool[P PeerConn] interface {
	process.Interface
	GetPeer(ctx context.Context) (_ P, err error)
	ReturnPeer(peer P, strike bool)
}

type PeerStore

type PeerStore interface {
	process.Interface

	AddDialInfo(dialInfo PeerDialInfo, deviceUniqueID string) PeerEndpoint
	AddVerifiedCredentials(dialInfo PeerDialInfo, deviceUniqueID string, address types.Address, sigpubkey *crypto.SigningPublicKey, encpubkey *crypto.AsymEncPubkey) PeerEndpoint
	RemovePeers(deviceUniqueIDs []string) error
	UnverifiedPeers() []PeerDialInfo
	VerifiedPeers() []PeerDevice
	Peers() []PeerDevice
	AllDialInfos() map[PeerDialInfo]struct{}
	PeerWithDeviceUniqueID(deviceUniqueID string) (PeerDevice, bool)
	PeerEndpoint(dialInfo PeerDialInfo) PeerEndpoint
	PeersWithAddress(address types.Address) []PeerDevice
	PeersFromTransport(transportName string) []PeerEndpoint
	PeersServingStateURI(stateURI string) []PeerDevice
	IsKnownPeer(dialInfo PeerDialInfo) bool
	OnNewUnverifiedPeer(fn func(dialInfo PeerDialInfo))
	OnNewVerifiedPeer(fn func(peer PeerDevice))

	DebugPrint()
}

type Transport

type Transport interface {
	process.Interface
	NewPeerConn(ctx context.Context, dialAddr string) (PeerConn, error)
}