sneaker

README

go-sneaker

A fast background processing framework for Golang and RabbitMQ

Usage

Use go get

go get github.com/grantchen/go-sneaker

Then import the go-sneaker package into your own code.

import "github.com/grantchen/go-sneaker"

Publisher

send background task code sample

Publisher, err := sneaker.NewPublisher(amqp_url, exchange_key))
if err != nil {
  panic(err)
}

var json = jsoniter.ConfigCompatibleWithStandardLibrary
dataJsonByte, _ := json.Marshal(map[string]string{"key": "value"})
err = Publisher.Publish("queue_name", "text/json", dataJsonByte)
if err != nil {
  panic(err)
}

// if not use can close it
Publisher.Close()

Publish Parameters

Publish(queueName, bodyContentType string, body []byte)

queueName - Queue Name
bodyContentType - send body content type. Default is text/json
body - send data body byte

Consumer

background task handle worker sample

Consumer, err := sneaker.NewConsumer(amqp_url, exchange_key))
if err != nil {
  panic(err)
}

err = Consumer.Consume("queue_name", map[string]interface{}{}, handleWorker)
if err != nil {
  panic(err)
}

func handleWorker(body []byte){
  //...
}

Consume Parameters

Consume(queueName string, args map[string]interface{}, f fn)

queueName - Queue Name
args - consume releated params. Deafult values:

     {"durable": true, "autoDelete": false, "autoAck": false,
		 "exclusive": false, "noWait": false,
		 "noLocal": false, "consumer": "", "threads": 5}
    
     durable - is queue durable
     autoDelete - is queue delete when unused
     autoAck - is consume auto ack
     exclusive - is queue and consume exclusive
     noWait - is queue and consume no wait
     noLocal - is consume no local
     consumer - consumer name
     threads - how many threads to handle worker method f

f - handle body method

Documentation

Index

• type Consumer
  • func NewConsumer(amqpUrl, exchangeName string) (*Consumer, error)
  • func (c *Consumer) AnnounceQueue(queueName string, args map[string]interface{}) (<-chan amqp.Delivery, error)
  • func (c *Consumer) AutoRecover()
  • func (c *Consumer) Connect() error
  • func (c *Consumer) Consume(queueName string, args map[string]interface{}, f fn) error
  • func (c *Consumer) ReConnect() error
• type Publisher
  • func NewPublisher(amqpUrl, exchangeName string) (*Publisher, error)
  • func (c *Publisher) Close()
  • func (c *Publisher) Publish(queueName, bodyContentType string, body []byte) error

Types

type Consumer

type Consumer struct {
	Connection   *amqp.Connection
	Channel      *amqp.Channel
	ExchangeName string
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(amqpUrl, exchangeName string) (*Consumer, error)

func (*Consumer) AnnounceQueue

func (c *Consumer) AnnounceQueue(queueName string, args map[string]interface{}) (<-chan amqp.Delivery, error)

AnnounceQueue declares and consume a queue

func (*Consumer) AutoRecover

func (c *Consumer) AutoRecover()

AutoRecover will recover connection and consumers

func (*Consumer) Connect

func (c *Consumer) Connect() error

Connect server

func (*Consumer) Consume

func (c *Consumer) Consume(queueName string, args map[string]interface{}, f fn) error

Consume consume a worker queue

func (*Consumer) ReConnect

func (c *Consumer) ReConnect() error

ReConnect server and re-consume all queues

type Publisher

type Publisher struct {
	Connection   *amqp.Connection
	Channel      *amqp.Channel
	ExchangeName string
}

func NewPublisher

func NewPublisher(amqpUrl, exchangeName string) (*Publisher, error)

func (*Publisher) Close

func (c *Publisher) Close()

func (*Publisher) Publish

func (c *Publisher) Publish(queueName, bodyContentType string, body []byte) error

publish a worker queue