saga

README

saga

saga is a Go library designed to streamline communication between microservices using RabbitMQ. It enables easy implementation of event-driven architectures and saga patterns, while ensuring reliable message delivery.

Features

Core Communication:

• Publish/Subscribe Messaging: Exchange messages between microservices using a publish-subscribe pattern.
• Headers-Based Routing: Leverage the power of RabbitMQ's headers exchange for flexible and dynamic routing of messages based on custom headers.
• Durable Exchanges and Queues: Ensure message persistence and reliability with durable RabbitMQ components.

Saga Management:

• Saga Orchestration: Coordinate complex, multi-step transactions across multiple microservices with saga orchestration.
• Saga Step Handlers: Implement step-by-step saga logic in your microservices using callbacks.
• Compensation Logic: Define compensating actions for saga steps to handle failures gracefully and maintain data consistency.

Contributors

Thanks to all contributors!

Author

Jorge Clavijo https://github.com/jym272

License

Distributed under the MIT License. See LICENSE for more information.

Documentation

Index

• Constants
• Variables
• func CommenceSaga(payload CommencePayload) error
• func ParsePayload[T any](handlerPayload map[string]interface{}, data *T) *T
• func PublishEvent(payload event.PayloadEvent) error
• type CommandHandler
• type CommencePayload
• type CompletedCryptoRanking
• type ConsumeChannel
  • func (c *ConsumeChannel) NackWithDelay(delay time.Duration, maxRetries int32) (int32, time.Duration, error)
  • func (c *ConsumeChannel) NackWithFibonacciStrategy(maxOccurrence, maxRetries int32) (int32, time.Duration, int32, error)
• type CryptoRankingWinners
• type Emitter
  • func (e *Emitter[T, U]) Emit(event U, data T)
  • func (e *Emitter[T, U]) On(event U, handler func(T))
• type EventHandler
  • func (e *EventHandler) ParseEventPayload(data any)
• type EventsConsumeChannel
  • func (m *EventsConsumeChannel) AckMessage()
• type Exchange
• type MicroserviceConsumeChannel
  • func (m *MicroserviceConsumeChannel) AckMessage(payloadForNextStep NextStepPayload)
• type NextStepPayload
• type Occurrence
• type Opts
• type PurchaseResourceFlowPayload
  • func (PurchaseResourceFlowPayload) Type() SagaTitle
• type Queue
• type QueueConsumerProps
• type RankingsUsersRewardPayload
  • func (RankingsUsersRewardPayload) Type() SagaTitle
• type SagaStep
• type SagaTitle
• type Status
• type StepHashId
• type Transactional
  • func Config(opts *Opts) *Transactional
  • func (t *Transactional) ConnectToEvents() *Emitter[EventHandler, event.MicroserviceEvent]
  • func (t *Transactional) ConnectToSagaCommandEmitter() *Emitter[CommandHandler, micro.StepCommand]
  • func (t *Transactional) HealthCheck() error
  • func (t *Transactional) StopRabbitMQ() error
• type TransferCryptoRewardToRankingWinnersPayload
  • func (TransferCryptoRewardToRankingWinnersPayload) Type() SagaTitle
• type UserReward

Constants

const (
	NACKING_DELAY_MS = 5000 // 5 seconds
	MAX_NACK_RETRIES = 3
	// MAX_OCCURRENCE
	/**
	 * Define the maximum occurrence in a fail saga step of the nack delay with fibonacci strategy
	 * | Occurrence | Delay in the next nack |
	 * |------------|------------------------|
	 * | 17         | 0.44 hours  |
	 * | 18         | 0.72 hours  |
	 * | 19         | 1.18 hours  |
	 * | 20         | 1.88 hours  |
	 * | 21         | 3.04 hours  |
	 * | 22         | 4.92 hours  |
	 * | 23         | 7.96 hours  |
	 * | 24         | 12.87 hours |
	 * | 25         | 20.84 hours |
	 */
	MAX_OCCURRENCE = 19
)

Variables

var RabbitUri string

RabbitUri is used for send channel connection.

Functions

func CommenceSaga

func CommenceSaga(payload CommencePayload) error

func ParsePayload

func ParsePayload[T any](handlerPayload map[string]interface{}, data *T) *T

func PublishEvent

func PublishEvent(payload event.PayloadEvent) error

Types

type CommandHandler

type CommandHandler struct {
	Channel *MicroserviceConsumeChannel `json:"channel"`
	Payload map[string]interface{}      `json:"payload"`
	SagaID  int                         `json:"sagaId"`
}

type CommencePayload

type CommencePayload interface {
	Type() SagaTitle
}

type CompletedCryptoRanking

type CompletedCryptoRanking struct {
	WalletAddress string                 `json:"walletAddress"`
	Winners       []CryptoRankingWinners `json:"winners"`
}

type ConsumeChannel

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

func (*ConsumeChannel) NackWithDelay

func (c *ConsumeChannel) NackWithDelay(delay time.Duration, maxRetries int32) (int32, time.Duration, error)

math.MaxInt8.

func (*ConsumeChannel) NackWithFibonacciStrategy

func (c *ConsumeChannel) NackWithFibonacciStrategy(maxOccurrence, maxRetries int32) (int32, time.Duration, int32, error)

NackWithFibonacciStrategy is a function that handles the nack of a message with a delay that increases with the fibonacci sequence. The delay is calculated as the fibonacci sequence of the occurrence of the message. The occurrence is the number of times the message has been nacked. The function returns the number of retries, the delay and the occurrence of the message.

type CryptoRankingWinners

type CryptoRankingWinners struct {
	UserID string `json:"userId"`
	Reward string `json:"reward"`
}

type Emitter

type Emitter[T any, U comparable] struct {
	// contains filtered or unexported fields
}

func (*Emitter[T, U]) Emit

func (e *Emitter[T, U]) Emit(event U, data T)

func (*Emitter[T, U]) On

func (e *Emitter[T, U]) On(event U, handler func(T))

type EventHandler

type EventHandler struct {
	Channel *EventsConsumeChannel  `json:"channel"`
	Payload map[string]interface{} `json:"payload"`
}

func (*EventHandler) ParseEventPayload

func (e *EventHandler) ParseEventPayload(data any)

ParseEventPayload also works, but you need to pass a reference to the variable and is not type safe to assure that, as the type is: any Works: var eventPayload1 saga.SocialNewUserPayload // or a pointer *saga.SocialNewUserPayload ------------------------->key, pass the reference<-----------------// handler.ParsePayload(&eventPayload1)

It does not work: handler.ParsePayload(eventPayload1).

type EventsConsumeChannel

type EventsConsumeChannel struct {
	*ConsumeChannel
}

func (*EventsConsumeChannel) AckMessage

func (m *EventsConsumeChannel) AckMessage()

type Exchange

type Exchange string

const (
	RequeueExchange         Exchange = "requeue_exchange"
	CommandsExchange        Exchange = "commands_exchange"
	MatchingExchange        Exchange = "matching_exchange"
	MatchingRequeueExchange Exchange = "matching_requeue_exchange"
)

type MicroserviceConsumeChannel

type MicroserviceConsumeChannel struct {
	*ConsumeChannel
	// contains filtered or unexported fields
}

func (*MicroserviceConsumeChannel) AckMessage

func (m *MicroserviceConsumeChannel) AckMessage(payloadForNextStep NextStepPayload)

type NextStepPayload

type NextStepPayload = map[string]interface{}

type Occurrence

type Occurrence int

type Opts

type Opts struct {
	RabbitUri    string                       `validate:"required,url"`
	Microservice micro.AvailableMicroservices `validate:"required,microservice"`
	Events       []event.MicroserviceEvent    `validate:"-"`
}

type PurchaseResourceFlowPayload

type PurchaseResourceFlowPayload struct {
	UserId     string `json:"userId"`
	ResourceId string `json:"resourceId"`
	Price      int    `json:"price"`
	Quantity   int    `json:"quantity"`
}

PurchaseResourceFlowPayload is the payload for the purchase_resource_flow event.

func (PurchaseResourceFlowPayload) Type

func (PurchaseResourceFlowPayload) Type() SagaTitle

type Queue

type Queue string

const (
	CommenceSagaQueue Queue = "commence_saga"
)

const (
	ReplyToSagaQ Queue = "reply_to_saga"
)

type QueueConsumerProps

type QueueConsumerProps struct {
	QueueName string
	Exchange  Exchange
}

type RankingsUsersRewardPayload

type RankingsUsersRewardPayload struct {
	Rewards []UserReward `json:"rewards"`
}

RankingsUsersRewardPayload is the payload for the rankings_users_reward event.

func (RankingsUsersRewardPayload) Type

func (RankingsUsersRewardPayload) Type() SagaTitle

type SagaStep

type SagaStep struct {
	Microservice    micro.AvailableMicroservices `json:"microservice"`
	Command         string                       `json:"command"`
	Status          Status                       `json:"status"`
	SagaID          int                          `json:"sagaId"`
	Payload         map[string]interface{}       `json:"payload"`
	PreviousPayload map[string]interface{}       `json:"previousPayload"`
	IsCurrentStep   bool                         `json:"isCurrentStep"`
}

type SagaTitle

type SagaTitle string

const (
	PurchaseResourceFlow                 SagaTitle = "purchase_resource_flow"
	RankingsUsersReward                  SagaTitle = "rankings_users_reward"
	TransferCryptoRewardToRankingWinners SagaTitle = "transfer_crypto_reward_to_ranking_winners"
)

type Status

type Status string

const (
	Success Status = "success"
	Failure Status = "failure"
	Sent    Status = "sent"
	Pending Status = "pending"
)

type StepHashId

type StepHashId string

type Transactional

type Transactional struct {
	Microservice micro.AvailableMicroservices
	Events       []event.MicroserviceEvent
	// contains filtered or unexported fields
}

func Config

func Config(opts *Opts) *Transactional

func (*Transactional) ConnectToEvents

func (t *Transactional) ConnectToEvents() *Emitter[EventHandler, event.MicroserviceEvent]

ConnectToEvents connects to the events exchange and returns an emitter.

func (*Transactional) ConnectToSagaCommandEmitter

func (t *Transactional) ConnectToSagaCommandEmitter() *Emitter[CommandHandler, micro.StepCommand]

ConnectToSagaCommandEmitter connects to the saga commands exchange and returns an emitter.

func (*Transactional) HealthCheck

func (t *Transactional) HealthCheck() error

HealthCheck checks if the rabbitmq connection is alive and the queue exists. The queue to check is the microservice related to the saga commands or events.

func (*Transactional) StopRabbitMQ

func (t *Transactional) StopRabbitMQ() error

StopRabbitMQ closes the rabbitmq connection and channels.

type TransferCryptoRewardToRankingWinnersPayload

type TransferCryptoRewardToRankingWinnersPayload struct {
	CompletedCryptoRankings []CompletedCryptoRanking `json:"completedCryptoRankings"`
}

TransferCryptoRewardToRankingWinnersPayload is the payload for the transfer_crypto_reward_to_ranking_winners event.

func (TransferCryptoRewardToRankingWinnersPayload) Type

func (TransferCryptoRewardToRankingWinnersPayload) Type() SagaTitle

type UserReward

type UserReward struct {
	UserId string `json:"userId"`
	Reward string `json:"coins"`
}