- Update all main abstractions

- Desighn a new interfaces responsible for the whole PubSub
- New plugin - websockets

Signed-off-by: Valery Piashchynski <[email protected]>

1 files changed, 50 insertions, 0 deletions

diff --git a/plugins/websockets/storage/storage.go b/plugins/websockets/storage/storage.go
new file mode 100644
index 00000000..a7e49207
--- /dev/null
+++ b/plugins/websockets/storage/storage.go
@@ -0,0 +1,50 @@
+package storage
+
+import (
+	"sync"
+
+	"github.com/spiral/roadrunner/v2/plugins/memory/bst"
+)
+
+type Storage struct {
+	sync.RWMutex
+	BST bst.Storage
+}
+
+func NewStorage() *Storage {
+	return &Storage{
+		BST: bst.NewBST(),
+	}
+}
+
+func (s *Storage) Store(connID string, topics []string) {
+	s.Lock()
+	defer s.Unlock()
+
+	for i := 0; i < len(topics); i++ {
+		s.BST.Insert(connID, topics[i])
+	}
+}
+
+func (s *Storage) Remove(connID string, topics []string) {
+	s.Lock()
+	defer s.Unlock()
+
+	for i := 0; i < len(topics); i++ {
+		s.BST.Remove(connID, topics[i])
+	}
+}
+
+func (s *Storage) Get(topics []string, res map[string]struct{}) {
+	s.RLock()
+	defer s.RUnlock()
+
+	for i := 0; i < len(topics); i++ {
+		d := s.BST.Get(topics[i])
+		if len(d) > 0 {
+			for ii := range d {
+				res[ii] = struct{}{}
+			}
+		}
+	}
+}