Messaging Library

This package provides a Go library that simplifies the implementation of communication patterns, like request/reply, on top of a messaging broker that supports queues and topics (destinations).

Building

To build the project clone the repository to your go path and run the make command.

To build the example messaging-tool and a testing broker messaging-server run the make binaries command.

The make binaries command will install the binaries into ./.gopath/bin/ path instead of $GOPATH/bin or $GOBIN, when executing the examples below use the correct path (e.g. run ./.gopath/bin/messaging-server if ./.gopath/bin is not in your $PATH).

Examples

List of examples

messaging-tool

messaging-server

request-tool

Running

Run messaging-server testing Server:

$ messaging-server serve

Run messaging-tool testing tool, and wait for incoming messages:

$ messaging-tool receive --host 127.0.0.1 --destination "hello"

Run messaging-tool testing tool, and send a messages:

$ messaging-tool send --host 127.0.0.1 --destination "hello" --body "world"

Run request-tool testing tool, and start the request echo-server:

$ request-tool respond --host 127.0.0.1 --requests-queue "requests"

Run request-tool testing tool, and send a request:

$ request-tool send --host 127.0.0.1 --requests-queue "requests" --responses-queue "response1" --body "my request"

Usage

Publish a string message to a destination (queue or topic)

Example:

send.go

import (
	...

	// Import the message broker client interface.
	"github.com/container-mgmt/messaging-library/pkg/client"

	// Import a STOMP client implementation.
	"github.com/container-mgmt/messaging-library/pkg/connections/stomp"
)

...

// Create a new connection object.
var c client.Connection

// Set a new STOMP connction to localhost:1888.
c, err = stomp.NewConnection(&client.ConnectionSpec{
	BrokerHost: "localhost",
	BrokerPort: 1888,
})
if err != nil {
	...
}
defer c.Close()

// Set some text to the message.
messageText := "Hello world"

// Create a message with data object to send to the message broker,
// the data payload must be of type client.MessageData{}.
m := client.Message{
	Data: client.MessageData{
		"kind": "helloMessage",
		"spec": map[string]string{"message": messageText},
	},
}

// Publish our hello message to the "destination name" destination.
err = c.Publish(m, "destination name")

Subscribe to a destination (queue or topic)

Example:

receive.go

import (
	...

	// Import the message broker client interface.
	"github.com/container-mgmt/messaging-library/pkg/client"

	// Import a STOMP client implementation.
	"github.com/container-mgmt/messaging-library/pkg/connections/stomp"
)

...

// We will use this function as a callback function for each new message
// published on specific destination.
//
// m.Data is of type client.MessageData{}
func callback(m client.Message, destination string) (err error) {
	glog.Infof(
		"Received message from destination '%s':\n%v",
		destination,
		m.Data,
	)

	return
}

...

// Create a new connection object.
var c client.Connection

// Set a new STOMP connction to localhost:1888.
c, err = stomp.NewConnection(&client.ConnectionSpec{
	BrokerHost: "localhost",
	BrokerPort: 1888,
})
if err != nil {
	...
}
defer c.Close()

...

// Subscribe to the destination "destination name", and run callback function for each
// new message.
err = c.Subscribe("destination name", callback)

Create a responder on a queue

Example:

respond.go

import (
	...

	// Import the message broker client interface.
	"github.com/container-mgmt/messaging-library/pkg/client"

	// Import a STOMP client implementation.
	"github.com/container-mgmt/messaging-library/pkg/connections/stomp"
)

...

// We will use this function as a callback function for each new reqquest
// received on the queue.
//
// request.Data is of type client.MessageData{}
func requestHandler(request client.Message) (response client.Message, err error) {
	glog.Infof(
		"Received request:\n%v",
		request.Data,
	)

	return
}

...

// Create a new connection object.
var c client.Connection

// Set a new STOMP connction to localhost:1888.
c, err = stomp.NewConnection(&client.ConnectionSpec{
	BrokerHost: "localhost",
	BrokerPort: 1888,
})
if err != nil {
	...
}
defer c.Close()

...

// Create a responder on the responses queue
r, err := c.NewResponder(
	client.ResponderSpec{
		RequestsQueue: "requests-queue",
		Callback:      requestHandler,
	})

if err != nil {
	...
}

defer r.Close()
...

Create a requestor that sends requests to a queue

Example:

request.go

import (
	...

	// Import the message broker client interface.
	"github.com/container-mgmt/messaging-library/pkg/client"

	// Import a STOMP client implementation.
	"github.com/container-mgmt/messaging-library/pkg/connections/stomp"
)

...

// We will use this function as a callback function a response
// received on the responses queue.
//
// response.Data is of type client.MessageData{}
func responseHandler(response client.Message, requestID string) (err error) {
	glog.Infof(
		"Received response for request id: %s:\n%v",
		requestID,
		response.Data,
	)

	return
}

...

// Create a new connection object.
var c client.Connection

// Set a new STOMP connction to localhost:1888.
c, err = stomp.NewConnection(&client.ConnectionSpec{
	BrokerHost: "localhost",
	BrokerPort: 1888,
})
if err != nil {
	...
}
defer c.Close()

...

// Create a requestor that sends requests to the requests-queue
// and receives responses on the responses queue
r, err := c.NewRequestor(
	client.RequestorSpec{
		RequestsQueue:  "requests-queue",
		ResponsesQueue: "responses-queue",
	})

if err != nil {
	...
}

defer r.Close()
...

// Send a request:
// Message data is a map [string]interface and can be populated
// with any data of that structure
request := client.Message{
	ContentType: contentType,
	Data: client.MessageData{
		"kind": "Request",
		"message": "this is my request"},
	},
}

// Send the request, and set the handler for the response
r.Send(m, responseHandler)

Running Tests and Benchmarks

Benchmarks and Tests should be run using an external STOMP broker.

Before running the Benchmarks or Tests, start a STOMP broker, for example ActiveMQ Artemis:

Run an external ActiveMQ Artemis broker:

$ artimis run

You can use the make command to run tests and benchmarks on your local computer.

To run the tests, use the make test command.

Run unit tests:

$ make test

Run the benchmark using the make command:

$ make bench