pubsub

Documentation

Index

• Variables
• type ErrEventSequenceGap
  • func (err *ErrEventSequenceGap) Error() string
• type Event
• type EventHandler
• type EventListener
• type EventSeqNum
• type EventSeqTable
• type Service
  • func NewService(factory func() (Transport, error)) (srv *Service, err error)
  • func (srv *Service) Close() error
  • func (srv *Service) Closed() <-chan struct{}
  • func (srv *Service) Monitor(errChan chan<- error)
  • func (srv *Service) Publish(eventKind string, eventObject interface{}) error
  • func (srv *Service) RemoveListener(listener EventListener) error
  • func (srv *Service) Subscribe(eventKindPrefix string, handler EventHandler) (EventListener, error)
  • func (srv *Service) Unsubscribe(eventKindPrefix string) error
  • func (srv *Service) Wait() error
• type Transport

Variables

var (
	ErrListenerHandlesDepleted = errors.New("EventListener handles depleted")
)

Types

type ErrEventSequenceGap

type ErrEventSequenceGap struct {
	EventKind   string
	ExpectedSeq EventSeqNum
	ReceivedSeq EventSeqNum
}

func (*ErrEventSequenceGap) Error

func (err *ErrEventSequenceGap) Error() string

type Event

type Event interface {
	// Kind returns the event kind this event was published as.
	Kind() string

	// Seq returns this event's sequence number.
	Seq() EventSeqNum

	// Unmarshal unmarshalls the received event into dst, which must support
	// decoding using github.com/ugorji/go/codec.
	Unmarshal(dst interface{}) error
}

Event represents an event received on a transport.

type EventHandler

type EventHandler func(event Event)

type EventListener

type EventListener uint16

type EventSeqNum

type EventSeqNum uint32

type EventSeqTable

type EventSeqTable map[string]EventSeqNum

type Service

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

Service represents a PubSub service instance.

All methods are thread-safe if not stated otherwise.

func NewService

func NewService(factory func() (Transport, error)) (srv *Service, err error)

NewService uses factory to construct a Transport instance that is then used for creating a Service instance.

factory can panic without any worries, it just makes NewService return an error, so some if errs can be saved.

func (*Service) Close

func (srv *Service) Close() error

Close terminated the service as well as the underlying transport.

func (*Service) Closed

func (srv *Service) Closed() <-chan struct{}

Closed return a channel that is closed once the service is terminated.

func (*Service) Monitor

func (srv *Service) Monitor(errChan chan<- error)

Monitor registers errChan for receiving service errors not really connected to the transport itself, but rather to some bad service conditions.

Possible error types that can be received on this channel:

• *ErrEventSequenceGap - some events were missed due to transport overload

func (*Service) Publish

func (srv *Service) Publish(eventKind string, eventObject interface{}) error

Publish sends eventObject to Meeko and thus publishes it for other apps.

eventObject must be marshallable by github.com/ugorji/go/codec.

func (*Service) RemoveListener

func (srv *Service) RemoveListener(listener EventListener) error

RemoveListener unregisters listener and possibly unsubscribes from the kind prefix that listener was listening for. That may fail and return an error.

func (*Service) Subscribe

func (srv *Service) Subscribe(eventKindPrefix string, handler EventHandler) (EventListener, error)

Subscribe registers a handler for events starting with eventKindPrefix. There can be multiple handlers registered for given event kind prefix, but Subscribe does not and cannot check for handler duplicates, so do not try to subscribe the same handler for the same event prefix multiple times because it will work.

EventListener returned by this method can be later used to remove the handler by calling RemoveListener method.

func (*Service) Unsubscribe

func (srv *Service) Unsubscribe(eventKindPrefix string) error

Unsubscribe cancels all subscriptions with the given event kind prefix and removes all assigned event handlers.

func (*Service) Wait

func (srv *Service) Wait() error

Wait blocks until the service is terminated and return the last unrecoverable error encountered.

type Transport

type Transport interface {
	// Inherit some basic functionality.
	services.Transport

	// Publish does exactly what the name says - it publishes the given event
	// object under eventKind.
	//
	// eventObject must be marshallable by github.com/ugorji/go/codec.
	Publish(eventKind string, eventObject interface{}) error

	// Subscribe sets this transport's event filter to receive all events
	// having their kind starting with eventKindPrefix.
	Subscribe(eventKindPrefix string) error

	// Unsubscribe does exactly the opposite to Subscribe. It cannot be,
	// however, called with any random event kind prefix. It must be one of the
	// prefixes that were used in a previous call to Subscribe.
	Unsubscribe(eventKindPrefix string) error

	// EventChan returns a channel that can be used for receiving events from
	// this transport, events that this transport is subscribed for.
	EventChan() <-chan Event

	// EventSeqTableChan returns a channel that can be used for receiving sync
	// messages from the broker. There messages are basically tables containing
	// the current event sequence numbers for specific events at the time that
	// the event table request reached the broker.
	EventSeqTableChan() <-chan EventSeqTable

	// ErrChan returns a channel for emitting internal transport errors.
	// Any error sent to this channel is treated as unrecoverable and makes
	// the service using the transport terminate.
	ErrorChan() <-chan error
}