eventbus

Documentation

Index

• Variables
• type DistributedEventBus
  • func NewDistributedEventBus(nodeID string, membership *cluster.Membership, consensus *consensus.Consensus) *DistributedEventBus
  • func (bus *DistributedEventBus) HandleRemoteEvent(eventBytes []byte) error
  • func (bus *DistributedEventBus) HandleSerfEvent(e serf.Event)
  • func (bus *DistributedEventBus) PublishCluster(namespace, topic string, payload map[string]interface{}, tags []string) error
  • func (bus *DistributedEventBus) PublishLeader(namespace, topic string, payload map[string]interface{}, tags []string) error
  • func (bus *DistributedEventBus) PublishLocal(namespace, topic string, payload map[string]interface{}, tags []string) error
  • func (bus *DistributedEventBus) Subscribe(topic string, handler EventHandler) Subscription
  • func (bus *DistributedEventBus) Unsubscribe(sub *Subscription)
• type Event
• type EventBus
• type EventHandler
• type EventRouter
  • func NewEventRouter() *EventRouter
  • func (r *EventRouter) AddRule(topicPrefix string, strategy RoutingStrategy)
  • func (r *EventRouter) GetStrategy(topic string) RoutingStrategy
  • func (r *EventRouter) RequiresLeader(topic string) bool
  • func (r *EventRouter) ShouldReplicate(topic string) bool
• type RoutingStrategy
• type Subscription
  • func (s *Subscription) Unsubscribe()

Variables

var ErrClusterEventOversized = errors.New("cluster event exceeds serf user-event size limit")

ErrClusterEventOversized indicates a marshalled ClusterEvent would not fit within serf's user-event size limit.

Types

type DistributedEventBus

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

DistributedEventBus manages cluster-wide event distribution

func NewDistributedEventBus

func NewDistributedEventBus(nodeID string, membership *cluster.Membership, consensus *consensus.Consensus) *DistributedEventBus

NewDistributedEventBus creates a new distributed event bus

func (*DistributedEventBus) HandleRemoteEvent

func (bus *DistributedEventBus) HandleRemoteEvent(eventBytes []byte) error

HandleRemoteEvent processes an event received from another node

func (*DistributedEventBus) HandleSerfEvent

func (bus *DistributedEventBus) HandleSerfEvent(e serf.Event)

HandleSerfEvent ingests a Serf event into the bus: goblin.event user events from other nodes are unmarshaled and dispatched to local subscribers. The supervisor's single Serf handler calls this - Serf supports one handler, so the bus cannot register its own (this closes the proto-2 'distributed eventbus integration point' seam).

func (*DistributedEventBus) PublishCluster

func (bus *DistributedEventBus) PublishCluster(namespace, topic string, payload map[string]interface{}, tags []string) error

PublishCluster publishes an event to all nodes in the cluster via Serf gossip

func (*DistributedEventBus) PublishLeader

func (bus *DistributedEventBus) PublishLeader(namespace, topic string, payload map[string]interface{}, tags []string) error

PublishLeader publishes an event that only the leader can send via Raft

func (*DistributedEventBus) PublishLocal

func (bus *DistributedEventBus) PublishLocal(namespace, topic string, payload map[string]interface{}, tags []string) error

PublishLocal publishes an event only on the local node

func (*DistributedEventBus) Subscribe

func (bus *DistributedEventBus) Subscribe(topic string, handler EventHandler) Subscription

Subscribe registers a handler for a specific topic

func (*DistributedEventBus) Unsubscribe

func (bus *DistributedEventBus) Unsubscribe(sub *Subscription)

Unsubscribe removes a handler

type Event

type Event struct {
	ID        string                 `json:"id"`
	Topic     string                 `json:"topic"`
	Namespace string                 `json:"namespace"`
	Tags      []string               `json:"tags"`
	Payload   map[string]interface{} `json:"payload"`
	NodeID    string                 `json:"node_id"`
	Timestamp time.Time              `json:"timestamp"`
}

Event represents a distributed event

type EventBus

type EventBus interface {
	Subscribe(topic string, handler EventHandler) Subscription
	PublishLocal(namespace, topic string, payload map[string]interface{}, tags []string) error
	PublishCluster(namespace, topic string, payload map[string]interface{}, tags []string) error
	PublishLeader(namespace, topic string, payload map[string]interface{}, tags []string) error
}

EventBus defines the interface for the event bus

type EventHandler

type EventHandler func(Event)

EventHandler is a function that handles events

type EventRouter

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

EventRouter determines routing strategy based on topic

func NewEventRouter

func NewEventRouter() *EventRouter

NewEventRouter creates a new event router with default rules

func (*EventRouter) AddRule

func (r *EventRouter) AddRule(topicPrefix string, strategy RoutingStrategy)

AddRule adds a routing rule for a topic prefix

func (*EventRouter) GetStrategy

func (r *EventRouter) GetStrategy(topic string) RoutingStrategy

GetStrategy returns the routing strategy for a given topic

func (*EventRouter) RequiresLeader

func (r *EventRouter) RequiresLeader(topic string) bool

RequiresLeader returns true if only the leader can publish this event

func (*EventRouter) ShouldReplicate

func (r *EventRouter) ShouldReplicate(topic string) bool

ShouldReplicate returns true if the event should be sent to other nodes

type RoutingStrategy

type RoutingStrategy int

RoutingStrategy determines how events are distributed

const (
	// RouteLocal - Event stays on local node only
	RouteLocal RoutingStrategy = iota
	// RouteGossip - Event is gossiped to all cluster nodes
	RouteGossip
	// RouteRaft - Event requires Raft consensus (leader only)
	RouteRaft
	// RouteDirect - Event sent directly to specific node
	RouteDirect
)

type Subscription

type Subscription struct {
	ID    string
	Topic string
	// contains filtered or unexported fields
}

Subscription represents an active event subscription

func (*Subscription) Unsubscribe

func (s *Subscription) Unsubscribe()

Unsubscribe removes the subscription