manager

Documentation

Index

• type Manager
  • func NewNatsManager(config NATSManagerConfig) Manager
• type Message
• type NATSManagerConfig
• type NatsSubscription
  • func NewNatsSubscription(subscriptionTopic topic.Topic, connection *nats.EncodedConn) (*NatsSubscription, error)
  • func (natsSub *NatsSubscription) GetChan() <-chan interface{}
  • func (natsSub *NatsSubscription) Unsubscribe() error
• type Subscription

Types

type Manager

type Manager interface {
	// Load loads GeoSubscriptions from backend store and launches the ones that aren't
	// up and running yet. The Load call might be performed multiple times to
	// update the list.
	Refresh([]output.GeoSubscription)

	// Update refreshes the GeoSubscription. If it isn't launched yet it will be
	// launched. If it is already running the new configuration will be applied.
	Update(output.GeoSubscription) error

	// Stop stops a single subscription, typically if they have been deleted.
	Stop(subscriptionID int64) error

	// Shutdown shuts down all of the running subscriptions.
	Shutdown()

	// Get returns the GeoSubscription. If the subscription isn't running or is unknown it
	// will return an error.
	Get(SubscriptionID int64) (output.GeoSubscription, error)

	// Publish publishes an event to the event bus on the given topic. If there's no
	// subscriptions subscribing to the topic it will be discarded.
	Publish(topic topic.Topic, event event.PublishableEvent)

	// Subscribe subscribes to a topic
	Subscribe(topic.Topic) (Subscription, error)
}

Manager is responsible for keeping track of events and event subscriptions along with handling GeoSubscriptions which contains geo indexes for lookup and calculations

func NewNatsManager

func NewNatsManager(config NATSManagerConfig) Manager

NewNatsManager creates a new manager

type Message

type Message interface {
	Type() string
	Payload() []byte
}

Message is an intermediate message wrapping the payload with a type so you can correctly get the right interface

type NATSManagerConfig

type NATSManagerConfig struct {
	Logging   bool   `param:"desc=Whether the NATS server should show debug messages;default=false"`
	StoreType string `param:"desc=The type of store used for NATS;options=memory,file;default=memory"`

	// File-specific config
	FileStoreDir string `param:"desc=The directory where to store the NATS subscriptions (if store-type is 'file');default=nats"`

	// NATS connection
	Host string `param:"desc=The host url;default=localhost"`
	Port int    `param:"desc=;default=4222"`
}

type NatsSubscription

type NatsSubscription struct {
	Subscription *nats.Subscription
	Chan         chan interface{}
}

NatsSubscription is a subscription with NATS as backend, creating a subscription and piping into a channel

func NewNatsSubscription

func NewNatsSubscription(subscriptionTopic topic.Topic, connection *nats.EncodedConn) (*NatsSubscription, error)

NewNatsSubscription initializes a Susbcription on given connection and pipes the data into a local channel

func (*NatsSubscription) GetChan

func (natsSub *NatsSubscription) GetChan() <-chan interface{}

GetChan returns a channel for retrieving data

func (*NatsSubscription) Unsubscribe

func (natsSub *NatsSubscription) Unsubscribe() error

Unsubscribe cleans up the NATS-subscription and closes the channel

type Subscription

type Subscription interface {
	Unsubscribe() error
	GetChan() <-chan interface{}
}

Subscription ...