events

Documentation

Overview

Package events provides a simple and effective implementation of event system.

Event is a battle proven way to decoupling services. Package event calls event listeners in a synchronous, sequential execution. The synchronous listener is only a "go" away from an asynchronous handler, but asynchronous listener can not be easily made synchronous.

The event listeners can also be used as hooks. If the event data is a pointer type, listeners may alter the data. This enables plugin/addon style decoupling.

Note: Package event focus on events within the system, not events outsource to eternal system. For that, use a message queue like kafka.

Example

package main

import (
	"context"
	"fmt"

	"github.com/DoNewsCode/core/events"
)

func main() {
	dispatcher := &events.SyncDispatcher{}

	// Subscribe to a string topic named foo.
	dispatcher.Subscribe(events.Listen("foo", func(ctx context.Context, event interface{}) error {
		fmt.Println(event)
		return nil
	}))

	// Subscribe to a struct topic.
	type Topic struct{}
	dispatcher.Subscribe(events.Listen(Topic{}, func(ctx context.Context, event interface{}) error {
		fmt.Println(event)
		return nil
	}))

	dispatcher.Dispatch(context.Background(), "foo", 100)
	dispatcher.Dispatch(context.Background(), Topic{}, "event")
}

Output:

100
event

Index

• Constants
• Variables
• type ListenerFunc
  • func Listen(topic interface{}, ...) *ListenerFunc
  • func (f *ListenerFunc) Listen() interface{}
  • func (f *ListenerFunc) Process(ctx context.Context, payload interface{}) error
• type OnReloadPayload
• type SyncDispatcher
  • func (d *SyncDispatcher) Dispatch(ctx context.Context, topic interface{}, payload interface{}) error
  • func (d *SyncDispatcher) ListenerCount(topic interface{}) int
  • func (d *SyncDispatcher) Prepend(listener contract.Listener)
  • func (d *SyncDispatcher) PrependOnce(listener contract.Listener)
  • func (d *SyncDispatcher) RemoveAllListeners(topic interface{})
  • func (d *SyncDispatcher) Subscribe(listener contract.Listener)
  • func (d *SyncDispatcher) SubscribeOnce(listener contract.Listener)
  • func (d *SyncDispatcher) Unsubscribe(listener contract.Listener) error

Examples

• Package

Constants

const OnReload event = "onReload"

OnReload is an event that triggers the configuration reloads. The event payload is OnReloadPayload.

Variables

var ErrNotSubscribed = errors.New("not subscribed")

ErrNotSubscribed is returned when trying to unsubscribe a listener that not subscribing.

Types

type ListenerFunc

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

ListenerFunc is a listener that can be constructed from one function Listen. It listens to the given topic and then execute the callback.

func Listen

func Listen(topic interface{}, callback func(ctx context.Context, payload interface{}) error) *ListenerFunc

Listen creates a functional listener in one line.

func (*ListenerFunc) Listen

func (f *ListenerFunc) Listen() interface{}

Listen implements contract.Listener

func (*ListenerFunc) Process

func (f *ListenerFunc) Process(ctx context.Context, payload interface{}) error

Process implements contract.Listener

type OnReloadPayload

type OnReloadPayload struct {
	// NewConf is the latest configuration after the reload.
	NewConf contract.ConfigAccessor
}

OnReload is an event that triggers the configuration reloads

type SyncDispatcher

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

SyncDispatcher is a contract.Dispatcher implementation that dispatches events synchronously. SyncDispatcher is safe for concurrent use.

func (*SyncDispatcher) Dispatch

func (d *SyncDispatcher) Dispatch(ctx context.Context, topic interface{}, payload interface{}) error

Dispatch dispatches events synchronously. If any listener returns an error, abort the process immediately and return that error to caller.

func (*SyncDispatcher) ListenerCount

func (d *SyncDispatcher) ListenerCount(topic interface{}) int

ListenerCount returns the number of listeners for a given event.

func (*SyncDispatcher) Prepend

func (d *SyncDispatcher) Prepend(listener contract.Listener)

Prepend adds the listener to the beginning of the listeners queue for the topic it listens to. The listeners will not be deduplicated. If subscribed more than once, the event will be added and processed more than once.

func (*SyncDispatcher) PrependOnce

func (d *SyncDispatcher) PrependOnce(listener contract.Listener)

PrependOnce adds a one-time listener function for the event it listens to, at the top of the listener queue waiting for the same event. The listener will be unsubscribed once after the event is processed by the listener .

func (*SyncDispatcher) RemoveAllListeners

func (d *SyncDispatcher) RemoveAllListeners(topic interface{})

RemoveAllListeners removes all listeners for a given event.

func (*SyncDispatcher) Subscribe

func (d *SyncDispatcher) Subscribe(listener contract.Listener)

Subscribe subscribes the listener to the dispatcher. The listeners will not be deduplicated. If subscribed more than once, the event will be added and processed more than once.

func (*SyncDispatcher) SubscribeOnce

func (d *SyncDispatcher) SubscribeOnce(listener contract.Listener)

SubscribeOnce subscribes the listener to the dispatcher and unsubscribe the listener once after the event is processed by the listener.

func (*SyncDispatcher) Unsubscribe

func (d *SyncDispatcher) Unsubscribe(listener contract.Listener) error

Unsubscribe unsubscribes the listener from the dispatcher. If the listener doesn't exist, ErrNotSubscribed will be returned. If there are multiple instance of the listener provided subscribed, only one of the will be unsubscribed.