pikaq

README

pikaq

PikaQ, I choose YOU!

PikaQ makes working with Rabbit MQ even easier. It provides a thin layer around streadway/amqp, the defacto standard golang library for publishing and consuming Rabbit MQ messages. This package makes writing your own high performance message-based services fun and type safe.

The name "pika" is used for any member of the Ochotonidae, a family within the order of lagomorphs, which also includes the Leporidae (rabbits and hares). -- Wikipedia

---

Install

go get https://github.com/xentek/pikaq

Import

import "github.com/xentek/pikaq"

Usage

The simplest consumer you can create just logs and acks each message it recieves. The example below uses the built in LoggingHandler.

For your consumer, write a function with the following signature: func(pikaq.Messages, chan error), and pass it as the last argument of pikaq.LoggingHandler. Refer to the examples for more information.

package main

import (
	"log"

	"github.com/xentek/pikaq"
)

func main() {
	c, err := pikaq.NewConsumer("amqp://localhost:5672", "amq.direct", "direct", "example-queue", "routing-key", "example", pikaq.LoggingHandler)
	if err != nil {
		log.Fatalf("New Consumer Error: %s", err)
	}
	log.Printf("Started Consumer: %s", c.tag.Tag())
}

Contributing

Refer to our Contributor's Guide to learn how you can participate in this project.

More Info

• GoDoc
• Wiki
• Examples

Documentation

Index

• Variables
• func DeclareQueue(session *Session, queue string) (*Session, Queue, error)
• type Consumer
  • func NewConsumer(url string, exchange string, exchangeType string, queue string, key string, ...) (*Consumer, error)
  • func (c *Consumer) Info() string
  • func (c *Consumer) Start(prefetch int) (Messages, error)
  • func (c *Consumer) Stop() error
• type ConsumerTag
  • func NewConsumerTag(name string) *ConsumerTag
  • func (c ConsumerTag) Tag() string
• type Message
• type MessageHandler
• type Messages
• type Queue
• type Session
  • func BindQueue(session *Session, queue Queue, key string, exchange string) (*Session, error)
  • func DeclareExchange(session *Session, exchange string, exchangeType string) (*Session, error)
  • func NewSession(url string) (*Session, error)

Variables

var LogLevel = log.InfoLevel

Functions

func DeclareQueue

func DeclareQueue(session *Session, queue string) (*Session, Queue, error)

Types

type Consumer

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

func NewConsumer

func NewConsumer(url string, exchange string, exchangeType string, queue string, key string, name string, prefetch int, handler MessageHandler) (*Consumer, error)

func (*Consumer) Info

func (c *Consumer) Info() string

func (*Consumer) Start

func (c *Consumer) Start(prefetch int) (Messages, error)

func (*Consumer) Stop

func (c *Consumer) Stop() error

type ConsumerTag

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

func NewConsumerTag

func NewConsumerTag(name string) *ConsumerTag

func (ConsumerTag) Tag

func (c ConsumerTag) Tag() string

type Message

type Message amqp.Delivery

type MessageHandler

type MessageHandler func(Messages, chan error)

var MessageLogger MessageHandler = func(msgs Messages, done chan error) {
	for m := range msgs {
		log.Printf(
			"got %dB delivery: [%v] %q",
			len(m.Body),
			m.DeliveryTag,
			m.Body,
		)
		m.Ack(false)
	}
	log.Printf("handle: msgs chan closed")
	done <- nil
}

type Messages

type Messages <-chan amqp.Delivery

type Queue

type Queue amqp.Queue

type Session

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

func BindQueue

func BindQueue(session *Session, queue Queue, key string, exchange string) (*Session, error)

func DeclareExchange

func DeclareExchange(session *Session, exchange string, exchangeType string) (*Session, error)

func NewSession

func NewSession(url string) (*Session, error)