From d2e9d8320857f5768c54843a43ad16f59d6a3e8f Mon Sep 17 00:00:00 2001
From: Valery Piashchynski <piashchynski.valery@gmail.com>
Date: Thu, 20 May 2021 22:46:19 +0300
Subject: - Update linters - Implement base interfaces - Implement BST search
 algo for the in-memory storage

Signed-off-by: Valery Piashchynski <piashchynski.valery@gmail.com>
---
 plugins/broadcast/ws/connection/connection.go | 69 +++++++++++++++++++++++++++
 1 file changed, 69 insertions(+)
 create mode 100644 plugins/broadcast/ws/connection/connection.go

(limited to 'plugins/broadcast/ws/connection/connection.go')

diff --git a/plugins/broadcast/ws/connection/connection.go b/plugins/broadcast/ws/connection/connection.go
new file mode 100644
index 00000000..cfb47e35
--- /dev/null
+++ b/plugins/broadcast/ws/connection/connection.go
@@ -0,0 +1,69 @@
+package connection
+
+import (
+	"sync"
+
+	"github.com/gofiber/websocket/v2"
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+// Connection represents wrapped and safe to use from the different threads websocket connection
+type Connection struct {
+	sync.RWMutex
+	log  logger.Logger
+	conn *websocket.Conn
+}
+
+func NewConnection(wsConn *websocket.Conn, log logger.Logger) *Connection {
+	return &Connection{
+		conn: wsConn,
+		log:  log,
+	}
+}
+
+func (c *Connection) Write(mt int, data []byte) error {
+	c.Lock()
+	defer c.Unlock()
+
+	const op = errors.Op("websocket_write")
+	// handle a case when a goroutine tried to write into the closed connection
+	defer func() {
+		if r := recover(); r != nil {
+			c.log.Warn("panic handled, tried to write into the closed connection")
+		}
+	}()
+
+	err := c.conn.WriteMessage(mt, data)
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	return nil
+}
+
+func (c *Connection) Read() (int, []byte, error) {
+	c.RLock()
+	defer c.RUnlock()
+	const op = errors.Op("websocket_read")
+
+	mt, data, err := c.conn.ReadMessage()
+	if err != nil {
+		return -1, nil, errors.E(op, err)
+	}
+
+	return mt, data, nil
+}
+
+func (c *Connection) Close() error {
+	c.Lock()
+	defer c.Unlock()
+	const op = errors.Op("websocket_close")
+
+	err := c.conn.Close()
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	return nil
+}
-- 
cgit v1.2.3