Golang Subscriber Pattern

main.go

package main

import (
	"time"
)

type Event struct {
	data int
}

// Fib generator
func Fib(n int) chan int {
	out := make(chan int)

	go func() {
		defer close(out)
		for i, j := 0, 1; i < n; i, j = i+j, i {
			out <- i
		}
	}()

	return out
}

func main() {
	subscriber1 := NewObserver(1, time.Now())
	subscriber2 := NewObserver(2, time.Now())

	subject := NewSubject()
	subject.Attach(subscriber1)
	subject.Attach(subscriber2)

	go func() {
		select {
		case <- time.After(1 * time.Millisecond):
			subject.Detach(subscriber2)
		}
	}()

	for x := range Fib(1000000) {
		subject.Notify(Event{data: x})
	}
}

observer.go

package main

import (
	"fmt"
	"time"
)

type Observer interface {
	NotifyCallback(evt Event)
}

// concrete observer for fib sequenece
type FibObserver struct {
	id   int64     // current observer id
	time time.Time // join time?
}

// Observer
func NewObserver(id int64, time time.Time) *FibObserver {
	return &FibObserver{
		id,
		time,
	}
}

func (o *FibObserver) NotifyCallback(evt Event) {
	fmt.Println(o.id, "received event with data:", evt.data, "with time:", time.Since(o.time))
}

subject.go

package main

import "sync"

type Subject interface {
	Attach(o Observer)
	Detach(o Observer)
	Notify(evt Event)
}

// concrete subject for this data source
// needs to maintain an array of observers
type FibSubject struct {
	observers sync.Map
}

// Subject
func NewSubject() *FibSubject {
	return &FibSubject{
		observers: sync.Map{},
	}
}

func (s *FibSubject) Attach(o Observer) {
	s.observers.Store(o, struct{}{})
}

func (s *FibSubject) Detach(o Observer) {
	s.observers.Delete(o)
}

func (s *FibSubject) Notify(evt Event) {
	s.observers.Range(func(key interface{}, value interface{}) bool {
		if key == nil || value == nil {
			return false
		}

		key.(Observer).NotifyCallback(evt)
		return true
	})
}

> go run *.go
1 received event with data: 0 with time: 19.473µs
2 received event with data: 0 with time: 63.963µs
1 received event with data: 1 with time: 70.843µs
2 received event with data: 1 with time: 71.289µs
1 received event with data: 1 with time: 75.456µs
2 received event with data: 1 with time: 75.423µs
2 received event with data: 2 with time: 79.546µs
1 received event with data: 2 with time: 83.079µs
1 received event with data: 3 with time: 85.293µs
2 received event with data: 3 with time: 85.321µs
1 received event with data: 5 with time: 90.782µs
2 received event with data: 5 with time: 90.857µs
2 received event with data: 8 with time: 93µs
1 received event with data: 8 with time: 96.518µs
1 received event with data: 13 with time: 100.294µs
2 received event with data: 13 with time: 100.468µs
1 received event with data: 21 with time: 104.376µs
2 received event with data: 21 with time: 104.415µs
1 received event with data: 34 with time: 109.838µs
2 received event with data: 34 with time: 109.86µs
1 received event with data: 55 with time: 146.662µs
2 received event with data: 55 with time: 147.46µs
1 received event with data: 89 with time: 151.693µs
2 received event with data: 89 with time: 151.8µs
1 received event with data: 144 with time: 157.433µs
2 received event with data: 144 with time: 160.193µs
1 received event with data: 233 with time: 163.999µs
2 received event with data: 233 with time: 164.135µs
1 received event with data: 377 with time: 168.133µs
2 received event with data: 377 with time: 168.234µs
2 received event with data: 610 with time: 170.167µs
1 received event with data: 610 with time: 173.591µs
1 received event with data: 987 with time: 179.017µs
2 received event with data: 987 with time: 194.795µs
1 received event with data: 1597 with time: 201.122µs
2 received event with data: 1597 with time: 201.451µs
1 received event with data: 2584 with time: 205.759µs
2 received event with data: 2584 with time: 205.765µs
1 received event with data: 4181 with time: 209.308µs
1 received event with data: 6765 with time: 1.331078ms
1 received event with data: 10946 with time: 1.341888ms
1 received event with data: 17711 with time: 1.383208ms
1 received event with data: 28657 with time: 1.386603ms
1 received event with data: 46368 with time: 1.388587ms
1 received event with data: 75025 with time: 1.390722ms
1 received event with data: 121393 with time: 1.392524ms
1 received event with data: 196418 with time: 1.394749ms
1 received event with data: 317811 with time: 1.396612ms
1 received event with data: 514229 with time: 1.398799ms
1 received event with data: 832040 with time: 1.400659ms

Because the Notify is async, it cannot ensure the subscriber's order to run callback function.