pubsubs

Documentation

Overview

Package pubsubs implements remote communication in actorkit using message queues and pubsub services.

Implemented are the ff:

NATS and NATS Streaming

Google PubSub Apache Kafka Redis PubSub

Index

• Constants
• Variables
• type Action
  • func (a Action) String() string
• type DesubscriptionError
  • func (m DesubscriptionError) Message() string
• type Marshaler
• type MarshalingError
  • func (m MarshalingError) Message() string
• type Message
  • func NewMessage(topic string, env actorkit.Envelope) Message
• type MessageHandlingError
  • func (m MessageHandlingError) Message() string
• type OpError
  • func (m OpError) Message() string
• type PubSubFactory
• type PubSubFactoryImpl
  • func (p PubSubFactoryImpl) NewPublisher(topic string) (Publisher, error)
  • func (p PubSubFactoryImpl) NewQueueGroupSubscriber(group string, topic string, id string, r Receiver) (Subscription, error)
  • func (p PubSubFactoryImpl) NewSubscriber(topic string, id string, r Receiver) (Subscription, error)
• type PublishError
  • func (m PublishError) Message() string
• type Publisher
• type PublisherFactory
• type PublisherHandler
• type QueueGroupSubscriberHandler
• type QueueGroupSubscriptionFactory
• type Receiver
• type SubscriberHandler
• type Subscription
• type SubscriptionError
  • func (m SubscriptionError) Message() string
• type SubscriptionFactory
• type Unmarshaler
• type UnmarshalingError
  • func (m UnmarshalingError) Message() string

Constants

const (
	// SubscriberTopicFormat defines the expected format for a subscriber group name, queue name can be formulated.
	SubscriberTopicFormat = "/pubsub/%s/project/%s/topics/%s/subscriber/%s"

	// QueueGroupSubscriberTopicFormat defines the expected format for a subscriber queue group name, queue name can be formulated.
	QueueGroupSubscriberTopicFormat = "/pubsub/%s/project/%s/topics/%s/subscriber/%s/%s"
)

Variables

var (
	// ErrNotSupported is returned when a giving feature or method has no implementation
	// support.
	ErrNotSupported = errors.New("method not supported")
)

Types

type Action

type Action uint8

Action defines a giving response to be provided by the processing of a message by a Receiver function type.

const (
	// ACK is for acknowledging a message received.
	ACK Action = 1 << iota

	// NACK is to not acknowledge or reject a message received.
	NACK

	// NOPN is to request a severe action as dictated by the implementation
	// detail as a action to a giving response/request.
	NOPN
)

constants of action types

func (Action) String

func (a Action) String() string

type DesubscriptionError

type DesubscriptionError struct {
	Topic string
	Err   error
}

DesubscriptionError defines a giving error struct for subscription error.

func (DesubscriptionError) Message

func (m DesubscriptionError) Message() string

Message implements the actorkit.Logs interface.

type Marshaler

type Marshaler interface {
	Marshal(actorkit.Envelope) ([]byte, error)
}

Marshaler exposes a method to turn an envelope into a byte slice.

type MarshalingError

type MarshalingError struct {
	Topic string
	Err   error
	Data  interface{}
}

MarshalingError to be used for errors corresponding with marshaling of data.

func (MarshalingError) Message

func (m MarshalingError) Message() string

Message implements the actorkit.Logs interface.

type Message

type Message struct {
	Topic    string
	Envelope actorkit.Envelope
}

Message defines a type which embodies a topic to be published to and the associated envelope for that topic.

func NewMessage

func NewMessage(topic string, env actorkit.Envelope) Message

NewMessage returns a new instance of a Message with a given topic and envelope.

type MessageHandlingError

type MessageHandlingError struct {
	Topic string
	Err   error
	Data  interface{}
}

MessageHandlingError is to be used for errors related to handling received messages.

func (MessageHandlingError) Message

func (m MessageHandlingError) Message() string

Message implements the actorkit.Logs interface.

type OpError

type OpError struct {
	Topic string
	Err   error
}

OpError is to be used for errors related to publishing giving data.

func (OpError) Message

func (m OpError) Message() string

Message implements the actorkit.Logs interface.

type PubSubFactory

type PubSubFactory interface {
	PublisherFactory
	SubscriptionFactory
	QueueGroupSubscriptionFactory
}

PubSubFactory defines an interface which embodies the methods exposed for the publishing and subscription of topics and their corresponding messages.

type PubSubFactoryImpl

type PubSubFactoryImpl struct {
	Publishers            PublisherHandler
	Subscribers           SubscriberHandler
	QueueGroupSubscribers QueueGroupSubscriberHandler
}

PubSubFactoryImpl implements the PubSubFactory interface, allowing providing custom generator functions which will returning appropriate Publishers and Subscribers for some underline platform.

func (PubSubFactoryImpl) NewPublisher

func (p PubSubFactoryImpl) NewPublisher(topic string) (Publisher, error)

NewPublisher returns a new Publisher using the Publishers handler function provided.

func (PubSubFactoryImpl) NewQueueGroupSubscriber

func (p PubSubFactoryImpl) NewQueueGroupSubscriber(group string, topic string, id string, r Receiver) (Subscription, error)

NewQueueGroupSubscriber returns a new Subscriber using the Subscribers handler function provided.

func (PubSubFactoryImpl) NewSubscriber

func (p PubSubFactoryImpl) NewSubscriber(topic string, id string, r Receiver) (Subscription, error)

NewSubscriber returns a new Subscriber using the Subscribers handler function provided.

type PublishError

type PublishError struct {
	Topic string
	Err   error
	Data  interface{}
}

PublishError is to be used for errors related to publishing giving data.

func (PublishError) Message

func (m PublishError) Message() string

Message implements the actorkit.Logs interface.

type Publisher

type Publisher interface {
	Close() error
	Publish(actorkit.Envelope) error
}

Publisher exposes a method for the publishing of a provided message.

type PublisherFactory

type PublisherFactory interface {
	NewPublisher(string) (Publisher, error)
}

PublisherFactory exposes a single method for the return of a giving publisher for a provided topic.

type PublisherHandler

type PublisherHandler func(string) (Publisher, error)

PublisherHandler defines a function type which takes a giving PublisherFactory and a given topic, returning a new publisher with all related underline specific details added and instantiated.

type QueueGroupSubscriberHandler

type QueueGroupSubscriberHandler func(group string, topic string, id string, r Receiver) (Subscription, error)

QueueGroupSubscriberHandler defines a function type which takes a giving SubscriptionFactory and a given topic, returning a new subscription for a giving queue group name.

type QueueGroupSubscriptionFactory

type QueueGroupSubscriptionFactory interface {
	NewQueueGroupSubscriber(string, string, string, Receiver) (Subscription, error)
}

QueueGroupSubscriptionFactory exposes a given method for the creation of a subscription.

type Receiver

type Receiver func(Message) (Action, error)

Receiver defines a function type to be used for processing of an incoming message.

type SubscriberHandler

type SubscriberHandler func(topic string, id string, r Receiver) (Subscription, error)

SubscriberHandler defines a function type which takes a giving SubscriptionFactory and a given topic, returning a new subscription with all related underline specific details added and instantiated.

type Subscription

type Subscription interface {
	actorkit.Subscription

	ID() string
	Topic() string
	Group() string
}

Subscription expects the implementer to provide methods to identify the topic, id and group/queueGroup name of giving subscription and a method to stop or end it.

type SubscriptionError

type SubscriptionError struct {
	Topic string
	Err   error
}

SubscriptionError defines a giving error struct for subscription error.

func (SubscriptionError) Message

func (m SubscriptionError) Message() string

Message implements the actorkit.Logs interface.

type SubscriptionFactory

type SubscriptionFactory interface {
	NewSubscriber(topic string, id string, r Receiver) (Subscription, error)
}

SubscriptionFactory exposes a given method for the creation of a subscription.

type Unmarshaler

type Unmarshaler interface {
	Unmarshal([]byte) (actorkit.Envelope, error)
}

Unmarshaler exposes a method to turn an byte slice into a envelope.

type UnmarshalingError

type UnmarshalingError struct {
	Topic string
	Err   error
	Data  interface{}
}

UnmarshalingError is to be used for errors relating to deserialization of serialized data.

func (UnmarshalingError) Message

func (m UnmarshalingError) Message() string

Message implements the actorkit.Logs interface.