broker

Documentation

Index

• Variables
• func AmqpConnect(resourceName string) *amqp.Connection
• func NatsConnect(resourceName string) *nats.Conn
• type AmqpConf
• type AmqpConsumerConf
• type AmqpConsumerOption
  • func GetConsumerResource(resourceName string) AmqpConsumerOption
• type AmqpOption
  • func GetAmqpResource(resourceName string) AmqpOption
• type Consumer
  • func NewConsumer(consumerTag, uri, exchange, exchangeType, bindingKey string, ...) *Consumer
  • func (c *Consumer) AnnounceQueue(queueName, bindingKey string) (<-chan amqp.Delivery, error)
  • func (c *Consumer) Connect() error
  • func (c *Consumer) Handle(d <-chan amqp.Delivery, fn func(<-chan amqp.Delivery), threads int, ...)
  • func (c *Consumer) ReConnect(queueName, bindingKey string) (<-chan amqp.Delivery, error)
• type NatsConf
• type NatsOption
  • func GetNatsResource(resourceName string) NatsOption

Variables

var (
	// Amqp type
	Amqp AmqpConf
	// AmqpConsumer type
	AmqpConsumer AmqpConsumerConf
)

var AmqpConn = make(map[string]*amqp.Connection)

AmqpConn variable

var NatsConn = make(map[string]*nats.Conn)

NatsConn variable

Functions

func AmqpConnect

func AmqpConnect(resourceName string) *amqp.Connection

AmqpConnect function

func NatsConnect

func NatsConnect(resourceName string) *nats.Conn

NatsConnect function

Types

type AmqpConf

type AmqpConf struct {
	AmqpOption []AmqpOption `json:"rabitmqResources" bson:"rabitmqResources"`
}

AmqpConf struct

type AmqpConsumerConf

type AmqpConsumerConf struct {
	AmqpConsumerOption []AmqpConsumerOption `json:"amqpConsumerResources" bson:"amqpConsumerResources"`
}

AmqpConsumerConf struct

type AmqpConsumerOption

type AmqpConsumerOption struct {
	Name         string `json:"name" bson:"name"`
	ConnString   string `json:"connString" bson:"connString"`
	HttpResource string `json:"httpResource" bson:"httpResource"`
	Threads      int    `json:"threads" bson:"threads"`
	Exchange     string `json:"exchange" bson:"exchange"`
	ExchangeType string `json:"exchangeType" bson:"exchangeType"`
	Queue        string `json:"queue" bson:"queue"`
	RoutingKey   string `json:"routingKey" bson:"routingKey"`
	WorkerName   string `json:"workerName" bson:"workerName"`
	Verbosity    bool   `json:"verbosity" bson:"verbosity"`
	ReConnect    int    `json:"reConnectSecInterval" bson:"reConnectSecInterval"`
}

AmqpConsumerOption struct

func GetConsumerResource

func GetConsumerResource(resourceName string) AmqpConsumerOption

GetConsumerResource function

type AmqpOption

type AmqpOption struct {
	Name       string `json:"name" bson:"name"`
	ConnString string `json:"connString" bson:"connString"`
}

AmqpOption struct

func GetAmqpResource

func GetAmqpResource(resourceName string) AmqpOption

GetAmqpResource function

type Consumer

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

Consumer holds all information about the RabbitMQ connection This setup does limit a consumer to one exchange. This should not be an issue. Having to connect to multiple exchanges means something else is structured improperly.

func NewConsumer

func NewConsumer(
	consumerTag,
	uri,
	exchange,
	exchangeType,
	bindingKey string,
	autoReconnectSec int) *Consumer

NewConsumer returns a Consumer struct that has been initialized properly essentially don't touch conn, channel, or done and you can create Consumer manually

func (*Consumer) AnnounceQueue

func (c *Consumer) AnnounceQueue(queueName, bindingKey string) (<-chan amqp.Delivery, error)

AnnounceQueue sets the queue that will be listened to for this connection...

func (*Consumer) Connect

func (c *Consumer) Connect() error

Connect to RabbitMQ server

func (*Consumer) Handle

func (c *Consumer) Handle(
	d <-chan amqp.Delivery,
	fn func(<-chan amqp.Delivery),
	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 redundant but allow you to have multiple queue listeners in main without them you would be tied into only using one queue per connection

func (*Consumer) ReConnect

func (c *Consumer) ReConnect(queueName, bindingKey string) (<-chan amqp.Delivery, error)

ReConnect is called in places where NotifyClose() channel is called wait 30 seconds before trying to reconnect. Any shorter amount of time will likely destroy the error log while waiting for servers to come back online. This requires two parameters which is just to satisfy the AccounceQueue call and allows greater flexability

type NatsConf

type NatsConf struct {
	NatsOption []NatsOption `json:"natsResources" bson:"natsResources"`
}

NatsConf is config name of nats resources

var (
	// Nats variable
	Nats NatsConf
)

type NatsOption

type NatsOption struct {
	Name              string `json:"name" bson:"name"`
	ConnString        string `json:"connString" bson:"connString"`
	MaxReconnectSec   int    `json:"maxReconnectSec" bson:"maxReconnectSec"`
	ReconnectDelaySec int    `json:"reconnectDelaySec" bson:"reconnectDelaySec"`
}

NatsOption struct

func GetNatsResource

func GetNatsResource(resourceName string) NatsOption

GetNatsResource function