pubsub

Documentation

Overview

The PubSub interface is a programming construct in Go that provides a way for components in a distributed system to communicate with each other asynchronously using a publish-subscribe messaging pattern. This interface defines a set of methods that allow a client to publish messages to a topic and subscribe to receive messages from a topic. The PubSub interface is useful in scenarios where multiple components need to be able to send and receive messages in a loosely-coupled manner, without having to know the details of each other's implementation.

To use the PubSub interface, a client would first create an instance of an implementation of the interface that is specific to the messaging system being used, such as NATS or Redis. The client would then use this instance to publish messages to topics and subscribe to receive messages from topics. When a message is published to a topic, all clients that are subscribed to that topic will receive the message. This allows components to communicate with each other without having to know the specific details of who they are communicating with.

The PubSub interface has a general purpose of enabling communication and coordination between components in a distributed system. It provides a flexible and scalable way for components to send and receive messages asynchronously, which can be useful in a variety of scenarios, such as event-driven architectures, microservices, and real-time applications. The interface also allows for easy integration with different messaging systems, making it possible to switch between messaging systems without having to change the code that uses the interface.

Index

• Constants
• type IPubSub
• type Mock
  • func (m *Mock) Close() error
  • func (m *Mock) GetClient() any
  • func (m *Mock) GetLogger() sypl.ISypl
  • func (m *Mock) GetName() string
  • func (m *Mock) Publish(topic string, message any) error
  • func (m *Mock) Subscribe(topic, queue string, cb func([]byte)) Subscription
  • func (m *Mock) Unsubscribe(topic string) error
• type PubSub
  • func New(name string) (*PubSub, error)
  • func (p *PubSub) ErrorHandler(ctx context.Context, operation string, err error) error
  • func (p *PubSub) FailedToErrorHandler(ctx context.Context, operation string, err error) error
  • func (p *PubSub) FailedToErrorMsgHandler(ctx context.Context, msg string, err error) error
  • func (p *PubSub) GetLogger() sypl.ISypl
  • func (p *PubSub) GetName() string
  • func (p *PubSub) GetPublishCounter() *expvar.Int
  • func (p *PubSub) GetSubscribeCounter() *expvar.Int
  • func (p *PubSub) GetType() string
  • func (p *PubSub) NotFoundErrorHandler(ctx context.Context) error
• type Subscription
  • func (s *Subscription) HandleMessage(msg *natsgo.Msg)

Constants

const (
	DefaultMetricCounterLabel = "counter"
	Type                      = "pubsub"

	// Operation name.
	OperationPublish   = "publish"
	OperationSubscribe = "subscribe"
)

Type is the type of the entity regarding the framework. It is used to for example, to identify the entity in the logs, metrics, and for tracing.

Types

type IPubSub

type IPubSub interface {
	// Publish sends a message to a topic.
	Publish(topic string, message any) error

	// Subscribe subscribes to a topic and returns a channel for receiving messages.
	Subscribe(topic, queue string, cb func([]byte)) (Subscription, error)

	// Unsubscribe unsubscribes from a topic.
	Unsubscribe(topic string) error

	// Close closes the connection to the Pub Sub broker.
	Close() error

	// GetClient returns the storage client. Use that to interact with the
	// underlying storage client.
	GetClient() any

	// GetLogger returns the logger.
	GetLogger() sypl.ISypl

	// GetName returns the pubsub name.
	GetName() string
}

IPubSub defines a PubSub does.

type Mock

type Mock struct {
	// Publish sends a message to a topic.
	MockPublish func(topic string, message any) error

	// Subscribe subscribes to a topic and returns a channel for receiving messages.
	MockSubscribe func(topic, queue string, cb func([]byte)) Subscription

	// Unsubscribe unsubscribes from a topic.
	MockUnsubscribe func(topic string) error

	// Close closes the connection to the Pub Sub broker.
	MockClose func() error

	// GetClient returns the storage client. Use that to interact with the
	// underlying storage client.
	MockGetClient func() any

	// GetLogger returns the logger.
	MockGetLogger func() sypl.ISypl

	// GetName returns the storage name.
	MockGetName func() string
}

Mock is a struct which satisfies the pubsub.IPubSub interface.

func (*Mock) Close

func (m *Mock) Close() error

Close mocked call.

func (*Mock) GetClient

func (m *Mock) GetClient() any

GetClient mocked call.

func (*Mock) GetLogger

func (m *Mock) GetLogger() sypl.ISypl

GetLogger returns the logger.

func (*Mock) GetName

func (m *Mock) GetName() string

GetName returns the storage name.

func (*Mock) Publish

func (m *Mock) Publish(topic string, message any) error

Publish mocked call.

func (*Mock) Subscribe

func (m *Mock) Subscribe(topic, queue string, cb func([]byte)) Subscription

Subscribe mocked call.

func (*Mock) Unsubscribe

func (m *Mock) Unsubscribe(topic string) error

Unsubscribe mocked call.

type PubSub

type PubSub struct {
	// Logger.
	Logger sypl.ISypl `json:"-" validate:"required"`

	// Name of the pubsub type.
	Name string `json:"name" validate:"required,lowercase,gte=1"`
	// contains filtered or unexported fields
}

PubSub definition.

func New

func New(name string) (*PubSub, error)

New returns a new pubsub.

func (*PubSub) ErrorHandler

func (p *PubSub) ErrorHandler(ctx context.Context, operation string, err error) error

ErrorHandler trace and logs an already wrapped (customerror) error.

func (*PubSub) FailedToErrorHandler

func (p *PubSub) FailedToErrorHandler(ctx context.Context, operation string, err error) error

FailedToErrorHandler trace and logs a failedTo error.

func (*PubSub) FailedToErrorMsgHandler

func (p *PubSub) FailedToErrorMsgHandler(ctx context.Context, msg string, err error) error

FailedToErrorMsgHandler trace and logs an already wrapped (customerror) error.

func (*PubSub) GetLogger

func (p *PubSub) GetLogger() sypl.ISypl

GetLogger returns the logger.

func (*PubSub) GetName

func (p *PubSub) GetName() string

GetName returns the storage name.

func (*PubSub) GetPublishCounter

func (p *PubSub) GetPublishCounter() *expvar.Int

GetPublishCounter returns the counterCount metric.

func (*PubSub) GetSubscribeCounter

func (p *PubSub) GetSubscribeCounter() *expvar.Int

GetSubscribeCounter returns the counterCount metric.

func (*PubSub) GetType

func (p *PubSub) GetType() string

GetType returns its type.

func (*PubSub) NotFoundErrorHandler

func (p *PubSub) NotFoundErrorHandler(ctx context.Context) error

NotFoundErrorHandler trace and logs a notFound error.

type Subscription

type Subscription struct {
	// Topic is the subject to subscribe to.
	Topic string

	// Queue is the queue to subscribe to.
	Queue string

	// Callback is the function to call when a message is received.
	Callback func(msg []byte)

	// Channel is the channel to receive messages.
	Channel <-chan []byte
}

Subscription is a subscription to a topic.

func (*Subscription) HandleMessage

func (s *Subscription) HandleMessage(msg *natsgo.Msg)

HandleMessage calls the callback function to handle the message.