Fix: websocket operation channel

cmd/api/ws/channels/operations.go

@@ -21,6 +21,7 @@ type OperationsChannel struct {
 	messages chan Message
 	stop     chan struct{}
 	wg       sync.WaitGroup
+	hashes   map[string]struct{}
 }
 
 // NewOperationsChannel -
@@ -32,6 +33,7 @@ func NewOperationsChannel(address, network string, opts ...ChannelOption) *Opera
 
 		messages: make(chan Message, 10),
 		stop:     make(chan struct{}),
+		hashes:   make(map[string]struct{}),
 	}
 }
 
@@ -105,6 +107,9 @@ func (c *OperationsChannel) createMessage(data datasources.Data) error {
 	if op.Destination != c.Address && op.Source != c.Address {
 		return nil
 	}
+	if _, ok := c.hashes[op.Hash]; ok {
+		return nil
+	}
 	operations, err := c.es.GetOperations(
 		map[string]interface{}{
 			"hash": op.Hash,
@@ -120,6 +125,8 @@ func (c *OperationsChannel) createMessage(data datasources.Data) error {
 	if err != nil {
 		return err
 	}
+
+	c.hashes[op.Hash] = struct{}{}
 	c.messages <- Message{
 		ChannelName: c.GetName(),
 		Body:        response,

cmd/api/ws/client.go

@@ -9,7 +9,6 @@ import (
 	"github.com/baking-bad/bcdhub/cmd/api/ws/channels"
 	"github.com/baking-bad/bcdhub/internal/logger"
 	"github.com/gorilla/websocket"
-	"github.com/valyala/fastjson"
 )
 
 // ClientHandler -
@@ -115,7 +114,6 @@ func (c *Client) sendOk(text string) error {
 }
 
 func (c *Client) receive() {
-	var p fastjson.Parser
 	for {
 		select {
 		case <-c.stop:
@@ -135,15 +133,15 @@ func (c *Client) receive() {
 				continue
 			}
 
-			val, err := p.ParseBytes(data)
-			if err != nil {
+			var msg ActionMessage
+			if err := json.Unmarshal(data, &msg); err != nil {
 				logger.Error(err)
 				continue
 			}
-			action := string(val.GetStringBytes("action"))
-			handler, ok := c.handlers[action]
+
+			handler, ok := c.handlers[msg.Action]
 			if !ok {
-				c.sendError(fmt.Errorf("Unknown handler action: %s", action))
+				c.sendError(fmt.Errorf("Unknown handler action: %s", msg.Action))
 				continue
 			}
 			if err := handler(c, data); err != nil {

cmd/api/ws/hub.go

@@ -201,8 +201,8 @@ func unsubscribeHandler(c *Client, data []byte) error {
 }
 
 func pingHandler(c *Client, data []byte) error {
-	return c.sendMessage(map[string]interface{}{
-		"action": "pong",
+	return c.sendMessage(ActionMessage{
+		Action: "pong",
 	})
 }
 

cmd/api/ws/messages.go

@@ -1,5 +1,9 @@
 package ws
 
+import jsoniter "github.com/json-iterator/go"
+
+var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
 // Statuses
 const (
 	ErrorStatus = "error"
@@ -11,3 +15,8 @@ type StatusMessage struct {
 	Status string `json:"status"`
 	Text   string `json:"text"`
 }
+
+// ActionMessage -
+type ActionMessage struct {
+	Action string `json:"action"`
+}