dtopic

Documentation

Index

• Constants
• Variables
• func NewService(e *environment.Environment) (service.Service, error)
• type DTopic
  • func (d *DTopic) AddListener(f func(Message)) (uint64, error)
  • func (d *DTopic) Destroy() error
  • func (d *DTopic) Publish(msg interface{}) error
  • func (d *DTopic) RemoveListener(listenerID uint64) error
• type Dispatcher
  • func NewDispatcher(ctx context.Context) *Dispatcher
• type Message
• type Service
  • func (s *Service) NewDTopic(name string, concurrency int, flag int16) (*DTopic, error)
  • func (s *Service) RegisterOperations(operations map[protocol.OpCode]func(w, r protocol.EncodeDecoder))
  • func (s *Service) Shutdown(ctx context.Context) error
  • func (s *Service) Start() error

Constants

const (
	// UnorderedDelivery means that messages are delivered in random order. It's good to distribute independent events in a distributed system.
	UnorderedDelivery = int16(1) << iota

	// OrderedDelivery means that messages are delivered in some order. Not implemented yet.
	OrderedDelivery
)

Variables

var (
	// PublishedTotal is the total number of published messages during the life of this instance.
	PublishedTotal = stats.NewInt64Counter()

	// CurrentListeners is the current number of listeners of DTopics.
	CurrentListeners = stats.NewInt64Gauge()

	// ListenersTotal is the total number of registered listeners during the life of this instance.
	ListenersTotal = stats.NewInt64Counter()
)

var ErrServerGone = errors.New("server is gone")

Functions

func NewService

func NewService(e *environment.Environment) (service.Service, error)

Types

type DTopic

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

DTopic implements a distributed topic to deliver messages between clients and Olric nodes. You should know that:

• Communication between parties is one-to-many (fan-out).
• All data is in-memory, and the published messages are not stored in the cluster.
• Fire&Forget: message delivery is not guaranteed.

func (*DTopic) AddListener

func (d *DTopic) AddListener(f func(Message)) (uint64, error)

AddListener adds a new listener for the topic. Returns a registration ID or a non-nil error. Registered functions are run by parallel.

func (*DTopic) Destroy

func (d *DTopic) Destroy() error

Destroy removes all listeners for this topic on the cluster. If Publish function is called again after Destroy, the topic will be recreated.

func (*DTopic) Publish

func (d *DTopic) Publish(msg interface{}) error

Publish publishes the given message to listeners of the topic. Message order and delivery are not guaranteed.

func (*DTopic) RemoveListener

func (d *DTopic) RemoveListener(listenerID uint64) error

RemoveListener removes a listener with the given listenerID.

type Dispatcher

type Dispatcher struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewDispatcher

func NewDispatcher(ctx context.Context) *Dispatcher

type Message

type Message struct {
	Message       interface{}
	PublisherAddr string
	PublishedAt   int64
}

Message is a message type for DTopic data structure.

type Service

type Service struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*Service) NewDTopic

func (s *Service) NewDTopic(name string, concurrency int, flag int16) (*DTopic, error)

NewDTopic returns a new distributed topic instance. Parameters:

• name: DTopic name.
• concurrency: Maximum number of concurrently processing DTopic messages.
• flag: Any flag to control DTopic behaviour.

Flags for delivery options:

• UnorderedDelivery: Messages are delivered in random order. It's good to distribute independent events in a distributed system.
• OrderedDelivery: Messages are delivered in order. Not implemented yet.

func (*Service) RegisterOperations

func (s *Service) RegisterOperations(operations map[protocol.OpCode]func(w, r protocol.EncodeDecoder))

func (*Service) Shutdown

func (s *Service) Shutdown(ctx context.Context) error

func (*Service) Start

func (s *Service) Start() error