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 Func
  • func WithSync(sync bool) Func
• type IPubSub
• type Map
  • func (m Map) MustPublishManyAsync(ctx context.Context, messages ...*message.Message)
  • func (m Map) MustSubscribeManyAsync(ctx context.Context, subscriptions ...*subscription.Subscription)
  • func (m Map) PublishMany(ctx context.Context, messages []*message.Message, opts ...Func)
  • func (m Map) SubscribeMany(ctx context.Context, subscriptions ...*subscription.Subscription)
• 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) MustPublishAsync(topic string, message any)
  • func (m *Mock) Publish(topic string, message any) error
  • func (m *Mock) Subscribe(topic, queue string, cb func([]byte)) (subscription.Subscription, error)
  • func (m *Mock) Unsubscribe(topic string) error
• type Options
  • func NewOptions() (*Options, error)
• type PubSub
  • func New(name string) (*PubSub, 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

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 Func

type Func func(o *Options) error

Func allows to set options.

func WithSync

func WithSync(sync bool) Func

WithSync set the sync option.

type IPubSub

type IPubSub interface {
	// Publish sends a message to a topic.
	Publish(ctx context.Context, messages []*message.Message, opts ...Func) ([]*message.Message, concurrentloop.Errors)

	// MustPublish sends a message to a topic. In case of error it will panic.
	MustPublish(ctx context.Context, msgs ...*message.Message) []*message.Message

	// MustPublishAsync sends a message to a topic asynchronously. In case of
	// error it will panic.
	MustPublishAsync(ctx context.Context, messages ...*message.Message)

	// Subscribe to a topic.
	Subscribe(ctx context.Context, subscriptions ...*subscription.Subscription) ([]*subscription.Subscription, concurrentloop.Errors)

	// MustSubscribe to a topic. In case of error it will panic.
	MustSubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) []*subscription.Subscription

	// MustSubscribeAsyn to a topic asynchronously. In case of error it will panic.
	MustSubscribeAsyn(ctx context.Context, subscriptions ...*subscription.Subscription)

	// Unsubscribe from a topic.
	Unsubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) error

	// Close 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 Map

type Map map[string]IPubSub

Map is a map of PubSubs

func (Map) MustPublishManyAsync

func (m Map) MustPublishManyAsync(ctx context.Context, messages ...*message.Message)

MustPublishManyAsync will make all PubSubs to concurrently publish many messages asynchronously.

func (Map) MustSubscribeManyAsync

func (m Map) MustSubscribeManyAsync(ctx context.Context, subscriptions ...*subscription.Subscription)

MustSubscribeManyAsync will make all PubSubs to concurrently subscribe to many subscriptions asynchronously.

func (Map) PublishMany

func (m Map) PublishMany(ctx context.Context, messages []*message.Message, opts ...Func)

PublishMany will make all PubSubs to concurrently publish many messages.

func (Map) SubscribeMany

func (m Map) SubscribeMany(ctx context.Context, subscriptions ...*subscription.Subscription)

SubscribeMany will make all PubSubs to concurrently subscribe to many subscriptions.

type Mock

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

	// MustPublishAsync sends a message to a topic. In case of error it will just log
	// it.
	MockPublishAsync func(topic string, message any)

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

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

	// Close 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) MustPublishAsync

func (m *Mock) MustPublishAsync(topic string, message any)

MustPublishAsync sends a message to a topic. In case of error it will just log it.

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.Subscription, error)

Subscribe mocked call.

func (*Mock) Unsubscribe

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

Unsubscribe mocked call.

type Options

type Options struct {
	// If the operation is synchronous.
	Sync bool `json:"sync" default:"false" env:"PUBSUB_SYNC"`
}

Options for operations.

func NewOptions

func NewOptions() (*Options, error)

NewOptions creates Options.

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) 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.