messenger

README

####Benchmark of the messenger.

$ go test -v -run=Benckmark* -bench=. 
PASS
BenchmarkMessengerSendSmallMessage	   50000	     70568 ns/op
BenchmarkMessengerSendMediumMessage	   50000	     70265 ns/op
BenchmarkMessengerSendBigMessage	   50000	     72693 ns/op
BenchmarkMessengerSendLargeMessage	   50000	     72896 ns/op
BenchmarkMessengerSendMixedMessage	   50000	     72631 ns/op
BenchmarkMessengerSendRecvSmallMessage	   20000	     78409 ns/op
BenchmarkMessengerSendRecvMediumMessage	   20000	     80471 ns/op
BenchmarkMessengerSendRecvBigMessage	   20000	     82629 ns/op
BenchmarkMessengerSendRecvLargeMessage	   20000	     85987 ns/op
BenchmarkMessengerSendRecvMixedMessage	   20000	     83678 ns/op
ok  	github.com/mesos/mesos-go/messenger	115.135s

$ go test -v -run=Benckmark* -bench=. -cpu=4 -send-routines=4 2>/dev/null
PASS
BenchmarkMessengerSendSmallMessage-4	   50000	     35529 ns/op
BenchmarkMessengerSendMediumMessage-4	   50000	     35997 ns/op
BenchmarkMessengerSendBigMessage-4	   50000	     36871 ns/op
BenchmarkMessengerSendLargeMessage-4	   50000	     37310 ns/op
BenchmarkMessengerSendMixedMessage-4	   50000	     37419 ns/op
BenchmarkMessengerSendRecvSmallMessage-4	   50000	     39320 ns/op
BenchmarkMessengerSendRecvMediumMessage-4	   50000	     41990 ns/op
BenchmarkMessengerSendRecvBigMessage-4	   50000	     42157 ns/op
BenchmarkMessengerSendRecvLargeMessage-4	   50000	     45472 ns/op
BenchmarkMessengerSendRecvMixedMessage-4	   50000	     47393 ns/op
ok  	github.com/mesos/mesos-go/messenger	105.173s

####environment:

OS: Linux yifan-laptop 3.13.0-32-generic #57-Ubuntu SMP Tue Jul 15 03:51:08 UTC 2014 x86_64 x86_64 x86_64 GNU/Linux
CPU: Intel(R) Core(TM) i5-3210M CPU @ 2.50GHz
MEM: 4G DDR3 1600MHz

Documentation

Overview

Package messenger includes a messenger and a transporter. The messenger provides interfaces to send a protobuf message through the underlying transporter. It also dispatches messages to installed handlers.

Index

• type HTTPTransporter
  • func NewHTTPTransporter(upid *upid.UPID) *HTTPTransporter
  • func (t *HTTPTransporter) Inject(ctx context.Context, msg *Message) error
  • func (t *HTTPTransporter) Install(msgName string)
  • func (t *HTTPTransporter) Listen() error
  • func (t *HTTPTransporter) Recv() *Message
  • func (t *HTTPTransporter) Send(ctx context.Context, msg *Message) (sendError error)
  • func (t *HTTPTransporter) Start() error
  • func (t *HTTPTransporter) Stop() error
  • func (t *HTTPTransporter) UPID() *upid.UPID
• type MesosMessenger
  • func New(upid *upid.UPID, t Transporter) *MesosMessenger
  • func NewHttp(upid *upid.UPID) *MesosMessenger
  • func (m *MesosMessenger) Install(handler MessageHandler, msg proto.Message) error
  • func (m *MesosMessenger) Route(ctx context.Context, upid *upid.UPID, msg proto.Message) error
  • func (m *MesosMessenger) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error
  • func (m *MesosMessenger) Start() error
  • func (m *MesosMessenger) Stop() error
  • func (m *MesosMessenger) UPID() *upid.UPID
• type Message
  • func (m *Message) RequestURI() string
• type MessageHandler
• type Messenger
• type MockedMessenger
  • func NewMockedMessenger() *MockedMessenger
  • func (m *MockedMessenger) Install(handler MessageHandler, msg proto.Message) error
  • func (m *MockedMessenger) Recv(from *upid.UPID, msg proto.Message)
  • func (m *MockedMessenger) Route(ctx context.Context, upid *upid.UPID, msg proto.Message) error
  • func (m *MockedMessenger) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error
  • func (m *MockedMessenger) Start() error
  • func (m *MockedMessenger) Stop() error
  • func (m *MockedMessenger) UPID() *upid.UPID
• type Transporter

Types

type HTTPTransporter

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

HTTPTransporter implements the interfaces of the Transporter.

func NewHTTPTransporter

func NewHTTPTransporter(upid *upid.UPID) *HTTPTransporter

NewHTTPTransporter creates a new http transporter.

func (*HTTPTransporter) Inject

func (t *HTTPTransporter) Inject(ctx context.Context, msg *Message) error

Inject places a message into the incoming message queue.

func (*HTTPTransporter) Install

func (t *HTTPTransporter) Install(msgName string)

Install the request URI according to the message's name.

func (*HTTPTransporter) Listen

func (t *HTTPTransporter) Listen() error

Listen starts listen on UPID. If UPID is empty, the transporter will listen on a random port, and then fill the UPID with the host:port it is listening.

func (*HTTPTransporter) Recv

func (t *HTTPTransporter) Recv() *Message

Recv returns the message, one at a time.

func (*HTTPTransporter) Send

func (t *HTTPTransporter) Send(ctx context.Context, msg *Message) (sendError error)

Send sends the message to its specified upid.

func (*HTTPTransporter) Start

func (t *HTTPTransporter) Start() error

Start starts the http transporter. This will block, should be put in a goroutine.

func (*HTTPTransporter) Stop

func (t *HTTPTransporter) Stop() error

Stop stops the http transporter by closing the listener.

func (*HTTPTransporter) UPID

func (t *HTTPTransporter) UPID() *upid.UPID

UPID returns the upid of the transporter.

type MesosMessenger

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

MesosMessenger is an implementation of the Messenger interface.

func New

func New(upid *upid.UPID, t Transporter) *MesosMessenger

func NewHttp

func NewHttp(upid *upid.UPID) *MesosMessenger

NewMesosMessenger creates a new mesos messenger.

func (*MesosMessenger) Install

func (m *MesosMessenger) Install(handler MessageHandler, msg proto.Message) error

/ Install installs the handler with the given message.

func (*MesosMessenger) Route

func (m *MesosMessenger) Route(ctx context.Context, upid *upid.UPID, msg proto.Message) error

Route puts a message either in the incoming or outgoing queue. This method is useful for: 1) routing internal error to callback handlers 2) testing components without starting remote servers.

func (*MesosMessenger) Send

func (m *MesosMessenger) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error

Send puts a message into the outgoing queue, waiting to be sent. With buffered channels, this will not block under moderate throughput. When an error is generated, the error can be communicated by placing a message on the incoming queue to be handled upstream.

func (*MesosMessenger) Start

func (m *MesosMessenger) Start() error

Start starts the messenger.

func (*MesosMessenger) Stop

func (m *MesosMessenger) Stop() error

Stop stops the messenger and clean up all the goroutines.

func (*MesosMessenger) UPID

func (m *MesosMessenger) UPID() *upid.UPID

UPID returns the upid of the messenger.

type Message

type Message struct {
	UPID         *upid.UPID
	Name         string
	ProtoMessage proto.Message
	Bytes        []byte
}

Message defines the type that passes in the Messenger.

func (*Message) RequestURI

func (m *Message) RequestURI() string

RequestURI returns the request URI of the message.

type MessageHandler

type MessageHandler func(from *upid.UPID, pbMsg proto.Message)

MessageHandler is the callback of the message. When the callback is invoked, the sender's upid and the message is passed to the callback.

type Messenger

type Messenger interface {
	Install(handler MessageHandler, msg proto.Message) error
	Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error
	Route(ctx context.Context, from *upid.UPID, msg proto.Message) error
	Start() error
	Stop() error
	UPID() *upid.UPID
}

Messenger defines the interfaces that should be implemented.

type MockedMessenger

type MockedMessenger struct {
	mock.Mock
	// contains filtered or unexported fields
}

MockedMessenger is a messenger that returns error on every operation.

func NewMockedMessenger

func NewMockedMessenger() *MockedMessenger

NewMockedMessenger returns a mocked messenger used for testing.

func (*MockedMessenger) Install

func (m *MockedMessenger) Install(handler MessageHandler, msg proto.Message) error

Install is a mocked implementation.

func (*MockedMessenger) Recv

func (m *MockedMessenger) Recv(from *upid.UPID, msg proto.Message)

Recv receives a upid and a message, it will dispatch the message to its handler with the upid. This is for testing.

func (*MockedMessenger) Route

func (m *MockedMessenger) Route(ctx context.Context, upid *upid.UPID, msg proto.Message) error

func (*MockedMessenger) Send

func (m *MockedMessenger) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error

Send is a mocked implementation.

func (*MockedMessenger) Start

func (m *MockedMessenger) Start() error

Start is a mocked implementation.

func (*MockedMessenger) Stop

func (m *MockedMessenger) Stop() error

Stop is a mocked implementation.

func (*MockedMessenger) UPID

func (m *MockedMessenger) UPID() *upid.UPID

UPID is a mocked implementation.

type Transporter

type Transporter interface {
	//Send sends message to remote process. Must use context to determine
	//cancelled requests.
	Send(ctx context.Context, msg *Message) error

	//Listen blocks and listens for incoming messages.
	Listen() error

	//Rcvd receives and delegate message handling to installed handlers.
	Recv() *Message

	//Inject injects a message to the incoming queue. Must use context to
	//determine cancelled requests.
	Inject(ctx context.Context, msg *Message) error

	//Install mount an handler based on incoming message name.
	Install(messageName string)

	//Start starts the transporter.
	Start() error

	//Stop kills the transporter.
	Stop() error

	//UPID returns the PID for transporter.
	UPID() *upid.UPID
}

Transporter defines methods for communicating with remote processes.