rabbitmq

README

go-rabbitmq

An slightly opiniated abstraction for using RabbitMQ services from various Cloud environments. Only tested on Cloudfoundry at this time.

usage

consumer

producer

integration test

docker-compose -f docker/docker-compose.test.yml up --build --abort-on-container-exit

author

Andy Lo-A-Foe andy.loafoe@aemian.com

license

MIT

Documentation

Overview

Copyright 2018 Andy Lo-A-Foe. All rights reserved. Use of this source code is governed by Apache-style license that can be found in the LICENSE file.

Index

• type AMQPConsumer
  • func NewConsumer(config Config) (*AMQPConsumer, error)
  • func (c *AMQPConsumer) Connect() error
  • func (c *AMQPConsumer) Handle(d <-chan amqp.Delivery, fn ConsumerHandlerFunc, threads int, queue string, ...)
  • func (c *AMQPConsumer) Start() error
• type AMQPProducer
  • func NewProducer(config Config) (*AMQPProducer, error)
  • func (p *AMQPProducer) Close()
  • func (p *AMQPProducer) Publish(exchange, routingKey string, msg amqp.Publishing) error
• type Config
• type Consumer
• type ConsumerHandlerFunc
• type Producer
• type Qos

Types

type AMQPConsumer

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

func NewConsumer

func NewConsumer(config Config) (*AMQPConsumer, error)

func (*AMQPConsumer) Connect

func (c *AMQPConsumer) Connect() error

Connect to RabbitMQ server

func (*AMQPConsumer) Handle

func (c *AMQPConsumer) Handle(
	d <-chan amqp.Delivery,
	fn ConsumerHandlerFunc,
	threads int,
	queue string,
	routingKey string)

Handle has all the logic to make sure your program keeps running d should be a delievey channel as created when you call announceQueue fn should be a function that handles the processing of deliveries this should be the last thing called in main as code under it will become unreachable unless put int a goroutine. The q and rk params are redundent but allow you to have multiple queue listeners in main without them you would be tied into only using one queue per connection

func (*AMQPConsumer) Start

func (c *AMQPConsumer) Start() error

type AMQPProducer

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

func NewProducer

func NewProducer(config Config) (*AMQPProducer, error)

func (*AMQPProducer) Close

func (p *AMQPProducer) Close()

func (*AMQPProducer) Publish

func (p *AMQPProducer) Publish(exchange, routingKey string, msg amqp.Publishing) error

type Config

type Config struct {
	ServiceName  string
	Exchange     string
	ExchangeType string
	Durable      bool
	AutoDelete   bool
	QueueName    string
	RoutingKey   string
	CTag         string
	Qos          *Qos
	HandlerFunc  ConsumerHandlerFunc
}

type Consumer

type Consumer interface {
	Connect() error
	Start() error
	Handle(d <-chan amqp.Delivery, f ConsumerHandlerFunc, threads int, queue string, routingKey string)
}

type ConsumerHandlerFunc

type ConsumerHandlerFunc func(deliveries <-chan amqp.Delivery, done <-chan bool)

type Producer

type Producer interface {
	Publish(exchange, routingKey string, msg amqp.Publishing) error
	Close()
}

type Qos

type Qos struct {
	PrefetchCount int
	PrefetchSize  int
	Global        bool
}