broadcast_manager.go

package main

import (
	"context"
	"sync"
)

type BroadcastManager struct {
	sync.Mutex

	subs map[chan string]struct{}
}

func (b *BroadcastManager) Broadcast(text string) {
	b.Lock()
	defer b.Unlock()

	for ch := range b.subs {
		ch <- text
	}
}

func (b *BroadcastManager) Subscribe() (chan string, func()) {
	b.Lock()
	defer b.Unlock()

	ch := make(chan string)
	b.subs[ch] = struct{}{}

	return ch, func() {
		b.Lock()
		defer b.Unlock()

		delete(b.subs, ch)
	}
}

var instanceOfBm = &BroadcastManager{
	Mutex: sync.Mutex{},
	subs:  map[chan string]struct{}{},
}

type Sender interface {
	Send(string) error
}

func handleIncomingGRPCStreamRequest(ctx context.Context, s Sender) error {
	ch, cancel := instanceOfBm.Subscribe()
	defer cancel()

	for {
		select {
		case <-ctx.Done():
			break
		case msg := <-ch:
			if err := s.Send(msg); err != nil {
				return err
			}
		}
	}
}