pubsub

Documentation

Index

• Variables
• type ChannelUnsubscribeCallBack
• type NatsClient
• type NatsClientFactory
  • func (r *NatsClientFactory[T, A, P]) Get() (P, error)
• type NatsConfig
  • func (e *NatsConfig) GetConfig() (string, string, error)
• type NatsConnection
  • func NewNatsConnection(urls, clientName string) (*NatsConnection, error)
  • func (psc *NatsConnection) Connect() (*nats.Conn, error)
• type PubSubOpt
• type Publication
  • func NewPublication[T any](conn *NatsConnection) Publication[T]
  • func (p *Publication[T]) Publish(subj string, data []byte) error
  • func (p *Publication[T]) PublishObject(subj string, o T) error
• type Subscription
  • func NewSubscription(psc *NatsConnection) Subscription
  • func (s *Subscription) Subscribe(subj string, sc nats.MsgHandler) error
  • func (s *Subscription) SubscribeChan(subj string, ch chan *nats.Msg) (ChannelUnsubscribeCallBack, error)
  • func (s *Subscription) UnsubscribeFn() ChannelUnsubscribeCallBack

Variables

var (
	ErrFailedToResolveDbNatsOpt = errors.New("failed to resolve nats options")
	ErrFailedToConnectToNats    = errors.New("failed to connect to nats")
)

Types

type ChannelUnsubscribeCallBack

type ChannelUnsubscribeCallBack func() error

type NatsClient

type NatsClient[T any, A any] interface {
	*T
	Connection(*NatsConnection)
	GetConnection() *NatsConnection
	MarshalEventPayload(A) ([]byte, error)
}

type NatsClientFactory

type NatsClientFactory[T any, A any, P NatsClient[T, A]] struct {
	// contains filtered or unexported fields
}

func (*NatsClientFactory[T, A, P]) Get

func (r *NatsClientFactory[T, A, P]) Get() (P, error)

type NatsConfig

type NatsConfig struct {
	NatsUrl string
}

func (*NatsConfig) GetConfig

func (e *NatsConfig) GetConfig() (string, string, error)

GetUrl gets the path to message queue from the env variable and client name

type NatsConnection

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

func NewNatsConnection

func NewNatsConnection(urls, clientName string) (*NatsConnection, error)

NewNatsConnection creates and returns a new nats connection with urls of the cluster and client name. The connection is not established at this stage. Urls are a string a instances of cluster seperated by space, or a single url. If urls and client name are not provided, they will be resolved from env variables.

func (*NatsConnection) Connect

func (psc *NatsConnection) Connect() (*nats.Conn, error)

Connect returns an already established connection or establish a new connection and returns it.

type PubSubOpt

type PubSubOpt struct {
	Urls       string
	ClientName string
}

type Publication

type Publication[T any] struct {
	// contains filtered or unexported fields
}

func NewPublication

func NewPublication[T any](conn *NatsConnection) Publication[T]

func (*Publication[T]) Publish

func (p *Publication[T]) Publish(subj string, data []byte) error

func (*Publication[T]) PublishObject

func (p *Publication[T]) PublishObject(subj string, o T) error

type Subscription

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

func NewSubscription

func NewSubscription(psc *NatsConnection) Subscription

func (*Subscription) Subscribe

func (s *Subscription) Subscribe(subj string, sc nats.MsgHandler) error

func (*Subscription) SubscribeChan

func (s *Subscription) SubscribeChan(subj string, ch chan *nats.Msg) (ChannelUnsubscribeCallBack, error)

func (*Subscription) UnsubscribeFn

func (s *Subscription) UnsubscribeFn() ChannelUnsubscribeCallBack