unicast

Documentation

Index

• Constants
• func FlowGzipProtocolId(sporkId flow.Identifier) protocol.ID
• func FlowProtocolID(sporkId flow.Identifier) protocol.ID
• func IsFlowProtocolStream(s libp2pnet.Stream) bool
• func PingProtocolId(sporkId flow.Identifier) protocol.ID
• type GzipStream
  • func NewGzipCompressedUnicast(logger zerolog.Logger, sporkId flow.Identifier, ...) *GzipStream
  • func (g GzipStream) Handler(s libp2pnet.Stream)
  • func (g GzipStream) ProtocolId() protocol.ID
  • func (g GzipStream) UpgradeRawStream(s libp2pnet.Stream) (libp2pnet.Stream, error)
• type LibP2PStreamFactory
  • func (l *LibP2PStreamFactory) ClearBackoff(p peer.ID)
  • func (l *LibP2PStreamFactory) Connect(ctx context.Context, pid peer.AddrInfo) error
  • func (l *LibP2PStreamFactory) DialAddress(p peer.ID) []multiaddr.Multiaddr
  • func (l *LibP2PStreamFactory) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error)
  • func (l *LibP2PStreamFactory) SetStreamHandler(pid protocol.ID, handler network.StreamHandler)
• type Manager
  • func NewUnicastManager(logger zerolog.Logger, streamFactory StreamFactory, sporkId flow.Identifier) *Manager
  • func (m *Manager) CreateStream(ctx context.Context, peerID peer.ID, maxAttempts int) (libp2pnet.Stream, []multiaddr.Multiaddr, error)
  • func (m *Manager) Register(unicast ProtocolName) error
  • func (m *Manager) WithDefaultHandler(defaultHandler libp2pnet.StreamHandler)
• type PlainStream
  • func (p PlainStream) Handler(s libp2pnet.Stream)
  • func (p PlainStream) ProtocolId() protocol.ID
  • func (p PlainStream) UpgradeRawStream(s libp2pnet.Stream) (libp2pnet.Stream, error)
• type Protocol
• type ProtocolFactory
  • func ToProtocolFactory(name ProtocolName) (ProtocolFactory, error)
• type ProtocolName
  • func ToProtocolNames(names []string) []ProtocolName
• type StreamFactory
  • func NewLibP2PStreamFactory(h host.Host) StreamFactory

Constants

const (
	// FlowLibP2PProtocolCommonPrefix is the common prefix for all Libp2p protocol IDs used for Flow
	// ALL Flow libp2p protocols must start with this prefix.
	FlowLibP2PProtocolCommonPrefix = "/flow"

	// FlowLibP2POneToOneProtocolIDPrefix is a unique Libp2p protocol ID prefix for Flow (https://docs.libp2p.io/concepts/protocols/)
	// All nodes communicate with each other using this protocol id suffixed with the id of the root block
	FlowLibP2POneToOneProtocolIDPrefix = FlowLibP2PProtocolCommonPrefix + "/push/"

	// FlowLibP2PPingProtocolPrefix is the Flow Ping protocol prefix
	FlowLibP2PPingProtocolPrefix = FlowLibP2PProtocolCommonPrefix + "/ping/"

	// FlowLibP2PProtocolGzipCompressedOneToOne represents the protocol id for compressed streams under gzip compressor.
	FlowLibP2PProtocolGzipCompressedOneToOne = FlowLibP2POneToOneProtocolIDPrefix + "/gzip/"
)

Flow Libp2p protocols

const GzipCompressionUnicast = ProtocolName("gzip-compression")

const MaxConnectAttemptSleepDuration = 5

MaxConnectAttemptSleepDuration is the maximum number of milliseconds to wait between attempts for a 1-1 direct connection

Functions

func FlowGzipProtocolId

func FlowGzipProtocolId(sporkId flow.Identifier) protocol.ID

func FlowProtocolID

func FlowProtocolID(sporkId flow.Identifier) protocol.ID

func IsFlowProtocolStream

func IsFlowProtocolStream(s libp2pnet.Stream) bool

IsFlowProtocolStream returns true if the libp2p stream is for a Flow protocol

func PingProtocolId

func PingProtocolId(sporkId flow.Identifier) protocol.ID

Types

type GzipStream

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

GzipStream is a stream compression creates and returns a gzip-compressed stream out of input stream.

func NewGzipCompressedUnicast

func NewGzipCompressedUnicast(logger zerolog.Logger, sporkId flow.Identifier, defaultHandler libp2pnet.StreamHandler) *GzipStream

func (GzipStream) Handler

func (g GzipStream) Handler(s libp2pnet.Stream)

func (GzipStream) ProtocolId

func (g GzipStream) ProtocolId() protocol.ID

func (GzipStream) UpgradeRawStream

func (g GzipStream) UpgradeRawStream(s libp2pnet.Stream) (libp2pnet.Stream, error)

UpgradeRawStream wraps gzip compression and decompression around the plain libp2p stream.

type LibP2PStreamFactory

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

func (*LibP2PStreamFactory) ClearBackoff

func (l *LibP2PStreamFactory) ClearBackoff(p peer.ID)

func (*LibP2PStreamFactory) Connect

func (l *LibP2PStreamFactory) Connect(ctx context.Context, pid peer.AddrInfo) error

func (*LibP2PStreamFactory) DialAddress

func (l *LibP2PStreamFactory) DialAddress(p peer.ID) []multiaddr.Multiaddr

func (*LibP2PStreamFactory) NewStream

func (l *LibP2PStreamFactory) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error)

func (*LibP2PStreamFactory) SetStreamHandler

func (l *LibP2PStreamFactory) SetStreamHandler(pid protocol.ID, handler network.StreamHandler)

type Manager

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

Manager manages libp2p stream negotiation and creation, which is utilized for unicast dispatches.

func NewUnicastManager

func NewUnicastManager(logger zerolog.Logger, streamFactory StreamFactory, sporkId flow.Identifier) *Manager

func (*Manager) CreateStream

func (m *Manager) CreateStream(ctx context.Context, peerID peer.ID, maxAttempts int) (libp2pnet.Stream, []multiaddr.Multiaddr, error)

CreateStream tries establishing a libp2p stream to the remote peer id. It tries creating streams in the descending order of preference until it either creates a successful stream or runs out of options. Creating stream on each protocol is tried at most `maxAttempt` one, and then falls back to the less preferred one.

func (*Manager) Register

func (m *Manager) Register(unicast ProtocolName) error

Register registers given protocol name as preferred unicast. Each invocation of register prioritizes the current protocol over previously registered ones.

func (*Manager) WithDefaultHandler

func (m *Manager) WithDefaultHandler(defaultHandler libp2pnet.StreamHandler)

WithDefaultHandler sets the default stream handler for this unicast manager. The default handler is utilized as the core handler for other unicast protocols, e.g., compressions.

type PlainStream

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

PlainStream is a stream factory that reflects the same input stream without any modification.

func (PlainStream) Handler

func (p PlainStream) Handler(s libp2pnet.Stream)

func (PlainStream) ProtocolId

func (p PlainStream) ProtocolId() protocol.ID

func (PlainStream) UpgradeRawStream

func (p PlainStream) UpgradeRawStream(s libp2pnet.Stream) (libp2pnet.Stream, error)

UpgradeRawStream implements protocol interface and returns the input stream without any modification.

type Protocol

type Protocol interface {
	// UpgradeRawStream wraps a specific unicast protocol implementation around
	// the plain libp2p stream.
	UpgradeRawStream(s libp2pnet.Stream) (libp2pnet.Stream, error)
	// Handler is a libp2p stream handler that implements the logic of handling
	// an established incoming stream to this node.
	Handler(stream libp2pnet.Stream)
	// ProtocolId returns the libp2p protocol id associated to this protocol. In libp2p
	// streams running with the same protocol are identified with the same  protocol id.
	ProtocolId() protocol.ID
}

Protocol represents a unicast protocol.

type ProtocolFactory

type ProtocolFactory func(zerolog.Logger, flow.Identifier, libp2pnet.StreamHandler) Protocol

func ToProtocolFactory

func ToProtocolFactory(name ProtocolName) (ProtocolFactory, error)

type ProtocolName

type ProtocolName string

func ToProtocolNames

func ToProtocolNames(names []string) []ProtocolName

type StreamFactory

type StreamFactory interface {
	SetStreamHandler(protocol.ID, network.StreamHandler)
	DialAddress(peer.ID) []multiaddr.Multiaddr
	ClearBackoff(peer.ID)
	Connect(context.Context, peer.AddrInfo) error
	NewStream(context.Context, peer.ID, ...protocol.ID) (network.Stream, error)
}

StreamFactory is a wrapper around libp2p host.Host to provide abstraction and encapsulation for unicast stream manager so that it can create libp2p streams with finer granularity.

func NewLibP2PStreamFactory

func NewLibP2PStreamFactory(h host.Host) StreamFactory