messaging

Documentation

Overview

Package messaging handles lower-level communication between a transaction processor and validator.

Index

• func DumpMsg(t validator_pb2.Message_MessageType, c []byte, corrId string) ([]byte, error)
• func GenerateId() string
• func LoadMsg(data []byte) (msg *validator_pb2.Message, err error)
• type Connection
• type ZmqConnection
  • func NewConnection(context *zmq.Context, t zmq.Type, uri string, bind bool) (*ZmqConnection, error)
  • func (self *ZmqConnection) Close()
  • func (self *ZmqConnection) Identity() string
  • func (self *ZmqConnection) Monitor(events zmq.Event) (*zmq.Socket, error)
  • func (self *ZmqConnection) RecvData() (string, []byte, error)
  • func (self *ZmqConnection) RecvMsg() (string, *validator_pb2.Message, error)
  • func (self *ZmqConnection) RecvMsgWithId(corrId string) (string, *validator_pb2.Message, error)
  • func (self *ZmqConnection) SendData(id string, data []byte) error
  • func (self *ZmqConnection) SendMsg(t validator_pb2.Message_MessageType, c []byte, corrId string) error
  • func (self *ZmqConnection) SendMsgTo(id string, t validator_pb2.Message_MessageType, c []byte, corrId string) error
  • func (self *ZmqConnection) SendNewMsg(t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)
  • func (self *ZmqConnection) SendNewMsgTo(id string, t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)
  • func (self *ZmqConnection) Socket() *zmq.Socket

Functions

func DumpMsg

func DumpMsg(t validator_pb2.Message_MessageType, c []byte, corrId string) ([]byte, error)

DumpMsg serializes a validator message

func GenerateId

func GenerateId() string

Generate a new UUID

func LoadMsg

func LoadMsg(data []byte) (msg *validator_pb2.Message, err error)

LoadMsg deserializes a validator message

Types

type Connection

type Connection interface {
	SendData(id string, data []byte) error
	SendNewMsg(t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)
	SendNewMsgTo(id string, t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)
	SendMsg(t validator_pb2.Message_MessageType, c []byte, corrId string) error
	SendMsgTo(id string, t validator_pb2.Message_MessageType, c []byte, corrId string) error
	RecvData() (string, []byte, error)
	RecvMsg() (string, *validator_pb2.Message, error)
	RecvMsgWithId(corrId string) (string, *validator_pb2.Message, error)
	Close()
	Socket() *zmq.Socket
	Monitor(zmq.Event) (*zmq.Socket, error)
	Identity() string
}

type ZmqConnection

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

Connection wraps a ZMQ DEALER socket or ROUTER socket and provides some utility methods for sending and receiving messages.

func NewConnection

func NewConnection(context *zmq.Context, t zmq.Type, uri string, bind bool) (*ZmqConnection, error)

NewConnection establishes a new connection using the given ZMQ context and socket type to the given URI.

func (*ZmqConnection) Close

func (self *ZmqConnection) Close()

Close closes the wrapped socket. This should be called with defer() after opening the socket.

func (*ZmqConnection) Identity

func (self *ZmqConnection) Identity() string

Identity returns the identity assigned to the wrapped socket.

func (*ZmqConnection) Monitor

func (self *ZmqConnection) Monitor(events zmq.Event) (*zmq.Socket, error)

Create a new monitor socket pair and return the socket for listening

func (*ZmqConnection) RecvData

func (self *ZmqConnection) RecvData() (string, []byte, error)

RecvData receives a ZMQ message from the wrapped socket and returns the identity of the sender and the data sent. If ZmqConnection does not wrap a ROUTER socket, the identity returned will be "".

func (*ZmqConnection) RecvMsg

func (self *ZmqConnection) RecvMsg() (string, *validator_pb2.Message, error)

RecvMsg receives a new validator message and returns it deserialized. If ZmqConnection wraps a ROUTER socket, id will be the identity of the sender. Otherwise, id will be "".

func (*ZmqConnection) RecvMsgWithId

func (self *ZmqConnection) RecvMsgWithId(corrId string) (string, *validator_pb2.Message, error)

RecvMsgWithId receives validator messages until a message with the given correlation id is found and returns this message. Any messages received that do not match the id are saved for subsequent receives.

func (*ZmqConnection) SendData

func (self *ZmqConnection) SendData(id string, data []byte) error

SendData sends the byte array.

If id is not "", the id is included as the first part of the message. This is useful for passing messages to a ROUTER socket so it can route them.

func (*ZmqConnection) SendMsg

func (self *ZmqConnection) SendMsg(t validator_pb2.Message_MessageType, c []byte, corrId string) error

Send a message with the given correlation id

func (*ZmqConnection) SendMsgTo

func (self *ZmqConnection) SendMsgTo(id string, t validator_pb2.Message_MessageType, c []byte, corrId string) error

Send a message with the given correlation id and the prepends the id like SendNewMsgTo()

func (*ZmqConnection) SendNewMsg

func (self *ZmqConnection) SendNewMsg(t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)

SendNewMsg creates a new validator message, assigns a new correlation id, serializes it, and sends it. It returns the correlation id created.

func (*ZmqConnection) SendNewMsgTo

func (self *ZmqConnection) SendNewMsgTo(id string, t validator_pb2.Message_MessageType, c []byte) (corrId string, err error)

SendNewMsgTo sends a new message validator message with the given id sent as the first part of the message. This is required when sending to a ROUTER socket, so it knows where to route the message.

func (*ZmqConnection) Socket

func (self *ZmqConnection) Socket() *zmq.Socket

Socket returns the wrapped socket.