rabbeasy

type Acker ¶

type Acker interface {
	Ack()
}

Acker confirms message with ok

type ClientNotifiable ¶

type ClientNotifiable interface {
	NotifyClose(e *amqp.Error)
}

ClientNotifiable notifies channel status

type CloserConnection ¶

type CloserConnection interface {
	Close() error
}

CloserConnection defines closeable connection interface

type Connection ¶

type Connection struct {
	ConsumerClients            []ConsumerClient
	PublisherClients           []PublisherClient
	ConsumerClientNotifiables  []ConsumerClientNotifiable
	PublisherClientNotifiables []PublisherClientNotifiable
	// contains filtered or unexported fields
}

Connection manages amqp connection

func NewConnection(param ConnectionParameter) (c *Connection, err error)

func (c *Connection) Close() (err error)

func (c *Connection) StartConsumer(consumer ConsumerClient) (err error)

func (c *Connection) StartNotifiableConsumer(consumer ConsumerClientNotifiable) (err error)

func (c *Connection) StartNotifiablePublisher(publisher PublisherClientNotifiable) (err error)

func (c *Connection) StartPublisher(publisher PublisherClient) (err error)

type ConnectionConfig ¶

type ConnectionConfig struct {
	Environment       string
	Domain            string
	Name              string
	Number            int
	Host              string
	Port              int
	User              string
	Password          string
	ReconnectInterval time.Duration
	GlobalConsumerConfig
}

ConnectionConfig holds connection settings

func (c ConnectionConfig) GetConnectionName() (cn string)

func (c ConnectionConfig) GetProperties() (p map[string]interface{})

func (c ConnectionConfig) GetURL() (u string)

type ConnectionParameter ¶

type ConnectionParameter struct {
	Config       ConnectionConfig
	Logger       Logger
	OnConnect    chan interface{}
	OnDisconnect chan *amqp.Error
}

ConnectionParameter holds creation parameters

type Consumer ¶

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

Consumer implements messaging consumer

func NewConsumer(param ConsumerParameter) (c *Consumer, err error)

func (c *Consumer) Destination() (d ConsumerConfig)

func (c *Consumer) From(messages <-chan amqp.Delivery)

type ConsumerClient ¶

type ConsumerClient interface {
	From(<-chan amqp.Delivery)
	Destination() ConsumerConfig
}

ConsumerClient processes channel messages

type ConsumerClientNotifiable ¶

type ConsumerClientNotifiable interface {
	ConsumerClient
	ClientNotifiable
}

ConsumerClientNotifiable processes channel messages

Also receives channel status

type ConsumerConfig ¶

type ConsumerConfig struct {
	DestinationConfig
	PrefetchCount int
	PrefetchSize  int
}

ConsumerConfig holds consumer configs

type ConsumerConnection ¶

type ConsumerConnection interface {
	StartConsumer(ConsumerClient) error
}

ConsumerConnection defines consumer connection interface

type ConsumerParameter ¶

type ConsumerParameter struct {
	Config     ConsumerConfig
	Connection ConsumerConnection
	Handler    Handler
}

ConsumerParameter holds creation parameters

type DeadLetterSender ¶

type DeadLetterSender interface {
	DeadLetter()
}

DeadLetterSender confirms message with sending to discard

type Delivery ¶

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

Delivery implements Message and wraps message delivered

func NewDelivery(d *amqp.Delivery) *Delivery

func (d *Delivery) Ack()

func (d *Delivery) Body() []byte

func (d *Delivery) DeadLetter()

func (d *Delivery) Delivery() *amqp.Delivery

func (d *Delivery) Requeue()

type Deliveryer ¶

type Deliveryer interface {
	Delivery() *amqp.Delivery
}

Deliveryer returns original received message

type DestinationConfig ¶

type DestinationConfig struct {
	// ConsumerConfig
	// PublisherConfig
	Queue           string
	Exchange        string
	ExchangeType    string
	RoutingKey      string
	DlxExchange     string
	DlxExchangeType string
	DlxRoutingKey   string
	Durable         bool
}

DestinationConfig holds exchange destination configs

type GlobalConsumerConfig ¶

type GlobalConsumerConfig struct {
	PrefetchCount int
	PrefetchSize  int
}

GlobalConsumerConfig holds connection global consumer configs

type Handler ¶

type Handler func(d Message)

Handler called when a message is received

Although you can implement business rules directly in the Handler, it is advisable to delegate the treatment to another function or channel.

package main

import (
	"fmt"

	"github.com/kimprado/rabbeasy/pkg/rabbeasy"
)

func main() {
	var h rabbeasy.Handler // Consumer will invoke handler for each message

	var listenerCh = make(chan rabbeasy.MessageBody)
	h = func(m rabbeasy.Message) {
		listenerCh <- m
	}

	rabbeasy.NewConsumer(rabbeasy.ConsumerParameter{
		Handler: h, // Consumer created with handler h reference
	})

	message := <-listenerCh
	fmt.Println(string(message.Body()))
}

Output:

package main

import (
	"fmt"

	"github.com/kimprado/rabbeasy/pkg/rabbeasy"
	"github.com/streadway/amqp"
)

func main() {
	type MySimpleMessageType interface {
		rabbeasy.MessageBody // Declared only what is used ( Body() )
		rabbeasy.Acker       // Declared only what is used ( Ack()  )
		// rabbeasy.Requeuer         // Not declared for non-use
		// rabbeasy.DeadLetterSender // Not declared for non-use
		// rabbeasy.Deliveryer       // Not declared for non-use
	}

	var listenerCh = make(chan MySimpleMessageType)
	var handler = func(m rabbeasy.Message) {
		listenerCh <- m
	}

	go produceMockMassege(handler)

	message := <-listenerCh
	fmt.Println(string(message.Body()))
	message.Ack()

}

func produceMockMassege(h rabbeasy.Handler) {
	h(&mockMessage{})
}

type mockMessage struct{}

func (m *mockMessage) Body() []byte {
	return []byte("hello")
}
func (m *mockMessage) Ack()     {}
func (m *mockMessage) Requeue() {}
func (m *mockMessage) Delivery() *amqp.Delivery {
	return nil
}
func (m *mockMessage) DeadLetter() {}

Output:

hello

type Logger ¶

type Logger interface {
	Errorf(msg string, v ...interface{})
	Warnf(msg string, v ...interface{})
	Infof(msg string, v ...interface{})
	Debugf(msg string, v ...interface{})
	Tracef(msg string, v ...interface{})
}

Logger specifies interface for loggers implementations

type Message ¶

type Message interface {
	MessageBody
	Acker
	Requeuer
	DeadLetterSender
	Deliveryer
}

Message wraps message delivered

type MessageBody ¶

type MessageBody interface {
	Body() []byte
}

MessageBody returns bytes of message

type MessageConsumer ¶

type MessageConsumer interface {
	Consume(Handler)
}

MessageConsumer processes channel messages

type MessagePublisher ¶

type MessagePublisher interface {
	Publish(body []byte) error
}

MessagePublisher publishes messages on the channel

type NotifiableConsumer ¶

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

NotifiableConsumer implements messaging notifiable consumer

func NewNotifiableConsumer(param NotifiableConsumerParameter) (c *NotifiableConsumer, err error)

func (c *NotifiableConsumer) NotifyClose(e *amqp.Error)

type NotifiableConsumerConnection ¶

type NotifiableConsumerConnection interface {
	StartNotifiableConsumer(ConsumerClientNotifiable) error
}

NotifiableConsumerConnection defines consumer connection interface

type NotifiableConsumerParameter ¶

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

NotifiableConsumerParameter holds creation parameters

type NotifiablePublisher ¶

type NotifiablePublisher struct {
	Publisher
	// contains filtered or unexported fields
}

NotifiablePublisher implements messaging notifiable publisher

func NewNotifiablePublisher(param NotifiablePublisherParameter) (p *NotifiablePublisher, err error)

func (p *NotifiablePublisher) NotifyClose(e *amqp.Error)

func (p *NotifiablePublisher) Publish(body []byte) (err error)

package main

import (
	"github.com/kimprado/rabbeasy/pkg/rabbeasy"
	"github.com/streadway/amqp"
)

func main() {
	var (
		publisher *rabbeasy.NotifiablePublisher
		param     rabbeasy.NotifiablePublisherParameter
		conn      rabbeasy.NotifiablePublisherConnection
		cfg       rabbeasy.PublisherConfig
		eCh       chan *amqp.Error
		err       error
	)

	cfg = rabbeasy.PublisherConfig{
		DestinationConfig: rabbeasy.DestinationConfig{
			RoutingKey: "sample", // Sending to queue 'sample'
		},
		Default: amqp.Publishing{
			ContentType: "text/plain",
		},
	}

	param = rabbeasy.NotifiablePublisherParameter{
		Config:     cfg,
		Connection: conn,
		Receiver:   eCh,
	}

	publisher, err = rabbeasy.NewNotifiablePublisher(param)
	if err != nil {
		return
	}

	publisher.Publish([]byte("New message to send"))
}

Output:

type NotifiablePublisherConnection ¶

type NotifiablePublisherConnection interface {
	StartNotifiablePublisher(PublisherClientNotifiable) error
}

NotifiablePublisherConnection defines publisher connection interface

type NotifiablePublisherParameter ¶

type NotifiablePublisherParameter struct {
	Config     PublisherConfig
	Connection NotifiablePublisherConnection
	Receiver   chan *amqp.Error
}

NotifiablePublisherParameter holds creation parameters

type Publisher ¶

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

Publisher implements messaging publisher

func NewPublisher(param PublisherParameter) (p *Publisher, err error)

func (p *Publisher) Destination() (d PublisherConfig)

func (p *Publisher) Publish(body []byte) (err error)

package main

import (
	"github.com/kimprado/rabbeasy/pkg/rabbeasy"
	"github.com/streadway/amqp"
)

func main() {
	var (
		publisher *rabbeasy.Publisher
		param     rabbeasy.PublisherParameter
		conn      rabbeasy.PublisherConnection
		cfg       rabbeasy.PublisherConfig
		err       error
	)

	cfg = rabbeasy.PublisherConfig{
		DestinationConfig: rabbeasy.DestinationConfig{
			RoutingKey: "sample", // Sending to queue 'sample'
		},
		Default: amqp.Publishing{
			ContentType: "text/plain",
		},
	}

	param = rabbeasy.PublisherParameter{
		Config:     cfg,
		Connection: conn,
	}

	publisher, err = rabbeasy.NewPublisher(param)
	if err != nil {
		return
	}

	publisher.Publish([]byte("New message to send"))
}

Output:

func (p *Publisher) PublishIn(amqpCh *amqp.Channel)

type PublisherClient ¶

type PublisherClient interface {
	PublishIn(*amqp.Channel)
	Destination() PublisherConfig
}

PublisherClient publishes messages on the channel

type PublisherClientNotifiable ¶

type PublisherClientNotifiable interface {
	PublisherClient
	ClientNotifiable
}

PublisherClientNotifiable publishes messages on the channel

Also receives channel status

type PublisherConfig ¶

type PublisherConfig struct {
	DestinationConfig
	Default amqp.Publishing
}

PublisherConfig holds message publishing configs

type PublisherConnection ¶

type PublisherConnection interface {
	StartPublisher(PublisherClient) error
}

PublisherConnection defines publisher connection interface

type PublisherParameter ¶

type PublisherParameter struct {
	Config     PublisherConfig
	Connection PublisherConnection
}

PublisherParameter holds creation parameters

type Requeuer ¶

type Requeuer interface {
	Requeue()
}

Requeuer confirms message with requeue