fix memory leak and optimize cpu usage

metrics/connection.go

@@ -30,13 +30,6 @@ var (
 		Name:      "send_blockings",
 		Help:      "Number of send blocking connections",
 	})
-
-	countMessageFromClosed = prometheus.NewCounter(prometheus.CounterOpts{
-		Namespace: promNamespace,
-		Subsystem: promSubsystem,
-		Name:      "sending_on_closed",
-		Help:      "Number of sending on closed connections",
-	})
 )
 
 func IncNewConnection() {
@@ -51,10 +44,6 @@ func IncSendBlocking() {
 	countSendBlocking.Inc()
 }
 
-func IncMessageFromClosed() {
-	countMessageFromClosed.Inc()
-}
-
 func SetCurrentConnections(num int) {
 	gaugeCurrentConnections.Set(float64(num))
 }

relay/schema.go

@@ -1,6 +1,8 @@
 package relay
 
 import (
+	"crypto/rand"
+	"encoding/base64"
 	"encoding/json"
 	"strings"
 	"sync"
@@ -159,6 +161,16 @@ func NewTopicSet() *TopicSet {
 	}
 }
 
+func (tm TopicSet) Get() map[string]struct{} {
+	tm.Lock()
+	defer tm.Unlock()
+	newmap := make(map[string]struct{})
+	for key, value := range tm.Data {
+		newmap[key] = value
+	}
+	return newmap
+}
+
 func (tm TopicSet) Set(topic string) {
 	tm.Lock()
 	defer tm.Unlock()
@@ -178,3 +190,11 @@ type ClientUnregisterEvent struct {
 	client *client
 	reason error
 }
+
+func generateRandomBytes16() string {
+	buf := make([]byte, 16)
+	if _, err := rand.Read(buf); err != nil {
+		return ""
+	}
+	return base64.StdEncoding.EncodeToString(buf)
+}

relay/schema_test.go

@@ -85,3 +85,27 @@ func TestGetTopicsByClientAndClear(t *testing.T) {
 		t.Errorf("length error, expected: %v, actual: %v", expectedLen, actualLen)
 	}
 }
+
+func TestTopicGet(t *testing.T) {
+	ts := NewTopicSet()
+
+	ts.Set("hello")
+	ts.Set("hello1")
+	ts.Set("hello2")
+
+	topics := ts.Get()
+	actualLen := len(topics)
+	expectedLen := 3
+	if actualLen != expectedLen {
+		t.Errorf("length error, expected: %v, actual: %v", expectedLen, actualLen)
+	}
+	if _, ok := topics["hello"]; !ok {
+		t.Errorf("key does not exists")
+	}
+	if _, ok := topics["hello1"]; !ok {
+		t.Errorf("key does not exists")
+	}
+	if _, ok := topics["hello2"]; !ok {
+		t.Errorf("key does not exists")
+	}
+}

relay/wsconn.go

@@ -3,7 +3,6 @@ package relay
 import (
 	"bytes"
 	"encoding/json"
-	"fmt"
 	"strings"
 
 	"github.com/RabbyHub/derelay/log"
@@ -19,7 +18,6 @@ type client struct {
 
 	id        string   // randomly generate, just for logging
 	role      RoleType // dapp or wallet
-	session   string   // session id
 	pubTopics *TopicSet
 	subTopics *TopicSet
 
@@ -31,7 +29,6 @@ func (c *client) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
 	if c != nil {
 		encoder.AddString("id", c.id)
 		encoder.AddString("role", string(c.role))
-		encoder.AddString("session", string(c.session))
 		encoder.AddArray("pubTopics", c.pubTopics)
 		encoder.AddArray("subTopics", c.subTopics)
 	}
@@ -87,7 +84,6 @@ func (c *client) send(message SocketMessage) {
 	case c.sendbuf <- message:
 	default:
 		metrics.IncSendBlocking()
-		log.Error("client sendbuf full", fmt.Errorf(""), zap.Any("client", c), zap.Any("len(sendbuf)", len(c.sendbuf)), zap.Any("message", message))
 	}
 }