4 files changed, 187 insertions, 19 deletions
diff --git a/plugins/broadcast/memory/bst/bst.go b/plugins/broadcast/memory/bst/bst.go
new file mode 100644
index 00000000..7d09a10f
--- /dev/null
+++ b/plugins/broadcast/memory/bst/bst.go
@@ -0,0 +1,134 @@
+package bst
+
+// BST ...
+type BST struct {
+	// registered topic, not unique
+	topic string
+	// associated connections with the topic
+	uuids map[string]struct{}
+
+	// left and right subtrees
+	left  *BST
+	right *BST
+}
+
+func NewBST() Storage {
+	return &BST{}
+}
+
+// Insert uuid to the topic
+func (B *BST) Insert(uuid string, topic string) {
+	curr := B
+
+	for {
+		if curr.topic == topic {
+			curr.uuids[uuid] = struct{}{}
+			return
+		}
+		// if topic less than curr topic
+		if curr.topic < topic {
+			if curr.left == nil {
+				curr.left = &BST{
+					topic: topic,
+					uuids: map[string]struct{}{uuid: {}},
+				}
+				return
+			}
+			// move forward
+			curr = curr.left
+		} else {
+			if curr.right == nil {
+				curr.right = &BST{
+					topic: topic,
+					uuids: map[string]struct{}{uuid: {}},
+				}
+				return
+			}
+
+			curr = curr.right
+		}
+	}
+}
+
+func (B *BST) Get(topic string) map[string]struct{} {
+	curr := B
+	for curr != nil {
+		if curr.topic == topic {
+			return curr.uuids
+		}
+		if curr.topic < topic {
+			curr = curr.left
+		}
+		if curr.topic > topic {
+			curr = curr.right
+		}
+	}
+
+	return nil
+}
+
+func (B *BST) Remove(uuid string, topic string) {
+	B.removeHelper(uuid, topic, nil)
+}
+
+func (B *BST) removeHelper(uuid string, topic string, parent *BST) {
+	curr := B
+	for curr != nil {
+		if topic < curr.topic {
+			parent = curr
+			curr = curr.left
+		} else if topic > curr.topic {
+			parent = curr
+			curr = curr.right
+		} else {
+			if len(curr.uuids) > 1 {
+				if _, ok := curr.uuids[uuid]; ok {
+					delete(curr.uuids, uuid)
+					return
+				}
+			}
+
+			if curr.left != nil && curr.right != nil {
+				curr.topic, curr.uuids = curr.right.traverseForMinString()
+				curr.right.removeHelper(curr.topic, uuid, curr)
+			} else if parent == nil {
+				if curr.left != nil {
+					curr.topic = curr.left.topic
+					curr.uuids = curr.left.uuids
+
+					curr.right = curr.left.right
+					curr.left = curr.left.left
+				} else if curr.right != nil {
+					curr.topic = curr.right.topic
+					curr.uuids = curr.right.uuids
+
+					curr.left = curr.right.left
+					curr.right = curr.right.right
+				} else {
+					// single node tree
+				}
+			} else if parent.left == curr {
+				if curr.left != nil {
+					parent.left = curr.left
+				} else {
+					parent.left = curr.right
+				}
+			} else if parent.right == curr {
+				if curr.left != nil {
+					parent.right = curr.left
+				} else {
+					parent.right = curr.right
+				}
+			}
+			break
+		}
+	}
+}
+
+//go:inline
+func (B *BST) traverseForMinString() (string, map[string]struct{}) {
+	if B.left == nil {
+		return B.topic, B.uuids
+	}
+	return B.left.traverseForMinString()
+}
diff --git a/plugins/broadcast/memory/bst/bst_test.go b/plugins/broadcast/memory/bst/bst_test.go
new file mode 100644
index 00000000..b5ad6c10
--- /dev/null
+++ b/plugins/broadcast/memory/bst/bst_test.go
@@ -0,0 +1,33 @@
+package bst
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewBST(t *testing.T) {
+	g := NewBST()
+
+	for i := 0; i < 100; i++ {
+		g.Insert(uuid.NewString(), "comments")
+	}
+
+	for i := 0; i < 100; i++ {
+		g.Insert(uuid.NewString(), "comments2")
+	}
+
+	for i := 0; i < 100; i++ {
+		g.Insert(uuid.NewString(), "comments3")
+	}
+
+	exist := g.Get("comments")
+	assert.Len(t, exist, 100)
+
+	exist2 := g.Get("comments2")
+	assert.Len(t, exist2, 100)
+
+	exist3 := g.Get("comments3")
+	assert.Len(t, exist3, 100)
+}
diff --git a/plugins/broadcast/memory/bst/interface.go b/plugins/broadcast/memory/bst/interface.go
new file mode 100644
index 00000000..ecf40414
--- /dev/null
+++ b/plugins/broadcast/memory/bst/interface.go
@@ -0,0 +1,11 @@
+package bst
+
+// Storage is general in-memory BST storage implementation
+type Storage interface {
+	// Insert inserts to a vertex with topic ident connection uuid
+	Insert(uuid string, topic string)
+	// Remove removes uuid from topic, if the uuid is single for a topic, whole vertex will be removed
+	Remove(uuid, topic string)
+	// Get will return all connections associated with the topic
+	Get(topic string) map[string]struct{}
+}
diff --git a/plugins/broadcast/memory/driver.go b/plugins/broadcast/memory/driver.go
index 2eb45c8e..80527e4b 100644
--- a/plugins/broadcast/memory/driver.go
+++ b/plugins/broadcast/memory/driver.go
@@ -1,39 +1,29 @@
 package memory
 
-import "github.com/spiral/roadrunner/v2/plugins/broadcast"
+import (
+	"github.com/spiral/roadrunner/v2/plugins/broadcast"
+)
 
 type Driver struct {
 }
 
-func NewInMemoryDriver() broadcast.Subscriber {
+func NewInMemoryDriver() broadcast.Storage {
 	b := &Driver{}
 	return b
 }
 
-func (d *Driver) Serve() error {
+func (d *Driver) Store(uuid string, topics ...string) {
 	panic("implement me")
 }
 
-func (d *Driver) Stop() {
+func (d *Driver) StorePattern(uuid string, pattern string) {
 	panic("implement me")
 }
 
-func (d *Driver) Subscribe(upstream chan *broadcast.Message, topics ...string) error {
+func (d *Driver) GetConnection(pattern string) []string {
 	panic("implement me")
 }
 
-func (d *Driver) SubscribePattern(upstream chan *broadcast.Message, pattern string) error {
-	panic("implement me")
-}
-
-func (d *Driver) Unsubscribe(upstream chan *broadcast.Message, topics ...string) error {
-	panic("implement me")
-}
-
-func (d *Driver) UnsubscribePattern(upstream chan *broadcast.Message, pattern string) error {
-	panic("implement me")
-}
-
-func (d *Driver) Publish(messages ...*broadcast.Message) error {
-	panic("implement me")
+func (d *Driver) Construct(key string) (broadcast.Storage, error) {
+	return nil, nil
 }