raven

Documentation

Index

• Variables
• func BindService(serviceName, queue string) error
• func Connect() chan bool
• func Consume(queue string) (deliveries <-chan amqp.Delivery, err error)
• func ConsumerNotifyClose() chan *amqp.Error
• func IsConnected() bool
• func SendHeartbeat(hb Heartbeat, InstanceID string) error
• func SendPublication(pub Publication, InstanceID string) error
• func SendRequest(req Request, InstanceID string) error
• func SendResponse(rsp Response, InstanceID string) error
• type AMQPChannel
  • func (self *AMQPChannel) Close() error
  • func (self *AMQPChannel) NotifyClose() chan *amqp.Error
• type AMQPConnection
  • func (self *AMQPConnection) Channel() (*amqp.Channel, error)
  • func (self *AMQPConnection) Close() error
  • func (self *AMQPConnection) NotifyClose() chan *amqp.Error
• type Heartbeat
• type Publication
• type Request
• type Response

Variables

var (
	// EXCHANGE is the default exchange
	EXCHANGE = "h2o"
	// REPLY_EXCHANGE is the exchange where replies go to
	REPLY_EXCHANGE = "h2o.direct"
	// TOPIC_EXCHANGE is the topic exchange
	TOPIC_EXCHANGE = "h2o.topic"
	// HOSTNAME is the hostname of rabbitmq
	HOSTNAME = "localhost"
	// USERNAME is the username to connect to rabbitmq with
	USERNAME = "rname"
	// PASSWORD is the password to connect to rabbitmq with
	PASSWORD = "rpass"
	// PORT is the rabbitmq port
	PORT = flag.Int("amqp-port", 5672, "<port> to connect to for AMQP")
	// ADMINPORT is the port of the admin interface
	ADMINPORT = flag.Int("amqp-admin-port", 15672, "<port> to connect to for AMQP")
	// AmqpUri is the URI to connect to rabbit, made up of username, host etc
	AmqpUri = flag.String("rabbituri", "", "example: amqp://guest:guest@rabbitmq:5672")
)

var (
	// Connection wraps a standard AMQP connection
	Connection *AMQPConnection
	// Publisher represents the channel we send messages on
	Publisher *AMQPChannel
	// Consumer represents the channel we consume message on
	Consumer  *AMQPChannel
	Connected bool
)

Functions

func BindService

func BindService(serviceName, queue string) error

BindService is used when self-binding

func Connect

func Connect() chan bool

Connect to AMQP + channels

func Consume

func Consume(queue string) (deliveries <-chan amqp.Delivery, err error)

Consume data from a queue

func ConsumerNotifyClose

func ConsumerNotifyClose() chan *amqp.Error

ConsumerNotifyClose returns a go channel to notify when the amqp channel closes

func IsConnected

func IsConnected() bool

IsConnected returns the status of the raven connection Todo: we should probably write a proper conn manager with locking and all.

func SendHeartbeat

func SendHeartbeat(hb Heartbeat, InstanceID string) error

SendHeartbeat via AMQP

func SendPublication

func SendPublication(pub Publication, InstanceID string) error

SendPublication via AMQP

func SendRequest

func SendRequest(req Request, InstanceID string) error

SendRequest via AMQP

func SendResponse

func SendResponse(rsp Response, InstanceID string) error

SendResponse back via AMQP

Types

type AMQPChannel

type AMQPChannel struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

AMQPChannel wraps an AMQP channel with state

func (*AMQPChannel) Close

func (self *AMQPChannel) Close() error

Close allows us to close the channel

func (*AMQPChannel) NotifyClose

func (self *AMQPChannel) NotifyClose() chan *amqp.Error

NotifyClose allows us to listen for the channel closing

type AMQPConnection

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

AMQPConnection wraps an AMQP connection with state

func (*AMQPConnection) Channel

func (self *AMQPConnection) Channel() (*amqp.Channel, error)

Channel makes sure we are connected and gets a new channel

func (*AMQPConnection) Close

func (self *AMQPConnection) Close() error

Close allows us to close the connection

func (*AMQPConnection) NotifyClose

func (self *AMQPConnection) NotifyClose() chan *amqp.Error

NotifyClose allows us to listen for the connection closing

type Heartbeat

type Heartbeat interface {
	ID() string
	ContentType() string
	Payload() []byte
}

Heartbeat interface

type Publication

type Publication interface {
	ContentType() string
	Topic() string
	MessageID() string
	Payload() []byte
	SessionID() string
}

Publication interface

type Request

type Request interface {
	ContentType() string
	Service() string
	Endpoint() string
	MessageID() string
	From() string
	TraceID() string
	TraceShouldPersist() bool
	SessionID() string
	ParentMessageID() string
	Payload() []byte
	Authorised() bool
}

Request interface

type Response

type Response interface {
	ContentType() string
	MessageType() string
	Payload() []byte
	ReplyTo() string
	MessageID() string
}

Response interface