wabbit

README

Wabbit - Go AMQP Mocking Library

Elmer Fudd: Shhh. Be vewy vewy quiet, I'm hunting wabbits

AMQP is a verbose protocol that makes it difficult to implement proper unit-testing on your application. The first goal of this package is provide a sane interface for an AMQP client implementation based on the specification AMQP-0-9-1 (no extension) and then an implementation of this interface using the well established package streadway/amqp (a wrapper).

What are the advantages of this?

Usage

How to use ?

Testing

This package have an AMQP interface and two possible implementations:

• wabbit/amqp - Bypass to streadway/amqp
• wabbit/amqptest

In the same way you can use the http package in your software and use the httptest for testing, when using wabbit is recommended to use the wabbit/amqp package on your software and wabbit/amqptest in your tests. Simple test example:

package main

import (
	"testing"
	"github.com/PeriscopeData/wabbit/amqptest"
	"github.com/PeriscopeData/wabbit/amqptest/server"
	"github.com/PeriscopeData/wabbit/amqp"
)


func TestChannelCreation(t *testing.T) {
	mockConn, err := amqptest.Dial("amqp://localhost:5672/%2f") // will fail,

	if err == nil {
		t.Error("This shall fail, because no fake amqp server is running...")
	}

	fakeServer := server.NewServer("amqp://localhost:5672/%2f")
	fakeServer.Start()

	mockConn, err = amqptest.Dial("amqp://localhost:5672/%2f") // now it works =D

	if err != nil {
		t.Error(err)
	}

	//Now you can use mockConn as a real amqp connection.
	channel, err := mockConn.Channel()

    // ...
}

The package amqptest/server implements a mock AMQP server and it can be used to simulate network partitions or broker crashs. To create a new server instance use server.NewServer passing any amqpuri. You can create more than one server, but they need to have different amqpuris. Example below:

    broker1 := server.NewServer("amqp://localhost:5672/%2f")
    broker2 := server.NewServer("amqp://192.168.10.165:5672/%2f")
    broker3 := server.NewServer("amqp://192.168.10.169:5672/%2f")

    broker1.Start()
    broker2.Start()
    broker3.Start()

Calling NewServer with same amqpuri will return the same server instance.

Use broker.Stop() to abruptly stop the amqp server.

There's no fake clustering support yet (maybe never)

It's a very straightforward implementation that need a lot of improvements yet. Take careful when using it.

[]'s

Documentation

Overview

Package wabbit provides an interface for AMQP client specification and a mock implementation of that interface.

Index

• type Channel
• type Confirmation
• type Conn
• type Delivery
• type Error
• type Option
• type Publisher
• type Queue

Types

type Channel

type Channel interface {
	Ack(tag uint64, multiple bool) error
	Nack(tag uint64, multiple bool, requeue bool) error
	Reject(tag uint64, requeue bool) error

	Confirm(noWait bool) error
	NotifyPublish(confirm chan Confirmation) chan Confirmation

	Cancel(consumer string, noWait bool) error
	ExchangeDeclare(name, kind string, opt Option) error
	ExchangeDeclarePassive(name, kind string, opt Option) error
	QueueDeclare(name string, args Option) (Queue, error)
	QueueDeclarePassive(name string, args Option) (Queue, error)
	QueueDelete(name string, args Option) (int, error)
	QueueBind(name, key, exchange string, opt Option) error
	QueueUnbind(name, route, exchange string, args Option) error
	Consume(queue, consumer string, opt Option) (<-chan Delivery, error)
	Qos(prefetchCount, prefetchSize int, global bool) error
	Close() error
	NotifyClose(chan Error) chan Error
	Publisher
}

Channel is an AMQP channel interface

type Confirmation

type Confirmation interface {
	Ack() bool
	DeliveryTag() uint64
}

Confirmation is an interface to confrimation messages

type Conn

type Conn interface {
	Channel() (Channel, error)
	AutoRedial(errChan chan Error, done chan bool)
	Close() error
	NotifyClose(chan Error) chan Error
}

Conn is the amqp connection interface

type Delivery

type Delivery interface {
	Ack(multiple bool) error
	Nack(multiple, request bool) error
	Reject(requeue bool) error

	Body() []byte
	Headers() Option
	DeliveryTag() uint64
	ConsumerTag() string
	CorrelationId() string
	MessageId() string
	ReplyTo() string
	Redelivered() bool
}

Delivery is an interface to delivered messages

type Error

type Error interface {
	Code() int
	Reason() string
	Server() bool
	Recover() bool
	error
}

Error is an interface for AMQP errors

type Option

type Option map[string]interface{}

Option is a map of AMQP configurations

type Publisher

type Publisher interface {
	Publish(exc, route string, msg []byte, opt Option) error
}

Publisher is an interface to something able to publish messages

type Queue

type Queue interface {
	Name() string
	Messages() int
	Consumers() int
}

Queue is a AMQP queue interface