broadcast.go

package kuiperbelt

import (
	"log"
	"sync"
)

type BroadcastNotifier struct {
	notifyChan    chan bool
	notifyingChan chan bool
	isNotifying   bool
	message       []byte
	cond          *sync.Cond
	wg            *sync.WaitGroup
	locker        *sync.Mutex
}

func NewBroadcastNotifier() *BroadcastNotifier {
	locker := new(sync.Mutex)
	cond := sync.NewCond(locker)
	wg := new(sync.WaitGroup)

	return &BroadcastNotifier{
		notifyChan:    make(chan bool),
		notifyingChan: make(chan bool),
		isNotifying:   false,
		message:       make([]byte, 0),
		locker:        locker,
		cond:          cond,
		wg:            wg,
	}
}
func (bn *BroadcastNotifier) ReceiveMessage(message []byte) {
	bn.message = message
	bn.notifyChan <- true
}

func (bn *BroadcastNotifier) runNotifying(uuid string) bool {
	_, ok := <-bn.notifyingChan
	connectorChan, existsConnector := connector.GetChan(uuid)
	if !existsConnector {
		return false
	}
	if ok || !bn.isNotifying {
		return true
	}
	connectorChan <- bn.message
	return true
}

func (bn *BroadcastNotifier) NotifyLoop(uuid string) {
	for {
		if !bn.runNotifying(uuid) {
			break
		}
	}
}

func (bn *BroadcastNotifier) Loop() {
	for {
		if !bn.runBroadcaster() {
			break
		}
	}
}

func (bn *BroadcastNotifier) runBroadcaster() bool {
	bn.locker.Lock()
	defer bn.locker.Unlock()
	_, ok := <-bn.notifyChan
	if !ok {
		log.Println("send destroy...")
		return false
	}
	bn.isNotifying = true
	close(bn.notifyingChan)
	log.Printf("sending of broadcast %s", bn.message)
	bn.isNotifying = false
	bn.notifyingChan = make(chan bool)
	log.Printf("end of broadcast %s", bn.message)
	return true
}