net

Documentation

Index

• Constants
• Variables
• func Pipe() (EndPoint, EndPoint)
• type Closer
• type Consumer
• type EndPoint
  • func ConnEndPoint(conn gonet.Conn) EndPoint
  • func DialEndPoint(addr string) (EndPoint, error)
  • func EndPointFinalizer(stream Stream, finalizer func(EndPoint)) EndPoint
  • func NewEndPoint(stream Stream) EndPoint
• type Filter
• type Handler
  • func NewHandler(f Filter, ch chan<- *Message, cl Closer) *Handler
• type Header
  • func NewHeader(typ uint8, service uint32, object uint32, action uint32, id uint32) Header
  • func (h *Header) Read(r io.Reader) (err error)
  • func (h Header) String() string
  • func (h *Header) Write(w io.Writer) (err error)
• type KeyNetContext
• type Listener
  • func Listen(addr string) (Listener, error)
• type Message
  • func NewMessage(header Header, payload []byte) Message
  • func (m *Message) Read(r io.Reader) error
  • func (m *Message) Write(w io.Writer) error
• type Stream
  • func ConnStream(conn gonet.Conn) Stream
  • func PipeStream(r, w *os.File) Stream

Constants

const (
	Unknown uint8 = iota
	Call
	Reply
	Error
	Post
	Event
	Capability
	Cancel
	Cancelled
)

Message types:

const HeaderSize = 28

HeaderSize is the size of a message header. It is the minimum size of a message.

const Magic = uint32(0x42dead42)

Magic is a constant to discriminate between message and garbage.

const MaxPayloadSize = uint32(10 * 1024 * 1024)

MaxPayloadSize limits the payload size of a message (10MB)

const Version = 0

Version is the supported version of the protocol.

Variables

var ErrConsumerBlocked = errors.New("message dropped: consumer blocked")

ErrConsumerBlocked is returned when the message cannot be sent to the consumer queue (the queue is full).

var ErrNoHandler = errors.New("message dropped: no handler registered")

ErrNoHandler is returned when there is no handler registered.

var ErrNoMatch = errors.New("message dropped: no handler match")

ErrNoMatch is returned when the message did not match any handler

Functions

func Pipe

func Pipe() (EndPoint, EndPoint)

Pipe returns a set of EndPoint connected to each other.

Types

type Closer

type Closer func(err error)

Closer informs the handler about a disconnection

type Consumer

type Consumer func(msg *Message) error

Consumer process a message which has been selected by a filter.

type EndPoint

type EndPoint interface {
	fmt.Stringer

	// Send pushes the message into the network.
	Send(m Message) error

	// ReceiveAny returns a chanel to receive a single message.
	ReceiveAny() (chan *Message, error)

	// MakeHandler registers the associated Filter and queue to
	// the incomming traffic from the EndPoint. Writing to the
	// queue must not block otherwise messages are discarded. If
	// an error occurs, closer is called, the consumer is closed.
	// closer can be nil. Do not attempt to add another handler
	// from within a Filter. Filter must not block.
	MakeHandler(f Filter, queue chan<- *Message, cl Closer) int

	// AddHandler creates a queue of 10 messages and spawn a
	// goroutine to forward those events into the consumer, then
	// calls MakeHandler. Do not attempt to add another handler
	// from within a Filter. Filter must not block.
	AddHandler(f Filter, c Consumer, cl Closer) int

	// RemoveHandler removes the associated Filter and Consumer.
	// RemoveHandler must not be called from within the Filter: use
	// the Filter returned value keep for this purpose.
	RemoveHandler(id int) error

	// Close close the underlying connection
	Close() error
}

EndPoint reprensents a network socket capable of sending and receiving messages.

func ConnEndPoint

func ConnEndPoint(conn gonet.Conn) EndPoint

ConnEndPoint returns an EndPoint using a go connection.

func DialEndPoint

func DialEndPoint(addr string) (EndPoint, error)

DialEndPoint construct an endpoint by contacting a given address.

func EndPointFinalizer

func EndPointFinalizer(stream Stream, finalizer func(EndPoint)) EndPoint

EndPointFinalizer creates a new EndPoint and let you process it before it start handling messages. This allows you to add handler or/and avoid data races.

func NewEndPoint

func NewEndPoint(stream Stream) EndPoint

NewEndPoint returns an EndPoint which already process incomming messages. Since no handler have been register at the time of the creation of the EndPoint, any message receive will be droped until and Handler is registered. Prefer EndPointFinalizer for a safe way to construct EndPoint.

type Filter

type Filter func(hdr *Header) (matched bool, keep bool)

Filter returns true if given message shall be processed by a Consumer. Returns two values: - matched: true if the message should be processed by the Consumer. - keep: true if the handler shall be kept in the dispatcher.

type Handler

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

Handler represents a client of a incomming message stream. It contains a filter used to decided if the message shall be sent to the handler, consumer will receive the messages in order of arrival. If an error occurs, closer is called, then the consumer is closed.

func NewHandler

func NewHandler(f Filter, ch chan<- *Message, cl Closer) *Handler

NewHandler returns an Handler: f is call on each incomming message, c is called if f returns true. cl is always called when the handler is effectively closed.

type Header

type Header struct {
	Magic   uint32 // magic number
	ID      uint32 // message id
	Size    uint32 // size of the payload
	Version uint16 // protocol version
	Type    uint8  // type of the message
	Flags   uint8  // flags
	Service uint32 // service id
	Object  uint32 // object id
	Action  uint32 // function or event id
}

Header represents a message header.

func NewHeader

func NewHeader(typ uint8, service uint32, object uint32, action uint32, id uint32) Header

NewHeader construct a message header given some parameters. The size of the message is zero.

func (*Header) Read

func (h *Header) Read(r io.Reader) (err error)

Read parses a message header from an io.Reader.

func (Header) String

func (h Header) String() string

func (*Header) Write

func (h *Header) Write(w io.Writer) (err error)

type KeyNetContext

type KeyNetContext uint32

KeyNetContext represents an entry in the sream context.

const (
	// DialAddress is the addressed dialed.
	DialAddress KeyNetContext = iota
	// ListenAddress is the addressed listen.
	ListenAddress
)

type Listener

type Listener interface {
	Accept() (Stream, error)
	Close() error
}

Listener accepts incomming connections in the form of Stream.

func Listen

func Listen(addr string) (Listener, error)

Listen reads the transport of addr and listen at the address. addr can be of the form: unix://, tcp:// or tcps://.

type Message

type Message struct {
	Header  Header
	Payload []byte
}

Message represents a QiMessaging message.

func NewMessage

func NewMessage(header Header, payload []byte) Message

NewMessage assemble an header and a payload to create a message. The size filed of the header is adjusted if necessary.

func (*Message) Read

func (m *Message) Read(r io.Reader) error

Read unmarshal a message from io.Reader. First the header is read, then if correct the payload is read. The payload will not be read if the header is not considerred well formatted. Forwards io.EOF if nothing was read.

func (*Message) Write

func (m *Message) Write(w io.Writer) error

Write marshal a message into an io.Writer. The header and the payload are written in a single write operation. Forwards io.EOF if nothing was written.

type Stream

type Stream interface {
	io.Reader
	io.Writer
	io.Closer
	fmt.Stringer
	Context() context.Context
}

Stream represents a network connection. Stream abstracts connections to allow for various transports.

func ConnStream

func ConnStream(conn gonet.Conn) Stream

ConnStream construct a Stream from a connection.

func PipeStream

func PipeStream(r, w *os.File) Stream

PipeStream returns a Stream based on the pipe:// protocol