basemtp

Documentation

Index

• Constants
• type Channel
• type Client
  • func NewClient(addr string, mode ClientMode, logger logging.Logger, opts Options) Client
  • func NewClientDialer(addr string, mode ClientMode, dialer *net.Dialer, logger logging.Logger, ...) Client
• type ClientMode
• type Conn
  • func Connect(ctx async.Context, addr string, logger logging.Logger, opts Options) (Conn, status.Status)
  • func ConnectDialer(ctx async.Context, addr string, dialer *net.Dialer, logger logging.Logger, ...) (Conn, status.Status)
• type ConnContext
• type Context
  • func ClosedContext() Context
• type HandleFunc
  • func (f HandleFunc) HandleChannel(ctx Context, ch Channel) status.Status
• type Handler
• type Options
  • func Default() Options
  • func (o Options) Merge(o1 Options) Options
• type Server
  • func NewServer(address string, handler Handler, logger logging.Logger, opts Options) Server
• type TestConnContext
• type TestContext
  • func TestNewContext() TestContext
  • func TestNextContext(super async.Context) TestContext

Constants

const (
	ProtocolLine = "baseMTP/1\n"
)

Types

type Channel

type Channel interface {
	// Conn returns a channel connection.
	Conn() Conn

	// Context returns a channel context.
	Context() Context

	// Send sends a message to the channel.
	Send(ctx async.Context, data []byte) status.Status

	// SendAndClose sends a close message with a payload.
	SendAndClose(ctx async.Context, data []byte) status.Status

	// Receive receives and returns a message, or an end status.
	//
	// The message is valid until the next call to Receive.
	// The method blocks until a message is received, or the channel is closed.
	Receive(ctx async.Context) ([]byte, status.Status)

	// ReceiveAsync receives and returns a message, or false/end.
	//
	// The message is valid until the next call to Receive.
	// The method does not block if no messages, and returns false instead.
	ReceiveAsync(ctx async.Context) ([]byte, bool, status.Status)

	// ReceiveWait returns a channel that is notified on a new message, or a channel close.
	ReceiveWait() <-chan struct{}

	// Free closes the channel and releases its resources.
	Free()
}

type Client

type Client interface {
	// Address returns the server address.
	Address() string

	// Options returns the client options.
	Options() Options

	// Closed indicates that the client is closed.
	Closed() async.Flag

	// Connected indicates that the client is connected to the server.
	Connected() async.Flag

	// Disconnected indicates that the client is disconnected from the server.
	Disconnected() async.Flag

	// Close closes the client.
	Close() status.Status

	// Conn returns an existing connection, or opens a new one.
	Conn(ctx async.Context) (Conn, status.Status)

	// Channel returns a new channel.
	Channel(ctx async.Context) (Channel, status.Status)
}

Client is a SpecMPX client which manages outgoing connections.

func NewClient

func NewClient(addr string, mode ClientMode, logger logging.Logger, opts Options) Client

NewClient returns a new client.

func NewClientDialer

func NewClientDialer(addr string, mode ClientMode, dialer *net.Dialer, logger logging.Logger,
	opts Options) Client

NewClientDialer returns a new client with the given dialer.

type ClientMode

type ClientMode int

ClientMode specifies how the client connects to the server.

const (
	// ClientMode_OnDemand connects to the server on demand, does not reconnect on errors.
	ClientMode_OnDemand ClientMode = iota

	// ClientMode_AutoConnect automatically connects and reconnects to the server.
	// The client reconnects with exponential backoff on errors.
	ClientMode_AutoConnect
)

type Conn

type Conn interface {
	// Context returns a connection context.
	Context() ConnContext

	// Close closes the connection and frees its internal resources.
	Close() status.Status

	// Closed returns a flag that is set when the connection is closed.
	Closed() async.Flag

	// OnClosed adds a disconnect listener, and returns an unsubscribe function,
	// or false if the connection is already closed.
	OnClosed(fn func()) (unsub func(), _ bool)

	// Channel opens a new channel.
	Channel(ctx async.Context) (Channel, status.Status)

	// Free closes and frees the connection, allows to wrap the connection into ref.R[Conn].
	Free()
}

func Connect

func Connect(ctx async.Context, addr string, logger logging.Logger, opts Options) (
	Conn, status.Status)

Connect dials an address and returns a connection.

func ConnectDialer

func ConnectDialer(ctx async.Context, addr string, dialer *net.Dialer, logger logging.Logger,
	opts Options) (Conn, status.Status)

ConnectDialer dials an address and returns a connection.

type ConnContext

type ConnContext interface {
	async.Context

	// Disconnected returns a connection disconnected flag.
	Disconnected() async.Flag

	// OnDisconnected adds a disconnect listener, and returns an unsubscribe function,
	// or false if the connection is already closed.
	OnDisconnected(fn func()) (unsub func(), _ bool)
}

ConnContext is a connection context.

type Context

type Context interface {
	async.Context

	// Conn returns a connection context.
	Conn() ConnContext
}

Context is a channel context.

func ClosedContext

func ClosedContext() Context

ClosedContext returns a closed context.

type HandleFunc

type HandleFunc func(ctx Context, ch Channel) status.Status

HandleFunc is a type adapter to allow use of ordinary functions as channel handlers.

func (HandleFunc) HandleChannel

func (f HandleFunc) HandleChannel(ctx Context, ch Channel) status.Status

HandleChannel handles an incoming channel.

type Handler

type Handler interface {
	// HandleChannel handles an incoming channel.
	HandleChannel(ctx Context, ch Channel) status.Status
}

Handler is a server channel handler.

type Options

type Options struct {
	// ClientMaxConns is a maximum number of client connections, zero means one connection.
	ClientMaxConns int `json:"client_max_conns"`

	// ClientConnChannels is a target number of channels per connection, zero means no limit.
	ClientConnChannels int `json:"client_conn_channels"`

	// ClientDialTimeout is a client dial timeout.
	ClientDialTimeout time.Duration `json:"client_dial_timeout"`

	// Compression enables compression.
	Compression bool `json:"compress"`

	// ChannelWindowSize is an initial channel window size.
	ChannelWindowSize units.Bytes `json:"channel_window_size"`

	// ReadBufferSize is a connection read buffer size.
	ReadBufferSize units.Bytes `json:"read_buffer_size"`

	// WriteBufferSize is a connection write buffer size.
	WriteBufferSize units.Bytes `json:"write_buffer_size"`

	// WriteQueueSize is a max connection write queue size (soft limit).
	WriteQueueSize units.Bytes `json:"write_queue_size"`
}

func Default

func Default() Options

Default returns the default options.

func (Options) Merge

func (o Options) Merge(o1 Options) Options

Merge merges non-zero values from another options and returns new options.

type Server

type Server interface {
	async.Service

	// Address returns the address the server is listening to.
	Address() string

	// Listening indicates that the server is listening.
	Listening() async.Flag

	// Options returns the server options.
	Options() Options
}

Server is a SpecMPX server.

func NewServer

func NewServer(address string, handler Handler, logger logging.Logger, opts Options) Server

NewServer creates a new server with a connection handler.

type TestConnContext

type TestConnContext interface {
	ConnContext

	// Disconnect sets the disconnected flag and calls the disconnected listeners.
	Disconnect()

	// OnDisconnectedNum returns the number of disconnect listeners.
	OnDisconnectedNum() int
}

type TestContext

type TestContext interface {
	Context

	// TestConn returns the underlying test connection context.
	TestConn() TestConnContext
}

func TestNewContext

func TestNewContext() TestContext

TestNewContext returns a test context with a test connection.

func TestNextContext

func TestNextContext(super async.Context) TestContext

TestNextContext returns a test context with a test connection.