-
-
package bus
-
-import (
- "fmt"
- "reflect"
- "sync"
-
- "github.com/gopatchy/metadata"
-)
-
-type Bus struct {
- mu sync.Mutex
- keyViews map[string]map[uintptr]chan<- any
- typeViews map[string]map[uintptr]chan<- any
-}
-
-func NewBus() *Bus {
- return &Bus{
- keyViews: map[string]map[uintptr]chan<- any{},
- typeViews: map[string]map[uintptr]chan<- any{},
- }
-}
-
-func (b *Bus) Announce(t string, obj any) {
- key := getObjKey(t, obj)
-
- b.mu.Lock()
- defer b.mu.Unlock()
-
- announce(obj, b.keyViews[key])
- announce(obj, b.typeViews[t])
-}
-
-func (b *Bus) Delete(t string, id string) {
- key := getKey(t, id)
-
- b.mu.Lock()
- defer b.mu.Unlock()
-
- for _, c := range b.keyViews[key] {
- close(c)
- }
-
- delete(b.keyViews, key)
-
- announce(id, b.typeViews[t])
-}
-
-func (b *Bus) SubscribeKey(t, id string, initial any) <-chan any {
- key := getKey(t, id)
-
- b.mu.Lock()
- defer b.mu.Unlock()
-
- ret := make(chan any, 100)
-
- ret <- initial
-
- if _, has := b.keyViews[key]; !has {
- b.keyViews[key] = map[uintptr]chan<- any{}
- }
-
- b.keyViews[key][chanID(ret)] = ret
-
- return ret
-}
-
-func (b *Bus) SubscribeType(t string, initial any) <-chan any {
- b.mu.Lock()
- defer b.mu.Unlock()
-
- ret := make(chan any, 100)
-
- ret <- initial
-
- if _, has := b.typeViews[t]; !has {
- b.typeViews[t] = map[uintptr]chan<- any{}
- }
-
- b.typeViews[t][chanID(ret)] = ret
-
- return ret
-}
-
-func (b *Bus) UnsubscribeKey(t, id string, c <-chan any) {
- key := getKey(t, id)
-
- b.mu.Lock()
- defer b.mu.Unlock()
-
- if cw, has := b.keyViews[key][chanID(c)]; has {
- close(cw)
- delete(b.keyViews[key], chanID(c))
- }
-
- if len(b.keyViews[key]) == 0 {
- delete(b.keyViews, key)
- }
-}
-
-func (b *Bus) UnsubscribeType(t string, c <-chan any) {
- b.mu.Lock()
- defer b.mu.Unlock()
-
- if cw, has := b.typeViews[t][chanID(c)]; has {
- close(cw)
- delete(b.typeViews[t], chanID(c))
- }
-
- if len(b.typeViews[t]) == 0 {
- delete(b.typeViews, t)
- }
-}
-
-func getObjKey(t string, obj any) string {
- return getKey(t, metadata.GetMetadata(obj).ID)
-}
-
-func getKey(t string, id string) string {
- return fmt.Sprintf("%s:%s", t, id)
-}
-
-func announce(obj any, chans map[uintptr]chan<- any) {
- for id, c := range chans {
- select {
- case c <- obj:
- default:
- close(c)
- delete(chans, id)
- }
- }
-}
-
-func chanID(c <-chan any) uintptr {
- return reflect.ValueOf(c).Pointer()
-}
-
-
-