chore(example48): add publish and subscribe.

Signed-off-by: Bo-Yi Wu <[email protected]>

example48-pub-sub-pattern/hub.go

@@ -10,16 +10,42 @@ type hub struct {
 	subs map[*subscriber]struct{}
 }
 
+func (h *hub) publish(ctx context.Context, msg *message) error {
+	h.Lock()
+	for s := range h.subs {
+		s.publish(ctx, msg)
+	}
+	h.Unlock()
+
+	return nil
+}
+
 func (h *hub) subscribe(ctx context.Context, s *subscriber) error {
 	h.Lock()
 	h.subs[s] = struct{}{}
 	h.Unlock()
 
+	go func() {
+		select {
+		case <-ctx.Done():
+			h.Lock()
+			delete(h.subs, s)
+			h.Unlock()
+		}
+	}()
+
 	go s.run(ctx)
 
 	return nil
 }
 
+func (h *hub) Subscribers() int {
+	h.Lock()
+	c := len(h.subs)
+	h.Unlock()
+	return c
+}
+
 func newHub() *hub {
 	return &hub{
 		subs: map[*subscriber]struct{}{},

example48-pub-sub-pattern/main.go

@@ -1,6 +1,9 @@
 package main
 
-import "context"
+import (
+	"context"
+	"time"
+)
 
 func main() {
 	ctx := context.Background()
@@ -12,4 +15,10 @@ func main() {
 	h.subscribe(ctx, sub01)
 	h.subscribe(ctx, sub02)
 	h.subscribe(ctx, sub03)
+
+	_ = h.publish(ctx, &message{data: []byte("test01")})
+	_ = h.publish(ctx, &message{data: []byte("test02")})
+	_ = h.publish(ctx, &message{data: []byte("test03")})
+
+	time.Sleep(2 * time.Second)
 }

example48-pub-sub-pattern/subscriber.go

@@ -22,7 +22,7 @@ func (s *subscriber) run(ctx context.Context) {
 	for {
 		select {
 		case msg := <-s.handler:
-			log.Println(msg.data)
+			log.Println(s.name, string(msg.data))
 		case <-s.quit:
 			return
 		case <-ctx.Done():
@@ -31,6 +31,15 @@ func (s *subscriber) run(ctx context.Context) {
 	}
 }
 
+func (s *subscriber) publish(ctx context.Context, msg *message) {
+	select {
+	case <-ctx.Done():
+		return
+	case s.handler <- msg:
+	default:
+	}
+}
+
 func newSubscriber(name string) *subscriber {
 	return &subscriber{
 		name:    name,