aw

README

🌪 airwave

A flexible P2P networking library for upgradable distributed systems. The core mission of airwave is to provide a simple P2P interface that can support a wide variety of different algorithms, with a focus on backwards compatible. The P2P interface supports:

• Peer discovery
• Handshake
• Casting (send to one)
• Multicasting (send to many)
• Broadcasting (send to everyone)

Handshake

Airwave uses a 3 way sync handshake method to authorize peers in the network. The process is as follows:

The client sends a signed rsa public key on connect. The server validates the signature, generates a random challenge, and sends the signed random challenge encrypted with the client's public key; and the server's public key. The client validates the server's signature decrypts the challenge encrypts it with the server's publickey, signs it and sends it back.

Built with ❤ by Ren.

Documentation

Index

• Variables
• type IncomingMessage
• type ListenerOptions
  • func DefaultListenerOptions() ListenerOptions
• type Options
  • func DefaultOptions() Options
• type Peer
  • func New(opts Options, privKey *id.PrivKey, peerTable dht.Table, ...) *Peer
  • func (peer *Peer) Gossip(ctx context.Context, contentID []byte, subnet *id.Hash) error
  • func (peer *Peer) GossipNonBlocking(contentID []byte, subnet *id.Hash) error
  • func (peer *Peer) ID() id.Signatory
  • func (peer *Peer) Link(remote id.Signatory) error
  • func (peer *Peer) Listen(ctx context.Context, address string) (uint16, error)
  • func (peer *Peer) Run(ctx context.Context) error
  • func (peer *Peer) Send(ctx context.Context, data []byte, remote id.Signatory) error
  • func (peer *Peer) SendNonBlocking(data []byte, remote id.Signatory) error
  • func (peer *Peer) Sync(ctx context.Context, contentID []byte, hint *id.Signatory) ([]byte, error)
  • func (peer *Peer) SyncNonBlocking(ctx context.Context, contentID []byte, hint *id.Signatory) ([]byte, error)
  • func (peer *Peer) Unlink(remote id.Signatory) error
• type RateLimiterOptions
  • func DefaultRateLimiterOptions() RateLimiterOptions

Variables

var (
	DefaultSubnet = id.Hash{
		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
		0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
	}

	DefaultMaxLinkedPeers               uint          = 100
	DefaultMaxEphemeralConnections      uint          = 20
	DefaultMaxPendingSyncs              uint          = 100
	DefaultMaxActiveSyncsForSameContent uint          = 10
	DefaultMaxGossipSubnets             uint          = 100
	DefaultMaxMessageSize               uint          = 4 * 1024 * 1024
	DefaultOutgoingBufferSize           uint          = 100
	DefaultIncomingBufferSize           uint          = 100
	DefaultEventLoopBufferSize          uint          = 100
	DefaultOutgoingBufferTimeout        time.Duration = time.Second
	DefaultWriteTimeout                 time.Duration = time.Second
	DefaultDialRetryInterval            time.Duration = time.Second
	DefaultEphemeralConnectionTTL       time.Duration = 10 * time.Second
	DefaultMinimumConnectionExpiryAge   time.Duration = time.Minute
	DefaultGossipAlpha                  int           = 5
	DefaultGossipTimeout                time.Duration = 5 * time.Second
	DefaultPingAlpha                    int           = 5
	DefaultPongAlpha                    int           = 5
	DefaultPeerDiscoveryInterval        time.Duration = 30 * time.Second
	DefaultPeerExpiryTimeout            time.Duration = time.Minute
)

var (
	ErrTooManyLinkedPeers          = errors.New("too many linked peers")
	ErrTooManyEphemeralConnections = errors.New("too many ephemeral connections")
	ErrMessageBufferFull           = errors.New("outgoing message buffer is full")
	ErrEventLoopFull               = errors.New("event loop buffer is full")
	ErrTooManyPendingSyncs         = errors.New("too many pending sync requests")
	ErrTooManySyncsForSameContent  = errors.New("too many simultaneous syncs for the same content ID")
)

var (
	DefaultRateLimiterCapacity        int                = 10
	DefaultListenerRateLimiterOptions RateLimiterOptions = RateLimiterOptions{Rate: 10, Burst: 20}
	DefaultRate                       rate.Limit         = 1024 * 1024
	DefaultBurst                      int                = 4 * 1024 * 1024
)

Types

type IncomingMessage

type IncomingMessage struct {
	From id.Signatory
	Data []byte
}

type ListenerOptions

type ListenerOptions struct {
	RateLimiterCapacity int
	RateLimiterOptions
}

func DefaultListenerOptions

func DefaultListenerOptions() ListenerOptions

type Options

type Options struct {
	Logger *zap.Logger

	MaxLinkedPeers               uint
	MaxEphemeralConnections      uint
	MaxPendingSyncs              uint
	MaxActiveSyncsForSameContent uint
	MaxGossipSubnets             uint
	MaxMessageSize               uint
	OutgoingBufferSize           uint
	IncomingBufferSize           uint
	EventLoopBufferSize          uint
	OutgoingBufferTimeout        time.Duration
	WriteTimeout                 time.Duration
	DialRetryInterval            time.Duration
	EphemeralConnectionTTL       time.Duration
	MinimumConnectionExpiryAge   time.Duration

	GossipAlpha   int
	GossipTimeout time.Duration

	PingAlpha             int
	PongAlpha             int
	PeerDiscoveryInterval time.Duration
	PeerExpiryTimeout     time.Duration

	ListenerOptions              ListenerOptions
	ConnectionRateLimiterOptions RateLimiterOptions
}

func DefaultOptions

func DefaultOptions() Options

type Peer

type Peer struct {
	Opts Options

	Port uint16

	PeerTable        dht.Table
	ContentResolver  dht.ContentResolver
	IncomingMessages chan IncomingMessage
	// contains filtered or unexported fields
}

func New

func New(opts Options, privKey *id.PrivKey, peerTable dht.Table, contentResolver dht.ContentResolver) *Peer

func (*Peer) Gossip

func (peer *Peer) Gossip(ctx context.Context, contentID []byte, subnet *id.Hash) error

func (*Peer) GossipNonBlocking

func (peer *Peer) GossipNonBlocking(contentID []byte, subnet *id.Hash) error

func (*Peer) ID

func (peer *Peer) ID() id.Signatory

func (*Peer) Link

func (peer *Peer) Link(remote id.Signatory) error

func (*Peer) Listen

func (peer *Peer) Listen(ctx context.Context, address string) (uint16, error)

func (*Peer) Run

func (peer *Peer) Run(ctx context.Context) error

func (*Peer) Send

func (peer *Peer) Send(ctx context.Context, data []byte, remote id.Signatory) error

func (*Peer) SendNonBlocking

func (peer *Peer) SendNonBlocking(data []byte, remote id.Signatory) error

func (*Peer) Sync

func (peer *Peer) Sync(ctx context.Context, contentID []byte, hint *id.Signatory) ([]byte, error)

func (*Peer) SyncNonBlocking

func (peer *Peer) SyncNonBlocking(ctx context.Context, contentID []byte, hint *id.Signatory) ([]byte, error)

func (*Peer) Unlink

func (peer *Peer) Unlink(remote id.Signatory) error

type RateLimiterOptions

type RateLimiterOptions struct {
	Rate  rate.Limit
	Burst int
}

func DefaultRateLimiterOptions

func DefaultRateLimiterOptions() RateLimiterOptions