rabbit

Documentation

Index

• Variables
• func ConnectToRabbitMQ() error
• type ConsumeMessage
  • func (message ConsumeMessage) GetPayload() (map[string]interface{}, error)
• type EventProcessor
  • func (eProcessor EventProcessor) ApplyEventProcessor() (*ConsumeMessage, error)
  • func (eProcessor EventProcessor) ApplyEvents(aggregateArticleID string) error
• type RabbitRepository
• type RabbitRepositoryImpl
  • func NewRabbitRepository(channel *amqp.Channel, queue amqp.Queue) *RabbitRepositoryImpl
  • func (rabbit *RabbitRepositoryImpl) Consume(eventBus *event.EventBus)
  • func (rabbit *RabbitRepositoryImpl) Publish(event string) error
  • func (rabbit *RabbitRepositoryImpl) QueueConnector(event interface{}) error

Variables

var RabbitChannel *amqp.Channel

var RabbitQueue amqp.Queue

Functions

func ConnectToRabbitMQ

func ConnectToRabbitMQ() error

ConnectToRabbitMQ is for connecting to rabbitmq

Types

type ConsumeMessage

type ConsumeMessage struct {
	EventType string      `json:"event_type"`
	Payload   interface{} `json:"payload"`
}

ConsumeMessage Send message to rabbitMQ

func (ConsumeMessage) GetPayload

func (message ConsumeMessage) GetPayload() (map[string]interface{}, error)

GetPayload Map payload as map[string]interface{} type

type EventProcessor

type EventProcessor struct {
	Message   []byte
	EventBus  *event.EventBus
	StoreInDB bool
}

EventProcessor Stores Body (events.ConsumeMessage type) and EventBus

func (EventProcessor) ApplyEventProcessor

func (eProcessor EventProcessor) ApplyEventProcessor() (*ConsumeMessage, error)

ApplyEventProcessor Unmarshal body and send it to EventProcessor

func (EventProcessor) ApplyEvents

func (eProcessor EventProcessor) ApplyEvents(aggregateArticleID string) error

ApplyEvents Unmarshal body and send it to EventProcessor

type RabbitRepository

type RabbitRepository interface {
	QueueConnector(interface{}) error
	Publish(string) error
	Consume(*event.EventBus)
}

type RabbitRepositoryImpl

type RabbitRepositoryImpl struct {
	Chan  *amqp.Channel
	Queue amqp.Queue
}

func NewRabbitRepository

func NewRabbitRepository(channel *amqp.Channel, queue amqp.Queue) *RabbitRepositoryImpl

func (*RabbitRepositoryImpl) Consume

func (rabbit *RabbitRepositoryImpl) Consume(eventBus *event.EventBus)

Consume Receives event and dispatch it to event handler

func (*RabbitRepositoryImpl) Publish

func (rabbit *RabbitRepositoryImpl) Publish(event string) error

Publish new event on rabbitmq

func (*RabbitRepositoryImpl) QueueConnector

func (rabbit *RabbitRepositoryImpl) QueueConnector(event interface{}) error

QueueConnector Serialize event and send to rabbitMQ queue