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"
)

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

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

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

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

	// subscribe topic "greet1"
	em.Subscribe(suber2, "greet1", func(subscriber event_emitter.Subscriber[any], msg any) {
		fmt.Printf("recv: sub_id=%d, msg=%v\n", subscriber.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 Socket struct{ *gws.Conn }

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

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

func Sub(em *event_emitter.EventEmitter[*Socket], topic string, socket *Socket) {
	em.Subscribe(socket, topic, func(subscriber *Socket, msg any) {
		_ = msg.(*gws.Broadcaster).Broadcast(subscriber.Conn)
	})
}

func Pub(em *event_emitter.EventEmitter[*Socket], topic string, op gws.Opcode, msg []byte) {
	var broadcaster = gws.NewBroadcaster(op, msg)
	defer broadcaster.Close()
	em.Publish(topic, broadcaster)
}

Documentation

Index

• type Config
• type EventEmitter
  • func New[T Subscriber[T]](conf *Config) *EventEmitter[T]
  • func (c *EventEmitter[T]) CountSubscriberByTopic(topic string) int
  • func (c *EventEmitter[T]) GetTopicsBySubscriber(suber T) []string
  • func (c *EventEmitter[T]) NewSubscriber() Subscriber[any]
  • func (c *EventEmitter[T]) Publish(topic string, msg any)
  • func (c *EventEmitter[T]) Subscribe(suber T, topic string, f func(subscriber T, msg any))
  • func (c *EventEmitter[T]) UnSubscribe(suber T, topic string)
  • func (c *EventEmitter[T]) UnSubscribeAll(suber T)
• type Int64Subscriber
  • func (c *Int64Subscriber) GetMetadata() Metadata
  • func (c *Int64Subscriber) GetSubscriberID() int64
• 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 Subscriber[T]] struct {
	// contains filtered or unexported fields
}

func New

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

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

func (*EventEmitter[T]) CountSubscriberByTopic

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

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

func (*EventEmitter[T]) GetTopicsBySubscriber

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

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

func (*EventEmitter[T]) NewSubscriber

func (c *EventEmitter[T]) NewSubscriber() Subscriber[any]

NewSubscriber 生成订阅ID. 也可以使用自己的ID, 保证唯一即可. Generate a subscription ID. You can also use your own ID, just make sure it's unique.

func (*EventEmitter[T]) Publish

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

Publish 向主题发布消息 Publish a message to the topic

func (*EventEmitter[T]) Subscribe

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

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

func (*EventEmitter[T]) UnSubscribe

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

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

func (*EventEmitter[T]) UnSubscribeAll

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

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

type Int64Subscriber

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

func (*Int64Subscriber) GetMetadata

func (c *Int64Subscriber) GetMetadata() Metadata

func (*Int64Subscriber) GetSubscriberID

func (c *Int64Subscriber) GetSubscriberID() int64

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 any] interface {
	GetSubscriberID() int64 // 获取订阅者唯一ID
	GetMetadata() Metadata
}