package main
import "fmt"
// Subject 接口，它相當于是發(fā)布者的定義
type Subject interface {
	Subscribe(observer Observer)
	Notify(msg string)
}
// Observer 觀察者接口
type Observer interface {
	Update(msg string)
}
// Subject 實現(xiàn)
type SubjectImpl struct {
	observers []Observer
}
// Subscribe 添加觀察者（訂閱者）
func (sub *SubjectImpl) Subscribe(observer Observer) {
	sub.observers = append(sub.observers, observer)
}
// Notify 發(fā)布通知
func (sub *SubjectImpl) Notify(msg string) {
	for _, o := range sub.observers {
		o.Update(msg)
	}
}
// Observer1 Observer1
type Observer1 struct{}
// Update 實現(xiàn)觀察者接口
func (Observer1) Update(msg string) {
	fmt.Printf("Observer1: %s\n", msg)
}
// Observer2 Observer2
type Observer2 struct{}
// Update 實現(xiàn)觀察者接口
func (Observer2) Update(msg string) {
	fmt.Printf("Observer2: %s\n", msg)
}
func main(){
	sub := &SubjectImpl{}
	sub.Subscribe(&Observer1{})
	sub.Subscribe(&Observer2{})
	sub.Notify("Hello")
}

// 代碼來自https://lailin.xyz/post/observer.html
package eventbus
import (
	"fmt"
	"reflect"
	"sync"
)
// Bus Bus
type Bus interface {
	Subscribe(topic string, handler interface{}) error
	Publish(topic string, args ...interface{})
}
// AsyncEventBus 異步事件總線
type AsyncEventBus struct {
	handlers map[string][]reflect.Value
	lock     sync.Mutex
}
// NewAsyncEventBus new
func NewAsyncEventBus() *AsyncEventBus {
	return &AsyncEventBus{
		handlers: map[string][]reflect.Value{},
		lock:     sync.Mutex{},
	}
}
// Subscribe 訂閱
func (bus *AsyncEventBus) Subscribe(topic string, f interface{}) error {
	bus.lock.Lock()
	defer bus.lock.Unlock()
	v := reflect.ValueOf(f)
	if v.Type().Kind() != reflect.Func {
		return fmt.Errorf("handler is not a function")
	}
	handler, ok := bus.handlers[topic]
	if !ok {
		handler = []reflect.Value{}
	}
	handler = append(handler, v)
	bus.handlers[topic] = handler
	return nil
}
// Publish 發(fā)布
// 這里異步執(zhí)行，并且不會等待返回結(jié)果
func (bus *AsyncEventBus) Publish(topic string, args ...interface{}) {
	handlers, ok := bus.handlers[topic]
	if !ok {
		fmt.Println("not found handlers in topic:", topic)
		return
	}
	params := make([]reflect.Value, len(args))
	for i, arg := range args {
		params[i] = reflect.ValueOf(arg)
	}
	for i := range handlers {
		go handlers[i].Call(params)
	}
}

package eventbus
import (
	"fmt"
	"testing"
	"time"
)
func sub1(msg1, msg2 string) {
	time.Sleep(1 * time.Microsecond)
	fmt.Printf("sub1, %s %s\n", msg1, msg2)
}
func sub2(msg1, msg2 string) {
	fmt.Printf("sub2, %s %s\n", msg1, msg2)
}
func TestAsyncEventBus_Publish(t *testing.T) {
	bus := NewAsyncEventBus()
	bus.Subscribe("topic:1", sub1)
	bus.Subscribe("topic:1", sub2)
	bus.Publish("topic:1", "test1", "test2")
	bus.Publish("topic:1", "testA", "testB")
	time.Sleep(1 * time.Second)
}