event

Documentation

Overview

Package event 事件转发基本的生产消费模型

Example

ctx, done := context.WithCancel(context.Background())
defer done()

bus := NewBus(ctx)

makeDoneHandler := func(label string) Handler {
	return func(ctx context.Context, e Event) error {
		fmt.Printf("%s handler called\n", label)
		return nil
	}
}

bus.SubscribeTypes(makeDoneHandler("first"), ETMainSaidHello, ETMainOpSucceeded)
bus.SubscribeTypes(makeDoneHandler("second"), ETMainSaidHello)
bus.SubscribeTypes(makeDoneHandler("third"), ETMainSaidHello)

_ = bus.Publish(ctx, ETMainSaidHello, "hello")
_ = bus.Publish(ctx, ETMainOpSucceeded, "operation worked!")

Output:

first handler called
second handler called
third handler called
first handler called

Index

• Constants
• Variables
• func AddIDToContext(ctx context.Context, s string) context.Context
• func GetIDFromCtx(ctx context.Context) string
• type Bus
  • func NewBus(ctx context.Context) Bus
• type CtxKey
• type DataEvent
• type Event
  • func MakeEvent(typ Type, sessionID string, payload interface{}) Event
• type Handler
• type OplogEvent
• type PrintEvent
• type Publisher
• type TaskEvent
• type Type

Examples

• Package

Constants

const (
	ETOplog = Type("op:Oplog")
	ETPrint = Type("op:Print")
	ETTask  = Type("op:Task")
	ETData  = Type("op:Data")
)

Variables

var (
	// ErrBusClosed 事件关闭
	ErrBusClosed = fmt.Errorf("event bus is closed")
	// NowFunc 时间戳生成
	NowFunc = time.Now
)

var NilBus = nilBus{}

NilBus 空接口实现

Functions

func AddIDToContext

func AddIDToContext(ctx context.Context, s string) context.Context

AddIDToContext 添加到上下文

func GetIDFromCtx

func GetIDFromCtx(ctx context.Context) string

GetIDFromCtx 从上下文取

Types

type Bus

type Bus interface {
	// Publish 发布事件
	Publish(ctx context.Context, typ Type, data interface{}) error
	// PublishID 发布ID
	PublishID(ctx context.Context, typ Type, sessionID string, data interface{}) error
	// Subscribe 订阅一个或者多个事件类型
	SubscribeTypes(handler Handler, eventTypes ...Type)
	// SubscribeID 订阅某个ID的事件
	SubscribeID(handler Handler, sessionID string)
	// SubscribeAll 订阅所有事件
	SubscribeAll(handler Handler)
	// NumSubscriptions 返回订阅者数量
	NumSubscribers() int
}

Bus 事件订阅生产转发集中处理

func NewBus

func NewBus(ctx context.Context) Bus

NewBus creates a new event bus. Event busses should be instantiated as a singleton. If the passed in context is cancelled, the bus will stop emitting events and close all subscribed channels

type CtxKey

type CtxKey string

CtxKey 声明一个唯一key

const OpCtxKey CtxKey = "OpCtxKey"

OpCtxKey 上下文唯一key

type DataEvent

type DataEvent struct {
	ID   string                 `json:"id"`
	Data map[string]interface{} `json:"data"`
}

DataEvent kv数据存档事件

type Event

type Event struct {
	Type      Type
	Timestamp int64
	SessionID string
	Payload   interface{}
}

Event 事件消息体

func MakeEvent

func MakeEvent(typ Type, sessionID string, payload interface{}) Event

MakeEvent 产生event

type Handler

type Handler func(ctx context.Context, e Event) error

Handler 事件处理方法定义

type OplogEvent

type OplogEvent struct {
	ID         string                 `json:"id"`
	Ctx        map[string]interface{} `json:"ctx"`
	RemoteAddr string                 `json:"remoteAddr"`
	Name       string                 `json:"name"`
	Status     string                 `json:"status"`
	StartTime  int64                  `json:"start_time"`
	EndTime    int64                  `json:"end_time"`
	TimeUsed   int64                  `json:"time_used"`
	Details    []string               `json:"detail"`
	Exception  string                 `json:"exception,omitempty"`
}

OplogEvent op相关的event

type PrintEvent

type PrintEvent struct {
	ID  string `json:"id"`
	Msg string `json:"message"`
}

PrintEvent 打印事件

type Publisher

type Publisher interface {
	Publish(ctx context.Context, typ Type, payload interface{}) error
	PublishID(ctx context.Context, typ Type, sessionID string, payload interface{}) error
}

Publisher 发布器接口声明

type TaskEvent

type TaskEvent struct {
	ID   string `json:"id"`
	From string `json:"from"`
	To   string `json:"to"`
}

TaskEvent 任务事件

type Type

type Type string

Type 订阅的事件类型