grav

Documentation

Index

• Constants
• Variables
• type ConnectFunc
• type Discovery
• type DiscoveryOpts
• type Grav
  • func New(opts ...OptionsModifier) *Grav
  • func (g *Grav) Connect() *Pod
  • func (g *Grav) ConnectEndpoint(endpoint string) error
  • func (g *Grav) ConnectEndpointWithReplay(endpoint string) error
  • func (g *Grav) ConnectWithReplay() *Pod
• type Message
  • func MsgFromBytes(bytes []byte) (Message, error)
  • func MsgFromRequest(r *http.Request) (Message, error)
  • func NewMsg(msgType string, data []byte) Message
  • func NewMsgReplyTo(ticket MessageTicket, msgType string, data []byte) Message
  • func NewMsgWithParentID(msgType, parentID string, data []byte) Message
• type MessageTicket
• type MsgChan
• type MsgFunc
• type Options
• type OptionsModifier
  • func UseDiscovery(discovery Discovery) OptionsModifier
  • func UseLogger(logger *vlog.Logger) OptionsModifier
  • func UsePort(port string) OptionsModifier
  • func UseTransport(transport Transport) OptionsModifier
• type Pod
  • func (p *Pod) Disconnect()
  • func (mf Pod) FilterType(msgType string, allow bool)
  • func (mf Pod) FilterUUID(uuid string, allow bool)
  • func (p *Pod) On(onFunc MsgFunc)
  • func (p *Pod) OnType(onFunc MsgFunc, msgTypes ...string)
  • func (p *Pod) ReplyTo(ticket MessageTicket, msg Message)
  • func (p *Pod) Send(msg Message)
  • func (p *Pod) SendAndWaitOnReply(msg Message, onFunc MsgFunc, timeoutSeconds ...int) error
  • func (p *Pod) WaitOn(onFunc MsgFunc, timeoutSeconds ...int) error
  • func (p *Pod) WaitOnReply(ticket MessageTicket, onFunc MsgFunc, timeoutSeconds ...int) error
• type Transport
• type TransportHandshake
• type TransportHandshakeAck
• type TransportOpts

Constants

const (
	TransportMsgTypeHandshake = 1
	TransportMsgTypeUser      = 2
)

TransportMsgTypeHandshake and others represent internal Transport message types used for handshakes and metadata transfer

const (
	MsgTypeDefault string = "grav.default"
)

MsgTypeDefault and other represent message consts

Variables

var ErrMsgNotWanted = errors.New("message not wanted")

ErrMsgNotWanted is used by WaitOn to determine if the current message is what's being waited on

var (
	ErrTransportNotConfigured = errors.New("transport plugin not configured")
)

ErrTransportNotConfigured represent package-level vars

var ErrWaitTimeout = errors.New("waited past timeout")

ErrWaitTimeout is returned if a timeout is exceeded

Types

type ConnectFunc

type ConnectFunc func() *Pod

ConnectFunc represents a function that returns a pod conntected to Grav

type Discovery

type Discovery interface {
	Start(*DiscoveryOpts, Transport, ConnectFunc) error
}

Discovery represents a discovery plugin

type DiscoveryOpts

type DiscoveryOpts struct {
	NodeUUID      string
	TransportPort string
	Logger        *vlog.Logger
	Custom        interface{}
}

DiscoveryOpts is a set of options for transports

type Grav

type Grav struct {
	NodeUUID string
	// contains filtered or unexported fields
}

Grav represents a Grav message bus instance

func New

func New(opts ...OptionsModifier) *Grav

New creates a new Grav with the provided options

func (*Grav) Connect

func (g *Grav) Connect() *Pod

Connect creates a new connection (pod) to the bus

func (*Grav) ConnectEndpoint

func (g *Grav) ConnectEndpoint(endpoint string) error

ConnectEndpoint uses the configured transport to connect the bus to an external endpoint

func (*Grav) ConnectEndpointWithReplay

func (g *Grav) ConnectEndpointWithReplay(endpoint string) error

ConnectEndpointWithReplay uses the configured transport to connect the bus to an external endpoint and replays recent messages to the endpoint when the pod registers its onFunc

func (*Grav) ConnectWithReplay

func (g *Grav) ConnectWithReplay() *Pod

ConnectWithReplay creates a new connection (pod) to the bus and replays recent messages when the pod sets its onFunc

type Message

type Message interface {
	// Unique ID for this message
	UUID() string
	// ID of the parent event or request, such as HTTP request
	ParentID() string
	// The UUID of the message being replied to, if any
	ReplyTo() string
	// Allow setting a message UUID that this message is a response to
	SetReplyTo(string)
	// Get a MessageTicket that references this message
	Ticket() MessageTicket
	// Type of message (application-specific)
	Type() string
	// Time the message was sent
	Timestamp() time.Time
	// Raw data of message
	Data() []byte
	// Unmarshal the message's data into a struct
	UnmarshalData(interface{}) error
	// Marshal the message itself to encoded bytes (JSON or otherwise)
	Marshal() ([]byte, error)
	// Unmarshal encoded Message into object
	Unmarshal([]byte) error
}

Message represents a message

func MsgFromBytes

func MsgFromBytes(bytes []byte) (Message, error)

MsgFromBytes returns a default _message that has been unmarshalled from bytes. Should only be used if the default _message type is being used.

func MsgFromRequest

func MsgFromRequest(r *http.Request) (Message, error)

MsgFromRequest extracts an encoded Message from an HTTP request

func NewMsg

func NewMsg(msgType string, data []byte) Message

NewMsg creates a new Message with the built-in `_message` type

func NewMsgReplyTo

func NewMsgReplyTo(ticket MessageTicket, msgType string, data []byte) Message

NewMsgReplyTo creates a new message in response to a previous message

func NewMsgWithParentID

func NewMsgWithParentID(msgType, parentID string, data []byte) Message

NewMsgWithParentID returns a new message with the provided parent ID

type MessageTicket

type MessageTicket struct {
	UUID string
}

MessageTicket represents a "ticket" that references a message that was sent with the hopes of getting a response

type MsgChan

type MsgChan chan Message

MsgChan is a channel that accepts a message

type MsgFunc

type MsgFunc func(Message) error

MsgFunc is a callback function that accepts a message and returns an error

type Options

type Options struct {
	Logger    *vlog.Logger
	Port      string
	Transport Transport
	Discovery Discovery
}

Options represent Grav options

type OptionsModifier

type OptionsModifier func(*Options)

OptionsModifier is function that modifies an option

func UseDiscovery

func UseDiscovery(discovery Discovery) OptionsModifier

UseDiscovery sets the discovery plugin to be used

func UseLogger

func UseLogger(logger *vlog.Logger) OptionsModifier

UseLogger allows a custom logger to be used

func UsePort

func UsePort(port string) OptionsModifier

UsePort sets the port that will be advertised by discovery

func UseTransport

func UseTransport(transport Transport) OptionsModifier

UseTransport sets the transport plugin to be used

type Pod

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

Pod is a connection to Grav Pods are bi-directional. Messages can be sent to them from the bus, and they can be used to send messages to the bus. Pods are meant to be extremely lightweight with no persistence they are meant to quickly and immediately route a message between its owner and the Bus. The Bus is responsible for any "smarts". Messages coming from the bus are filtered using the pod's messageFilter, which is configurable by the caller.

func (*Pod) Disconnect

func (p *Pod) Disconnect()

Disconnect indicates to the bus that this pod is no longer needed and should be disconnected. Sending will immediately become unavailable, and the pod will soon stop recieving messages.

func (Pod) FilterType

func (mf Pod) FilterType(msgType string, allow bool)

func (Pod) FilterUUID

func (mf Pod) FilterUUID(uuid string, allow bool)

FilterUUID likely should not be used in normal cases, it adds a message UUID to the pod's filter.

func (*Pod) On

func (p *Pod) On(onFunc MsgFunc)

On sets the function to be called whenever this pod recieves a message from the bus. If nil is passed, the pod will ignore all messages. Calling On multiple times causes the function to be overwritten. To recieve using two different functions, create two pods.

func (*Pod) OnType

func (p *Pod) OnType(onFunc MsgFunc, msgTypes ...string)

OnType sets the function to be called whenever this pod recieves a message and sets the pod's filter to only include certain message types

func (*Pod) ReplyTo

func (p *Pod) ReplyTo(ticket MessageTicket, msg Message)

ReplyTo sends a response to a message

func (*Pod) Send

func (p *Pod) Send(msg Message)

Send emits a message to be routed to the bus

func (*Pod) SendAndWaitOnReply

func (p *Pod) SendAndWaitOnReply(msg Message, onFunc MsgFunc, timeoutSeconds ...int) error

SendAndWaitOnReply sends a message and then blocks until a message is recieved in ReplyTo that message

func (*Pod) WaitOn

func (p *Pod) WaitOn(onFunc MsgFunc, timeoutSeconds ...int) error

WaitOn takes a function to be called whenever this pod recieves a message and blocks until that function returns something other than ErrMsgNotWanted. WaitOn should be used if there is a need to wait for a particular message. When the onFunc returns something other than ErrMsgNotWanted (such as nil or a different error), WaitOn will return and set the onFunc to nil. If an error other than ErrMsgNotWanted is returned from the onFunc, it will be propogated to the caller. An optional timeout (default 10s) can be provided (only the first value will be used). If the timeout is exceeded, ErrWaitTimeout is returned.

func (*Pod) WaitOnReply

func (p *Pod) WaitOnReply(ticket MessageTicket, onFunc MsgFunc, timeoutSeconds ...int) error

WaitOnReply waits on a reply message to arrive at the pod and then calls onFunc with that message. If the onFunc produces an error, it will be propogated to the caller. an optionsl timrout can be provided (only the first value will be used)

type Transport

type Transport interface {
	// Serve is a transport-specific function that exposes a connection point
	Serve(*TransportOpts, ConnectFunc) error
	// ConnectEndpoint indicates to the Transport that a connection to a remote endpoint is needed
	ConnectEndpoint(string, ConnectFunc) error
	// ConnectEndpointWithUUID connects to an endpoint with a known identifier
	ConnectEndpointWithUUID(string, string, ConnectFunc) error
}

Transport represents a Grav transport plugin

type TransportHandshake

type TransportHandshake struct {
	UUID string `json:"uuid"`
}

TransportHandshake represents a handshake sent to a node that you're trying to connect to

type TransportHandshakeAck

type TransportHandshakeAck struct {
	UUID string `json:"uuid"`
}

TransportHandshakeAck represents a handshake response

type TransportOpts

type TransportOpts struct {
	NodeUUID string
	Port     string
	Logger   *vlog.Logger
	Custom   interface{}
}

TransportOpts is a set of options for transports