eventbus

package module
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 14, 2021 License: Apache-2.0 Imports: 21 Imported by: 0

README

概述

事件总线。使用事件驱动的方式进行业务解耦。
建议使用集群的方式运行，这样可以支持编程多语言环境。

获取

go get github.com/aacfactory/eventbus

使用

本地

// 构建
eb := eventbus.NewEventbus()

// 挂载处理器

type Arg struct {
    Id       string    `json:"id,omitempty"`
    Num      int       `json:"num,omitempty"`
    Datetime time.Time `json:"datetime,omitempty"`
}

type Result struct {
    Value string `json:"value,omitempty"`
}

func HandlerReply(head eventbus.MultiMap, body []byte) (result interface{}, err error) {
    arg := &Arg{}
    _ = json.Unmarshal(body, arg)
    fmt.Println("handle reply", head, arg)
    if arg.Num < 0 {
        err = errors.InvalidArgumentErrorWithDetails("bad number", "num", "less than 0")
        return
    }
    result = &Result{
        Value: "result",
    }
    return
}

func HandlerVoid(head eventbus.MultiMap, body []byte) (result interface{}, err error) {
    arg := &Arg{}
    _ = json.Unmarshal(body, arg)
    fmt.Println("handle void", head, arg)
    return
}

_ = eb.RegisterHandler("void", HandlerVoid)
_ = eb.RegisterHandler("reply", HandlerReply)

// 启动 （启动必须晚于挂载处理器）
eb.Start(context.TODO())

// 执行
options := eventbus.NewDeliveryOptions()
options.Add("h1", "1")
options.Add("h2", "2")

sendErr := eb.Send("reply", &Arg{
    Id:       "id",
    Num:      10,
    Datetime: time.Now(),
}, options)

if sendErr != nil {
fmt.Println("send failed", sendErr)
}

for i := 0; i < 2; i++ {
    rf := eb.Request("reply", &Arg{
        Id:       "id",
        Num:      i - 1,
        Datetime: time.Now(),
    }, options)
    result := &Result{}
    requestErr := rf.Result(result)
    if requestErr != nil {
    	fmt.Println("request failed", requestErr)
    } else {
        fmt.Println("request succeed", result)
    }
}

// 优雅的关闭
eb.Close(context.TODO())

集群

TCP + 地址发现 模式

当前没有实现服务发现，请使用 aacfactory/cluster ，或自行实现。

// options
options := eventbus.ClusterEventbusOption{
    Host:                       "0.0.0.0", // 实际监听地址
    Port:                       9090, // 实际监听端口
    PublicHost:                 "127.0.0.1", // 注册地址，如果为空，则默认使用监听地址
    PublicPort:                 0, // 注册端口，如果为空，则默认使用监听端口
    Meta:                       &eventbus.EndpointMeta{}, // 注册源数据
    Tags:                       nil, // 标签，一般用于版本化与运行隔离化
    TLS:                        &eventbus.EndpointTLS{}, // TLS 配置
    EventChanCap:               64, // 事件 chan 的长度
    EventHandlerInstanceNumber: 2,  // 事件处理器的实例个数
    EnableLocal:                true, // 是否开启本地处理器，如果关闭，则直接使用远程模式，建议开启
}
// discovery
discovery := Foo{}
// 创建
bus, err = eventbus.NewClusterEventbus(discovery, options)
if err != nil {
    return
}
// 操作与本地Eventbus一样
NATS 模型
// todo

Documentation

Index

Constants

View Source 
const (
	EndpointStatusRunning = EndpointStatus("RUNNING")
	EndpointStatusClosing = EndpointStatus("CLOSING")
)

Variables

This section is empty.

Functions

func NewEventbusWithOption 

func NewEventbusWithOption(option LocaledEventbusOption) (eb *localedEventbus)

Types

type ClusterEventbusOption 

type ClusterEventbusOption struct {
	Host                       string        `json:"host,omitempty"`
	Port                       int           `json:"port,omitempty"`
	PublicHost                 string        `json:"publicHost,omitempty"`
	PublicPort                 int           `json:"publicPort,omitempty"`
	Meta                       *EndpointMeta `json:"meta,omitempty"`
	Tags                       []string      `json:"tags,omitempty"`
	TLS                        *EndpointTLS  `json:"tls,omitempty"`
	EventChanCap               int           `json:"eventChanCap,omitempty"`
	EventHandlerInstanceNumber int           `json:"eventHandlerInstanceNumber,omitempty"`
	EnableLocal                bool          `json:"enableLocal,omitempty"`
}

type DeliveryOptions 

type DeliveryOptions struct {
	MultiMap
}

func NewDeliveryOptions 

func NewDeliveryOptions() DeliveryOptions

func (*DeliveryOptions) AddTag 

func (options *DeliveryOptions) AddTag(tags ...string)

type EndpointMeta 

type EndpointMeta map[string]string

func NewEndpointMeta 

func NewEndpointMeta() EndpointMeta

func (EndpointMeta) Empty 

func (meta EndpointMeta) Empty() bool

func (EndpointMeta) Get 

func (meta EndpointMeta) Get(key string) (string, bool)

func (EndpointMeta) Keys 

func (meta EndpointMeta) Keys() []string

func (EndpointMeta) Merge 

func (meta EndpointMeta) Merge(o ...Meta)

func (EndpointMeta) Put 

func (meta EndpointMeta) Put(key string, value string)

func (EndpointMeta) Rem 

func (meta EndpointMeta) Rem(key string)

type EndpointStatus 

type EndpointStatus string

func (EndpointStatus) Closing 

func (s EndpointStatus) Closing() bool

func (EndpointStatus) Ok 

func (s EndpointStatus) Ok() bool

type EndpointTLS 

type EndpointTLS struct {
	Enable_     bool   `json:"enable,omitempty"`
	VerifySSL_  bool   `json:"verifySsl,omitempty"`
	CA_         string `json:"ca,omitempty"`
	ServerCert_ string `json:"serverCert,omitempty"`
	ServerKey_  string `json:"serverKey,omitempty"`
	ClientCert_ string `json:"clientCert,omitempty"`
	ClientKey_  string `json:"clientKey,omitempty"`
}

func (EndpointTLS) CA 

func (s EndpointTLS) CA() string

func (EndpointTLS) ClientCert 

func (s EndpointTLS) ClientCert() string

func (EndpointTLS) ClientKey 

func (s EndpointTLS) ClientKey() string

func (EndpointTLS) Enable 

func (s EndpointTLS) Enable() bool

func (EndpointTLS) ServerCert 

func (s EndpointTLS) ServerCert() string

func (EndpointTLS) ServerKey 

func (s EndpointTLS) ServerKey() string

func (EndpointTLS) ToClientTLSConfig 

func (s EndpointTLS) ToClientTLSConfig() (config *tls.Config, err error)

func (EndpointTLS) ToServerTLSConfig 

func (s EndpointTLS) ToServerTLSConfig() (config *tls.Config, err error)

func (EndpointTLS) VerifySSL 

func (s EndpointTLS) VerifySSL() bool

type EventHandler 

type EventHandler func(head MultiMap, body []byte) (result interface{}, err error)

type Eventbus 

type Eventbus interface {
	Send(address string, v interface{}, options ...DeliveryOptions) (err error)
	Request(address string, v interface{}, options ...DeliveryOptions) (reply *ReplyFuture)
	RegisterHandler(address string, handler EventHandler, tags ...string) (err error)
	Start(context context.Context)
	Close(context context.Context)
}

func NewClusterEventbus 

func NewClusterEventbus(discovery ServiceDiscovery, option ClusterEventbusOption) (bus Eventbus, err error)

func NewEventbus 

func NewEventbus() Eventbus

type LocaledEventbusOption 

type LocaledEventbusOption struct {
	EventChanCap               int
	EventHandlerInstanceNumber int
}

type Meta 

type Meta interface {
	Put(key string, value string)
	Get(key string) (value string, has bool)
	Rem(key string)
	Keys() (keys []string)
	Empty() (ok bool)
	Merge(o ...Meta)
}

type MultiMap 

type MultiMap map[string][]string

func (MultiMap) Add 

func (h MultiMap) Add(key string, value string)

func (MultiMap) Empty 

func (h MultiMap) Empty() bool

func (MultiMap) Get 

func (h MultiMap) Get(key string) (string, bool)

func (MultiMap) Keys 

func (h MultiMap) Keys() []string

func (MultiMap) Merge 

func (h MultiMap) Merge(o ...MultiMap)

func (MultiMap) Put 

func (h MultiMap) Put(key string, value []string)

func (MultiMap) Remove 

func (h MultiMap) Remove(key string)

func (MultiMap) Values 

func (h MultiMap) Values(key string) ([]string, bool)

type Registration 

type Registration interface {
	NodeId() (nodeId string)
	NodeName() (nodeName string)
	Id() (id string)
	Group() (group string)
	Name() (name string)
	Status() (status Status)
	Protocol() (protocol string)
	Address() (address string)
	Tags() (tags []string)
	Meta() (meta Meta)
	TLS() (registrationTLS RegistrationTLS)
}

type RegistrationTLS 

type RegistrationTLS interface {
	Enable() bool
	VerifySSL() bool
	CA() string
	ServerCert() string
	ServerKey() string
	ClientCert() string
	ClientKey() string
	ToServerTLSConfig() (config *tls.Config, err error)
	ToClientTLSConfig() (config *tls.Config, err error)
}

type ReplyError 

type ReplyError struct {
	Id          string          `json:"id,omitempty"`
	FailureCode int             `json:"failureCode,omitempty"`
	Code        string          `json:"code,omitempty"`
	Message     string          `json:"message,omitempty"`
	Meta        errors.MultiMap `json:"meta,omitempty"`
}

func (*ReplyError) Error 

func (e *ReplyError) Error() string

func (*ReplyError) GetMeta 

func (e *ReplyError) GetMeta() errors.MultiMap

func (*ReplyError) GetStacktrace 

func (e *ReplyError) GetStacktrace() (fn string, file string, line int)

func (*ReplyError) SetFailureCode 

func (e *ReplyError) SetFailureCode(failureCode int) errors.CodeError

func (*ReplyError) SetId 

func (e *ReplyError) SetId(id string) errors.CodeError

func (*ReplyError) String 

func (e *ReplyError) String() string

func (*ReplyError) ToJson 

func (e *ReplyError) ToJson() []byte

type ReplyFuture 

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

func (*ReplyFuture) Result 

func (r *ReplyFuture) Result(v interface{}) (err error)

type ServiceDiscovery 

type ServiceDiscovery interface {
	Publish(group string, name string, protocol string, address string, tags []string, meta Meta, registrationTLS RegistrationTLS) (registration Registration, err error)
	UnPublish(registration Registration) (err error)
	Get(group string, name string, tags ...string) (registration Registration, has bool, err error)
	GetALL(group string, name string, tags ...string) (registrations []Registration, has bool, err error)
}

type Status 

type Status interface {
	Ok() bool
	Closing() bool
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.