go-event是一个在Docker我的项目中应用到的一个事件散发组件，实现了惯例的播送，队列等事件散发模型，代码简洁明了，也适宜初学者对Go语言的入门，对channel用来同步，通信也会加深了解。

外围数据结构

Event

type Event interface{}

Event被封装为一个空接口，承受任意类型。在go-events示意一个能够被散发的事件。

interface{}的底层相似于c语言中的void*,但比void*弱小很多，比方interface{}保留了指向对象的指针和类型，而c程序员应用void*时，必须本人去保障对象的类型是正确的）

Sink

type Sink interface {
    Write(event Event) error
    Close() error
}

Sink是一个用来散发事件（Event）的构造。能够当作事件的解决者，应用接口的形式申明。只有对象实现了这两个办法，就能够被当作一个Sink。
核型办法

  • Write(event Event) error

    • 定义了事件如何被散发的策略。

  • Close() error

    • 当Sink被敞开的解决策略。

go-event外围就是围绕Sink做文章，docker官网给出了一个http的例子，就是当调用Write时，发动一次post申请。：

func (h *httpSink) Write(event Event) error {
    p, err := json.Marshal(event)
    if err != nil {
        return err
    }
    body := bytes.NewReader(p)
    resp, err := h.client.Post(h.url, "application/json", body)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    
    if resp.Status != 200 {
        return errors.New("unexpected status")
    }

    return nil
}

// implement (*httpSink).Close()
实现模型

到此为止，sink定义了事件散发的根本单位。在go-event中，封装了播送，音讯队列两种音讯散发的模型，具体来说，就是实现了Sink接口的两个构造体。

Boadcaster

type Broadcaster struct {
    sinks   []Sink //所蕴含的Sink
    events  chan Event// 同步Event的channel
    adds    chan configureRequest //adds和remove必须保障thread-safe，所以采纳channel同步
    removes chan configureRequest

    shutdown chan struct{}
    closed   chan struct{}
    once     sync.Once
}

Boardcaster由多个Sink组成，当Boardcaster接管到一个事件时，会调用本身蕴含的所有Sink的Write()办法
go-events设计之初就实现协程之间的音讯散发，须要保障thread-safe，所以对event的解决，增加，移除Sink都应用管道来通信。这也是Go的一个应用准则：

应用通信来共享内存,而不是通过共享内存来通信

在Broadcaster中所有的临界资源(sinks,event)都通过本身的run()函数对立治理，外界则通过相应的channel 同步给Broadcaster
例如Write()

func (b *Broadcaster) Write(event Event) error {
    select {
    case b.events <- event:
    case <-b.closed:
        return ErrSinkClosed
    }
    return nil
}

能够看到增减sink都是通过向对应的channel写入数据进行的。

func (b *Broadcaster) Add(sink Sink) error {
    return b.configure(b.adds, sink) //  will be block until ch can be writen
}

func (b *Broadcaster) configure(ch chan configureRequest, sink Sink) error {
    response := make(chan error, 1)

    for {
        select {
        case ch <- configureRequest{
            sink:     sink,
            response: response}:
            ch = nil // ？
        case err := <-response: 
            return err
        case <-b.closed:
            return ErrSinkClosed
        }
    }
}

外围run函数的实现,监听Boardcast管道上的相应事件，并作出解决。

func (b *Broadcaster) run() {
    defer close(b.closed)
    //将remove封装了一下，因为上面两处都会用到
    remove := func(target Sink) {
        for i, sink := range b.sinks {
            if sink == target {
                b.sinks = append(b.sinks[:i], b.sinks[i+1:]...)
                break
            }
        }
    }
    // 轮训解决channel上的事件
    for {
        select {
        case event := <-b.events: //有事件到来，进行播送
            for _, sink := range b.sinks {
                if err := sink.Write(event); err != nil {
                    if err == ErrSinkClosed {
                        // remove closed sinks
                        remove(sink)
                        continue
                    }
                    logrus.WithField("event", event).WithField("events.sink", sink).WithError(err).
                        Errorf("broadcaster: dropping event")
                }
            }
        case request := <-b.adds: //减少sink事件
            // while we have to iterate for add/remove, common iteration for
            // send is faster against slice.

            var found bool
            for _, sink := range b.sinks {
                if request.sink == sink {
                    found = true
                    break
                }
            }

            if !found {
                b.sinks = append(b.sinks, request.sink)
            }
            // b.sinks[request.sink] = struct{}{}
            request.response <- nil // 唤醒阻塞的configure（）函数
            
        case request := <-b.removes://删除sink事件
            remove(request.sink)
            request.response <- nil
        case <-b.shutdown:
            // close all the underlying sinks
            for _, sink := range b.sinks {
                if err := sink.Close(); err != nil && err != ErrSinkClosed {
                    logrus.WithField("events.sink", sink).WithError(err).
                        Errorf("broadcaster: closing sink failed")
                }
            }
            return
        }
    }
}
queue

queue应用contaienr/list实现了典型的生产消费者模型

type Queue struct {
    dst    Sink
    events *list.List
    cond   *sync.Cond 
    mu     sync.Mutex
    closed bool
}

外围函数run(),在队列中取出下一个event，交给本身的sink解决，在没有事件队列的状况下，eq.next()总是阻塞的(应用条件变量进行同步)

func (eq *Queue) run() {
    for {
        event := eq.next()

        if event == nil {
            return // nil block means event queue is closed.
        }

        if err := eq.dst.Write(event); err != nil {
            logrus.WithFields(logrus.Fields{
                "event": event,
                "sink":  eq.dst,
            }).WithError(err).Debug("eventqueue: dropped event")
        }
    }
}

生产者:q.next()
消费者:write()

func (eq *Queue) Write(event Event) error {
    eq.mu.Lock()
    defer eq.mu.Unlock()

    if eq.closed {
        return ErrSinkClosed
    }

    eq.events.PushBack(event)
    eq.cond.Signal() // signal waiters

    return nil
}

func (eq *Queue) next() Event {
    eq.mu.Lock()
    defer eq.mu.Unlock()

    for eq.events.Len() < 1 {
        if eq.closed {
            eq.cond.Broadcast()
            return nil
        }

        eq.cond.Wait()
    }

    front := eq.events.Front()
    block := front.Value.(Event)
    eq.events.Remove(front)

    return block
}