disp

Documentation

Overview

Package disp implements a generic message dispatcher for request/reply protocols.

Supported message exchanges are Request (send a request message, block until a reply message with the same key is received), Notify (send a reliable notification, i.e., one that is either sent via a lower-level reliable transport or waits for an ACK on an unreliable transport), and NotifyUnreliable (send a message, return immediately).

A Dispatcher can be customized by implementing interface MessageAdapter. The interface instructs the dispatcher how to convert a message to its raw representation, how to parse a raw representation into a message, how to determine which messages are replies and how to extract keys (unique IDs) from messages.

Protocols must clearly differentiate between request/notifications and replies. This is done by implementing MessageAdapter.IsReply. Replies are only used to finalize pending requests, and are not propagated back to the app via RecvFrom.

Requests and Replies are paired via Keys. Once a request is sent out, its key is stored internally. If a reply is received for that same key, the request is marked as fulfilled and the waiting goroutine returns. If no request is outstanding for a reply key, it is ignored.

Index

• type Dispatcher
  • func New(conn net.PacketConn, adapter MessageAdapter, logger log.Logger) *Dispatcher
  • func (d *Dispatcher) Close(ctx context.Context) error
  • func (d *Dispatcher) Notify(ctx context.Context, msg proto.Cerealizable, address net.Addr) error
  • func (d *Dispatcher) NotifyUnreliable(ctx context.Context, msg proto.Cerealizable, address net.Addr) error
  • func (d *Dispatcher) RecvFrom(ctx context.Context) (proto.Cerealizable, int, net.Addr, error)
  • func (d *Dispatcher) Request(ctx context.Context, msg proto.Cerealizable, address net.Addr) (proto.Cerealizable, error)
• type MessageAdapter

Types

type Dispatcher

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

func New

func New(conn net.PacketConn, adapter MessageAdapter, logger log.Logger) *Dispatcher

New creates a new dispatcher backed by transport t, and using adapter to convert generic Message objects to and from their raw representation.

All methods guarantee to return immediately once their context expires. Calling the context's cancel function does not guarantee immediate return (lower levels might be blocked on an uninterruptible call).

A Dispatcher can be safely used by concurrent goroutines.

func (*Dispatcher) Close

func (d *Dispatcher) Close(ctx context.Context) error

Close shuts down the background goroutine and closes the transport.

func (*Dispatcher) Notify

func (d *Dispatcher) Notify(ctx context.Context, msg proto.Cerealizable, address net.Addr) error

Notify sends msg to address in a reliable way (i.e., either via a lower-level reliable transport or by waiting for an ACK on an unreliable transport).

func (*Dispatcher) NotifyUnreliable

func (d *Dispatcher) NotifyUnreliable(ctx context.Context, msg proto.Cerealizable,
	address net.Addr) error

NotifyUnreliable sends msg to address, and returns immediately.

func (*Dispatcher) RecvFrom

func (d *Dispatcher) RecvFrom(ctx context.Context) (proto.Cerealizable, int, net.Addr, error)

RecvFrom returns the next non-reply message.

func (*Dispatcher) Request

func (d *Dispatcher) Request(ctx context.Context, msg proto.Cerealizable,
	address net.Addr) (proto.Cerealizable, error)

Request sends msg to address, and returns a reply with the same key. This method always blocks while waiting for the response.

No type validations are performed. Upper layer code should verify whether the message is the expected type.

type MessageAdapter

type MessageAdapter interface {
	// Convert msg to a format suitable for sending on a wire
	MsgToRaw(msg proto.Cerealizable) (common.RawBytes, error)
	// Convert a raw byte slice to a message
	RawToMsg(common.RawBytes) (proto.Cerealizable, error)
	// Return a key used to match requests and replies
	MsgKey(proto.Cerealizable) string
}

MessageAdapter converts application level messages to and from elements the Dispatcher understands.