eventbus

README

github.com/jponge/vertx-go-tcp-eventbus-bridge/eventbus

This is the package you should be using in your Go clients.

Documentation

Overview

This package provides a client API to the Vert.x TCP EventBus bridge, See http://vertx.io/docs/vertx-tcp-eventbus-bridge/java/ for pointers to the reference protocol.

Index

• type Dispatcher
  • func NewDispatcher(eventBus *EventBus) *Dispatcher
  • func (dispatcher *Dispatcher) Register(address string, chanSize uint32) (<-chan *Message, string, error)
  • func (dispatcher *Dispatcher) Start()
  • func (dispatcher *Dispatcher) Stop()
  • func (dispatcher *Dispatcher) Unregister(address, channelId string) error
• type EventBus
  • func NewEventBus(address string) (*EventBus, error)
  • func (eventBus *EventBus) Close() error
  • func (eventBus *EventBus) Publish(address string, headers, body interface{}) error
  • func (eventBus *EventBus) Receive() (*Message, error)
  • func (eventBus *EventBus) Register(address string) error
  • func (eventBus *EventBus) Send(address string, headers, body interface{}) error
  • func (eventBus *EventBus) SendWithReplyAddress(address, replyAddress string, headers, body interface{}) error
  • func (eventBus *EventBus) Unregister(address string) error
• type Message
  • func (msg *Message) IsError() bool

Types

type Dispatcher

type Dispatcher struct {
	ErrChan chan error
	// contains filtered or unexported fields
}

A dispatcher provides an idiomatic way to receive messages from remote Vert.x application event buses, and consume messages from Go channels.

The ErrChan channel is useful to be notified of an error while messages are being received. Only 1 error may be sent to that channel, in which case the dispatcher stops and will not receive further messages.

func NewDispatcher

func NewDispatcher(eventBus *EventBus) *Dispatcher

Makes a new dispatcher on top of an EventBus object.

func (*Dispatcher) Register

func (dispatcher *Dispatcher) Register(address string, chanSize uint32) (<-chan *Message, string, error)

Register a channel of size chanSize to listener on a destination.

The method returns a channel of messages, a registration key, or possibly a non-nil error.

This method is safe to use from concurrent goroutines.

func (*Dispatcher) Start

func (dispatcher *Dispatcher) Start()

Starts the dispatcher.

Messages will be received from a goroutine that this method starts. The corresponding goroutine runs until either:

- no more messages arrive, and Stop() has been called, or

- an error is detected while receiving messages, in which case the said error is sent to the ErrChan field of this object.

func (*Dispatcher) Stop

func (dispatcher *Dispatcher) Stop()

Stop this dispatcher.

func (*Dispatcher) Unregister

func (dispatcher *Dispatcher) Unregister(address, channelId string) error

Unregistration based on an address and a registration key.

This method simply returns nil shall the address or channelId values not match current registrations.

This method is safe to use from concurrent goroutines.

type EventBus

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

A connection to an event bus TCP bridge.

In all publishing methods, the headers and body parameters can be anything that the encoding/json package knows how to map. It is recommended to use either struct or map objects. The headers parameter can also be nil in case no header is useful.

func NewEventBus

func NewEventBus(address string) (*EventBus, error)

Connects to a remote Vert.x application over the event bus TCP bridge.

The address shall be specified as for 'net.Dial' connections, like 'somewhere.tld:port'.

An EventBus pointer or an error are returned.

func (*EventBus) Close

func (eventBus *EventBus) Close() error

Closes the connection to a remote Vert.x application.

func (*EventBus) Publish

func (eventBus *EventBus) Publish(address string, headers, body interface{}) error

Publish a message to an address.

This method is safe to use from concurrent goroutines.

func (*EventBus) Receive

func (eventBus *EventBus) Receive() (*Message, error)

Receive an incoming message from the remote Vert.x application event bus.

The message can be for any destination, it is up to the caller to decide how to dispatch it. You may want to use eventbus.Dispatcher to use Go channels.

This method is blocking and it is not safe to use from concurrent goroutines.

func (*EventBus) Register

func (eventBus *EventBus) Register(address string) error

Register this client to receive messages on a destination.

This method is safe to use from concurrent goroutines.

func (*EventBus) Send

func (eventBus *EventBus) Send(address string, headers, body interface{}) error

Send a message to an address.

This method is safe to use from concurrent goroutines.

func (*EventBus) SendWithReplyAddress

func (eventBus *EventBus) SendWithReplyAddress(address, replyAddress string, headers, body interface{}) error

Send a message to an address, and also specify a destination for an expected response. The response will be eventually fetched from the Receive() method.

This method is safe to use from concurrent goroutines.

func (*EventBus) Unregister

func (eventBus *EventBus) Unregister(address string) error

Unregisters this client from receiving messages from a destination.

This method is safe to use from concurrent goroutines.

type Message

type Message struct {
	Type         string      `json:"type"`
	ReplyAddress interface{} `json:"replyAddress,omitempty"`
	Address      string      `json:"address"`
	Headers      interface{} `json:"headers"`
	Body         interface{} `json:"body"`
}

A Vert.x event bus message that can be mapped to JSON.

func (*Message) IsError

func (msg *Message) IsError() bool

This method returns true if the Type field has value 'err'.