event_emitter

README

EventEmitter

Simple, Fast and Thread-Safe Event Pub/Sub Library for Golang

Introduction

EventEmitter is a simple, fast and thread-safe event Pub/Sub library for Golang.

EventEmitter only supports in memory storage, and does not support persistence, so it is not suitable for scenarios that require persistence. It can only be used in the application itself, does not support distribution, if you need distribution, please use RabittMQ, Kafka etc.

EventEmitter designed for scenarios that require high concurrency and high performance.

Install

go get -v github.com/lxzan/event_emitter@latest

Quick Start

It very easy to use, just create a event emitter, then subscribe and publish messages. It can be used in any place of your application.

The following is a simple example.

package main

import (
	"fmt"
	"github.com/lxzan/event_emitter"
	"sync/atomic"
)

func main() {
	// create a event emitter
	var em = event_emitter.New[int64, event_emitter.Subscriber[int64]](&event_emitter.Config{
		BucketNum:  16,
		BucketSize: 128,
	})

	var increaser = new(atomic.Int64)

	// create a subscriber
	var suber1 = em.NewSubscriber(increaser.Add(1))

	// subscribe topic "greet"
	em.Subscribe(suber1, "greet", func(msg any) {
		fmt.Printf("recv: sub_id=%d, msg=%v\n", suber1.GetSubscriberID(), msg)
	})
	// subscribe topic "greet1"
	em.Subscribe(suber1, "greet1", func(msg any) {
		fmt.Printf("recv: sub_id=%d, msg=%v\n", suber1.GetSubscriberID(), msg)
	})

	// create another subscriber
	var suber2 = em.NewSubscriber(increaser.Add(1))

	// subscribe topic "greet1"
	em.Subscribe(suber2, "greet1", func(msg any) {
		fmt.Printf("recv: sub_id=%d, msg=%v\n", suber2.GetSubscriberID(), msg)
	})

	// publish message to topic "greet"
	em.Publish("greet1", "hello!")
}

More Examples

GWS Broadcast

Use the event_emitter package to implement the publish-subscribe model. Wrap gws.Conn in a structure and implement the GetSubscriberID method to get the subscription ID, which must be unique. The subscription ID is used to identify the subscriber, who can only receive messages on the subject of his subscription.

This example is useful for building chat rooms or push messages using gws. This means that a user can subscribe to one or more topics via websocket, and when a message is posted to that topic, all subscribers will receive the message.

package main

import (
	"github.com/lxzan/event_emitter"
	"github.com/lxzan/gws"
)

type Subscriber gws.Conn

func NewSubscriber(conn *gws.Conn) *Subscriber { return (*Subscriber)(conn) }

func (c *Subscriber) GetSubscriberID() int64 {
	userId, _ := c.GetMetadata().Load("userId")
	return userId.(int64)
}

func (c *Subscriber) GetMetadata() event_emitter.Metadata { return c.Conn().Session() }

func (c *Subscriber) Conn() *gws.Conn { return (*gws.Conn)(c) }

func Subscribe(em *event_emitter.EventEmitter[int64, *Subscriber], s *Subscriber, topic string) {
	em.Subscribe(s, topic, func(msg any) {
		_ = msg.(*gws.Broadcaster).Broadcast(s.Conn())
	})
}

func Publish(em *event_emitter.EventEmitter[int64, *Subscriber], topic string, msg []byte) {
	var broadcaster = gws.NewBroadcaster(gws.OpcodeText, msg)
	defer broadcaster.Close()
	em.Publish(topic, broadcaster)
}

Documentation

Index

• type Config
• type EventEmitter
  • func New[T comparable, S Subscriber[T]](conf *Config) *EventEmitter[T, S]
  • func (c *EventEmitter[T, S]) CountSubscriberByTopic(topic string) int
  • func (c *EventEmitter[T, S]) GetTopicsBySubscriber(suber S) []string
  • func (c *EventEmitter[T, S]) NewSubscriber(id T) Subscriber[T]
  • func (c *EventEmitter[T, S]) Publish(topic string, msg any)
  • func (c *EventEmitter[T, S]) Subscribe(suber S, topic string, f func(msg any))
  • func (c *EventEmitter[T, S]) UnSubscribe(suber S, topic string)
  • func (c *EventEmitter[T, S]) UnSubscribeAll(suber S)
• type Metadata
• type Subscriber

Types

type Config

type Config struct {
	// 分片数
	// Number of slices
	BucketNum int64

	// 每个分片里主题表的初始化容量, 根据主题订阅量估算, 默认为0.
	// The initialization capacity of the topic table in each slice is estimated based on the topic subscriptions and is 0 by default.
	BucketSize int64
}

type EventEmitter

type EventEmitter[T comparable, S Subscriber[T]] struct {
	// contains filtered or unexported fields
}

EventEmitter T: 订阅ID模板类型 S: 订阅者模板类型 T:subscription id template type S:subscriber template type

func New

func New[T comparable, S Subscriber[T]](conf *Config) *EventEmitter[T, S]

New 创建事件发射器实例 Creating an EventEmitter Instance

func (*EventEmitter[T, S]) CountSubscriberByTopic

func (c *EventEmitter[T, S]) CountSubscriberByTopic(topic string) int

CountSubscriberByTopic 获取主题订阅人数 Get the number of subscribers to a topic

func (*EventEmitter[T, S]) GetTopicsBySubscriber

func (c *EventEmitter[T, S]) GetTopicsBySubscriber(suber S) []string

GetTopicsBySubscriber 通过订阅者获取主题列表 Get a list of topics by subscriber

func (*EventEmitter[T, S]) NewSubscriber

func (c *EventEmitter[T, S]) NewSubscriber(id T) Subscriber[T]

NewSubscriber 创建一个订阅者, 订阅Id需要保证唯一. Create a subscriber, the subscription id needs to be unique.

func (*EventEmitter[T, S]) Publish

func (c *EventEmitter[T, S]) Publish(topic string, msg any)

Publish 发布消息到指定主题 Post a message to the topic.

func (*EventEmitter[T, S]) Subscribe

func (c *EventEmitter[T, S]) Subscribe(suber S, topic string, f func(msg any))

Subscribe 订阅主题消息. 注意: 回调函数必须是非阻塞的. Subscribe messages from the topic. Note: Callback functions must be non-blocking.

func (*EventEmitter[T, S]) UnSubscribe

func (c *EventEmitter[T, S]) UnSubscribe(suber S, topic string)

UnSubscribe 取消订阅一个主题 Cancel a subscribed topic

func (*EventEmitter[T, S]) UnSubscribeAll

func (c *EventEmitter[T, S]) UnSubscribeAll(suber S)

UnSubscribeAll 取消订阅所有主题 Cancel all subscribed topics

type Metadata

type Metadata interface {
	Load(key string) (value any, exist bool)
	Store(key string, value any)
	Delete(key string)
	Range(f func(key string, value any) bool)
}

type Subscriber

type Subscriber[T comparable] interface {
	// GetSubscriberID 获取订阅者唯一ID
	// Get subscriber unique ID
	GetSubscriberID() T

	// GetMetadata 获取元数据
	// Getting Metadata
	GetMetadata() Metadata
}