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) ([]*message.Message, error)
  • func (m Map) SubscribeMany(ctx context.Context, subscriptions []*subscription.Subscription, opts ...Func) ([]*subscription.Subscription, concurrentloop.Errors)
• type Mock
  • func (m *Mock) Close() error
  • func (m *Mock) GetClient() any
  • func (m *Mock) GetCounterPingFailed() *expvar.Int
  • func (m *Mock) GetLogger() sypl.ISypl
  • func (m *Mock) GetName() string
  • func (m *Mock) GetPublishedCounter() *expvar.Int
  • func (m *Mock) GetPublishedFailedCounter() *expvar.Int
  • func (m *Mock) GetSubscribedCounter() *expvar.Int
  • func (m *Mock) GetSubscribedFailedCounter() *expvar.Int
  • func (m *Mock) GetType() string
  • func (m *Mock) MustPublish(ctx context.Context, msgs ...*message.Message) []*message.Message
  • func (m *Mock) MustPublishAsync(ctx context.Context, messages ...*message.Message)
  • func (m *Mock) MustSubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) []*subscription.Subscription
  • func (m *Mock) MustSubscribeAsync(ctx context.Context, subscriptions ...*subscription.Subscription)
  • func (m *Mock) Publish(ctx context.Context, messages []*message.Message, opts ...Func) ([]*message.Message, concurrentloop.Errors)
  • func (m *Mock) Subscribe(ctx context.Context, subscriptions []*subscription.Subscription, opts ...Func) ([]*subscription.Subscription, concurrentloop.Errors)
  • func (m *Mock) Unsubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) error
• type Options
  • func NewOptions() (*Options, error)
• type PubSub
  • func New(ctx context.Context, name string) (*PubSub, error)
  • func (p *PubSub) GetCounterPingFailed() *expvar.Int
  • func (p *PubSub) GetLogger() sypl.ISypl
  • func (p *PubSub) GetName() string
  • func (p *PubSub) GetPublishedCounter() *expvar.Int
  • func (p *PubSub) GetPublishedFailedCounter() *expvar.Int
  • func (p *PubSub) GetSubscribedCounter() *expvar.Int
  • func (p *PubSub) GetSubscribedFailedCounter() *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, opts ...Func) ([]*subscription.Subscription, concurrentloop.Errors)

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

	// MustSubscribeAsync to a topic asynchronously. In case of error it will panic.
	MustSubscribeAsync(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

	// GetType returns its type.
	GetType() string

	// GetCounterPingFailed returns the metric.
	GetCounterPingFailed() *expvar.Int

	// GetPublishedCounter returns the metric.
	GetPublishedCounter() *expvar.Int

	// GetPublishedFailedCounter returns the metric.
	GetPublishedFailedCounter() *expvar.Int

	// GetSubscribedCounter returns the metric.
	GetSubscribedCounter() *expvar.Int

	// GetSubscribedFailedCounter returns the metric.
	GetSubscribedFailedCounter() *expvar.Int
}

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,
) ([]*message.Message, error)

PublishMany will make all PubSubs to concurrently publish many messages.

func (Map) SubscribeMany

func (m Map) SubscribeMany(
	ctx context.Context,
	subscriptions []*subscription.Subscription,
	opts ...Func,
) ([]*subscription.Subscription, concurrentloop.Errors)

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(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.
	MockMustPublish func(ctx context.Context, msgs ...*message.Message) []*message.Message

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

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

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

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

	// Unsubscribe from a topic.
	MockUnsubscribe func(ctx context.Context, subscriptions ...*subscription.Subscription) 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 pubsub name.
	MockGetName func() string

	// GetType returns its type.
	MockGetType func() string

	// GetCounterPingFailed returns the metric.
	MockGetCounterPingFailed func() *expvar.Int

	// GetPublishedCounter returns the metric.
	MockGetPublishedCounter func() *expvar.Int

	// GetPublishedFailedCounter returns the metric.
	MockGetPublishedFailedCounter func() *expvar.Int

	// GetSubscribedCounter returns the metric.
	MockGetSubscribedCounter func() *expvar.Int

	// GetSubscribedFailedCounter returns the metric.
	MockGetSubscribedFailedCounter func() *expvar.Int
}

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

func (*Mock) Close

func (m *Mock) Close() error

Close the connection to the Pub Sub broker.

func (*Mock) GetClient

func (m *Mock) GetClient() any

GetClient returns the storage client. Use that to interact with the underlying storage client.

func (*Mock) GetCounterPingFailed

func (m *Mock) GetCounterPingFailed() *expvar.Int

GetCounterPingFailed returns the metric.

func (*Mock) GetLogger

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

GetLogger returns the logger.

func (*Mock) GetName

func (m *Mock) GetName() string

GetName returns the pubsub name.

func (*Mock) GetPublishedCounter

func (m *Mock) GetPublishedCounter() *expvar.Int

GetPublishedCounter returns the metric.

func (*Mock) GetPublishedFailedCounter

func (m *Mock) GetPublishedFailedCounter() *expvar.Int

GetPublishedFailedCounter returns the metric.

func (*Mock) GetSubscribedCounter

func (m *Mock) GetSubscribedCounter() *expvar.Int

GetSubscribedCounter returns the metric.

func (*Mock) GetSubscribedFailedCounter

func (m *Mock) GetSubscribedFailedCounter() *expvar.Int

GetSubscribedFailedCounter returns the metric.

func (*Mock) GetType

func (m *Mock) GetType() string

GetType returns its type.

func (*Mock) MustPublish

func (m *Mock) MustPublish(ctx context.Context, msgs ...*message.Message) []*message.Message

MustPublish sends a message to a topic. In case of error it will panic.

func (*Mock) MustPublishAsync

func (m *Mock) MustPublishAsync(ctx context.Context, messages ...*message.Message)

MustPublishAsync sends a message to a topic asynchronously. In case of error it will panic.

func (*Mock) MustSubscribe

func (m *Mock) MustSubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) []*subscription.Subscription

MustSubscribe to a topic. In case of error it will panic.

func (*Mock) MustSubscribeAsync

func (m *Mock) MustSubscribeAsync(ctx context.Context, subscriptions ...*subscription.Subscription)

MustSubscribeAsync to a topic asynchronously. In case of error it will panic.

func (*Mock) Publish

func (m *Mock) Publish(ctx context.Context, messages []*message.Message, opts ...Func) ([]*message.Message, concurrentloop.Errors)

Publish sends a message to a topic.

func (*Mock) Subscribe

func (m *Mock) Subscribe(ctx context.Context, subscriptions []*subscription.Subscription, opts ...Func) ([]*subscription.Subscription, concurrentloop.Errors)

Subscribe to a topic.

func (*Mock) Unsubscribe

func (m *Mock) Unsubscribe(ctx context.Context, subscriptions ...*subscription.Subscription) error

Unsubscribe from a topic.

type Options

type Options struct {
	// If the operation is synchronous.
	Sync bool `default:"false" env:"PUBSUB_SYNC" json:"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(ctx context.Context, name string) (*PubSub, error)

New returns a new pubsub.

func (*PubSub) GetCounterPingFailed

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

GetCounterPingFailed returns the metric.

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

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

GetPublishedCounter returns the metric.

func (*PubSub) GetPublishedFailedCounter

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

GetPublishedFailedCounter returns the metric.

func (*PubSub) GetSubscribedCounter

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

GetSubscribedCounter returns the metric.

func (*PubSub) GetSubscribedFailedCounter

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

GetSubscribedFailedCounter returns the metric.

func (*PubSub) GetType

func (p *PubSub) GetType() string

GetType returns its type.