Initial Bus implementation

bus.go

@@ -0,0 +1,50 @@
+package main
+
+import "sync"
+
+type Bus struct {
+	mu    sync.Mutex
+	chans map[string][]chan Object
+}
+
+func NewBus() *Bus {
+	return &Bus{
+		chans: map[string][]chan Object{},
+	}
+}
+
+func (b *Bus) Announce(obj Object) {
+	key := ObjectKey(obj)
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	chans := b.chans[key]
+	newChans := []chan Object{}
+
+	for _, ch := range chans {
+		select {
+		case ch <- obj:
+			newChans = append(newChans, ch)
+		default:
+			close(ch)
+		}
+	}
+
+	if len(chans) != len(newChans) {
+		b.chans[key] = newChans
+	}
+}
+
+func (b *Bus) Subscribe(obj Object) chan Object {
+	key := ObjectKey(obj)
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	ch := make(chan Object, 100)
+
+	b.chans[key] = append(b.chans[key], ch)
+
+	return ch
+}

main.go

@@ -21,4 +21,7 @@ func main() {
 	store.Read(in)
 
 	fmt.Printf("%+v\n", in)
+
+	bus := NewBus()
+	bus.Announce(in)
 }

object.go

@@ -1,5 +1,6 @@
 package main
 
+import "encoding/hex"
 import "fmt"
 
 type Object interface {
@@ -7,12 +8,10 @@ type Object interface {
 	GetId() string
 }
 
-func ObjectKey(obj Object) string {
-  return fmt.Sprintf(
-    "%d:%s:%d:%s",
-    len(obj.GetType()),
-    obj.GetType(),
-    len(obj.GetId()),
-    obj.GetId(),
-  )
+func ObjectSafeId(obj Object) string {
+	return hex.EncodeToString([]byte(obj.GetId()))
+}
+
+func ObjectKey(obj Object) string {
+	return fmt.Sprintf("%s:%s", obj.GetType(), ObjectSafeId(obj))
 }

store.go

@@ -1,6 +1,5 @@
 package main
 
-import "encoding/hex"
 import "encoding/json"
 import "fmt"
 import "os"
@@ -18,7 +17,7 @@ func NewStore(root string) *Store {
 
 func (s *Store) Write(obj Object) error {
 	dir := filepath.Join(s.root, obj.GetType())
-  filename := hex.EncodeToString([]byte(obj.GetId()))
+	filename := ObjectSafeId(obj)
 
 	err := os.MkdirAll(dir, 0700)
 	if err != nil {
@@ -54,7 +53,7 @@ func (s *Store) Write(obj Object) error {
 
 func (s *Store) Read(obj Object) error {
 	dir := filepath.Join(s.root, obj.GetType())
-  filename := hex.EncodeToString([]byte(obj.GetId()))
+	filename := ObjectSafeId(obj)
 
 	fh, err := os.Open(filepath.Join(dir, filename))
 	if err != nil {