plex

README

Plex the net.Conn Multiplexer

Plex is a library for managing net.Conn connection types with minimal effort this library utilizes the options pattern to allow for easy configuration and customization of the multiplexer. This library provides multiple types which are used as wrappers to the configured net.Conn types which allow for general streaming of data through channels of bytes as an alternative to the more go traditional io.Reader/io.Writer interfaces.

One important caveat is that the Plex library makes no assumptions of which net.Conn object is returned from either Reader or Writer. This means that messages must be routed by the consumer based on the data in the message rather than assuming that a response to a received message will be sent to the same connection which initiated the message round trip transmission.

The goal of this library is to abstract the complexity of managing multiple connections and their associated channels of data into a single API which consumers can request Reader and Writer types from. These types implement the io.Closer method which will re-queue the Reader or Writer to the correct internal buffer within the multiplexer for another consumer down the line.

Installation

go get -u go.atomizer.io/plex@latest

Usage

Plex uses the options pattern to allow for users to configure the multiplexer. For a full list of options see the options documentation.

import "go.atomizer.io/plex"

...

m, err := plex.New(
    ctx, 
    plex.WithConnections(
        mynetconn1,
        mynetconn2,
        mynetconnN,
    ),
)

// Access a plex.Reader
r, err := m.Reader(ctx, timeout)

// Access a plex.Writer
w, err := m.Writer(ctx, timeout)

Users are responsible for closing any plex.Readers/plex.Writers they acquire from the multiplexer. If they are not closed they will not be returned to the multiplexer pool for reuse.

For further options and documentation visit our documentation page using the docs badge at the top of the README.

Options

Availble configuration options include the following:

NOTE: For the full documentation see Options

WithConnections(connections ...net.Conn)

Adds a pre-configured list of net.Conn objects to the multiplexer. This option also sets the internal buffer size to the number of the Multiplexer when NOT used in conjunction with the WithMaxCapacity option.

NOTE: Buffer size of the multiplexer does NOT change for the lifetime of the multiplexer.

WithMaxCapacity(max int)

Sets a pre-configured maximum capacity for the multiplexer. This max capacity will not change for the lifetime of the multiplexer.

WithConnector(c Connect)

Allows for the configuration of a custom Connect function. This function has the following signature: func(ctx context.Context, conn net.Conn) error. The configuration of the Connector allows the Multiplexer to re-create failed connections that already exist or to create new connections when using the WithAutoScaling option.

WithAutoScaling(timeout time.Duration)

Configures an internal timeout within the multiplexer to automatically add new connections to the multiplexer in the event that there are not any available connections in the multiplexer. This option is only available when using the WithConnector option.

Documentation

Overview

plex is a library for managing net.Conn connection types with minimal effort this library utilizes the options pattern to allow for easy configuration and customization of the multiplexer. This library provides multiple types which are used as wrappers to the configured net.Conn types which allow for general streaming of data through channels of bytes as an alternative to the more go traditional io.Reader/io.Writer interfaces.

One important caveat is that the Plex library makes no assumptions of which net.Conn object is returned from either Reader or Writer. This means that messages must be routed by the consumer based on the data in the message rather than assuming that a response to a received message will be sent to the same connection which initiated the message round trip transmission.

The goal of this library is to abstract the complexity of managing multiple connections and their associated channels of data into a single API which consumers can request Reader and Writer types from. These types implement the io.Closer method which will re-queue the Reader or Writer to the correct internal buffer within the multiplexer for another consumer down the line.

Index

• type Conn
• type Connect
• type ErrConnection
• type Killer
• type Multiplexer
  • func New(ctx context.Context, opts ...Option) (*Multiplexer, error)
  • func (m *Multiplexer) Add(ctx context.Context, conns ...Conn) (left []Conn, err error)
  • func (m *Multiplexer) Addr() net.Addr
  • func (m *Multiplexer) Close() (err error)
  • func (m *Multiplexer) Reader(ctx context.Context, timeout *time.Duration) (Reader, error)
  • func (m *Multiplexer) Writer(ctx context.Context, timeout *time.Duration) (Writer, error)
• type Option
  • func WithAutoScaling(timeout time.Duration) Option
  • func WithConnections(connections ...net.Conn) Option
  • func WithConnector(c Connect) Option
  • func WithMaxCapacity(capacity int) Option
• type Reader
• type Writer

Types

type Conn

type Conn interface {
	io.ReadWriteCloser
	RemoteAddr() net.Addr
}

type Connect

type Connect func(context.Context, net.Addr) (net.Conn, error)

type ErrConnection

type ErrConnection struct {
	net.Addr
	// contains filtered or unexported fields
}

type Killer

type Killer interface {
	Kill() error
}

type Multiplexer

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

Multiplexer is the type which manages the multiplexing of net.Conn types through the use of channels and goroutines.

func New

func New(ctx context.Context, opts ...Option) (*Multiplexer, error)

New creates a net.Conn Multiplexer with the provided options.

CRITICAL: A Multiplexer must be closed by calling the Close method when it is no longer needed. Failure to do so may result in goroutines leaking.

NOTE: A single instance of the Multiplexer can ONLY be used to manage a set of connections for a SINGLE host.

NOTE: It is important to read the documentation for the options which you wish to use to ensure proper configuration of the multiplexer.

func (*Multiplexer) Add

func (m *Multiplexer) Add(
	ctx context.Context,
	conns ...Conn,
) (left []Conn, err error)

Add adds the provided net.Conn instances to the multiplexer.

NOTE: If the Multiplexer does not have the capacity to add all of the provided connections, then the connections that were unable to be added will be returned. It is possible the the connection was able to be added for a reader or writer, or both but will not indicate which.

TODO: Correct this in the future, we should be able to know if it was added to one of them or both.

func (*Multiplexer) Addr

func (m *Multiplexer) Addr() net.Addr

Addr returns the net.Addr of the connections in this instance of the multiplexer.

func (*Multiplexer) Close

func (m *Multiplexer) Close() (err error)

Close closes the multiplexer and all of the connections it manages.

func (*Multiplexer) Reader

func (m *Multiplexer) Reader(
	ctx context.Context,
	timeout *time.Duration,
) (Reader, error)

Reader returns a wrapped net.Conn object as a Reader which provides stream capabilities to the net.Conn as well as cuts down on the available methods on the net.Conn. This is done on purpose to ensure that the consumer is less likely to misuse the Reader when requesting one for the purpose of reading. nolint:dupl

func (*Multiplexer) Writer

func (m *Multiplexer) Writer(
	ctx context.Context,
	timeout *time.Duration,
) (Writer, error)

Writer returns a wrapped net.Conn object as a Writer which provides stream capabilities to the net.Conn as well as cuts down on the available methods on the net.Conn. This is done on purpose to ensure that the consumer is less likely to misuse the Writer when requesting one for the purpose of writing. nolint:dupl

type Option

type Option func(*Multiplexer) error

Option is the function literate that is used to configure the multiplexer during initialization.

func WithAutoScaling

func WithAutoScaling(timeout time.Duration) Option

WithAutoScaling sets a timeout on the multiplexer to automatically scale connections when a connection request exceeds the current the timeout when attempting to read from the internal buffer.

NOTE: This options is ONLY available when a connector is provided via the `WithConnector` option.

TODO: This needs to have the ability to set a timer for the new connection to be killed when the connection is no longer used.

func WithConnections

func WithConnections(connections ...net.Conn) Option

WithConnections provides an initial set of connections to be used by the multiplexer.

NOTE: See note on `WithMaxCapacity` regarding the default buffer size when this option is used in conjunction with `WithMaxCapacity`, or no connections are provided via the `WithConnections` option.

func WithConnector

func WithConnector(c Connect) Option

WithConnector provides a function that can be used to create new connections in the multiplexer.

NOTE: This method is only used when the current length of the multiplexer's connections is less than the multiplexer's capacity and a connection already in the multiplexer fails and must be recreated OR the multiplexer is configured for auto scaling.

func WithMaxCapacity

func WithMaxCapacity(capacity int) Option

WithMaxCapacity allocates the internal buffer for the multiplexer to a static capacity not based on the number of connections supplied at initialization but a set capacity.

NOTE: If no connections are provided via the `WithConnections` option, and the MaxCapacity is NOT set, the multiplexer will allocate a buffer of the default size which is 1.

type Reader

type Reader interface {
	io.ReadCloser
	Recv(ctx context.Context, buffer int) (<-chan byte, error)
	Killer
}

type Writer

type Writer interface {
	io.WriteCloser
	Send(ctx context.Context, buffer int) (chan<- byte, error)
	Killer
}