func (b *Broker) Close() {
	close(b.done)
}
Close stops the broker and releases its resources.
package inmemory
import (
	"errors"
	"log"
	"sync"
	"github.com/genkami/kiara/types"
)
var (
	ErrAlreadySubscribed = errors.New("already subscribed")
	ErrNotSubscribed     = errors.New("not subscribed")
)
type Broker struct {
	lock     sync.RWMutex
	messages chan *types.Message
	adapters adapterSet
	opts     brokerOptions
	done     chan struct{}
}
func NewBroker() *Broker {
	opts := defaultBrokerOptions()
	b := &Broker{
		adapters: newAdapterSet(),
		messages: make(chan *types.Message, opts.messagesChSize),
		opts:     opts,
		done:     make(chan struct{}),
	}
	go b.run()
	return b
}
func (b *Broker) run() {
	for {
		select {
		case <-b.done:
			return
		case msg := <-b.messages:
			b.send(msg)
		}
	}
}
func (b *Broker) send(msg *types.Message) {
	b.lock.RLock()
	defer b.lock.RUnlock()
	b.adapters.ForEach(func(a *Adapter) {
		a.noticed <- msg
	})
}
func (b *Broker) registerAdapter(a *Adapter) {
	b.lock.Lock()
	defer b.lock.Unlock()
	if b.adapters.Has(a) {
		log.Panicf("BUG: the same adapter registered more than once: %v", a)
	}
	b.adapters.Add(a)
}
func (b *Broker) unregisterAdapter(a *Adapter) {
	b.lock.Lock()
	defer b.lock.Unlock()
	if !b.adapters.Has(a) {
		log.Panicf("BUG: unregistering Adapter that is not registered: %v", a)
	}
	b.adapters.Delete(a)
}