[Server] add linter error fixes

Author: unit-adm

go.mod

@@ -8,7 +8,6 @@ require (
 	github.com/golang/snappy v0.0.3
 	github.com/gorilla/websocket v1.4.2
 	github.com/rs/zerolog v1.21.0
-	github.com/stretchr/testify v1.7.0
 	github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264
 	github.com/unit-io/unitdb-go v0.0.0-20210407101657-d9db9270f78d
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2

server/internal/batch.go

@@ -30,8 +30,6 @@ type (
 		count int
 		size  int
 		msgs  []*lp.PublishMessage
-
-		timeRefs []timeID
 	}
 	batchManager struct {
 		mu           sync.RWMutex
@@ -158,22 +156,18 @@ func (m *batchManager) publishLoop(interval time.Duration) {
 // dispatch handles publishing messages for the batches in queue.
 func (m *batchManager) dispatch(timeout time.Duration) {
 LOOP:
-	for {
-		select {
-		case b, ok := <-m.publishQueue:
-			if !ok {
-				close(m.send)
-				m.stopWg.Done()
-				return
-			}
+	b, ok := <-m.publishQueue
+	if !ok {
+		close(m.send)
+		m.stopWg.Done()
+		return
+	}
 
-			select {
-			case m.send <- b:
-			default:
-				// pool is full, let GC handle the batches
-				goto WAIT
-			}
-		}
+	select {
+	case m.send <- b:
+	default:
+		// pool is full, let GC handle the batches
+		goto WAIT
 	}
 
 WAIT:
@@ -189,26 +183,21 @@ func (m *batchManager) publish(c *_Conn, publishWaitTimeout time.Duration) {
 		case <-m.stop:
 			// run queued messges from the publish queue and
 			// process it until queue is empty.
-			for {
-				select {
-				case b, ok := <-m.send:
-					if !ok {
-						m.stopWg.Done()
-						return
-					}
-					pub := &lp.Publish{Messages: b.msgs}
-					pub.MessageID = c.MessageIds.NextID(lp.PUBLISH.Value())
-
-					// persist outbound
-					store.Log.PersistOutbound(c.adp, uint32(c.connID), pub)
-
-					select {
-					case c.pub <- pub:
-					case <-time.After(publishWaitTimeout):
-						// b.r.setError(errors.New("publish timeout error occurred"))
-					}
-				default:
-				}
+			b, ok := <-m.send
+			if !ok {
+				m.stopWg.Done()
+				return
+			}
+			pub := &lp.Publish{Messages: b.msgs}
+			pub.MessageID = c.MessageIds.NextID(lp.PUBLISH.Value())
+
+			// persist outbound
+			store.Log.PersistOutbound(c.adp, uint32(c.connID), pub)
+
+			select {
+			case c.pub <- pub:
+			case <-time.After(publishWaitTimeout):
+				// b.r.setError(errors.New("publish timeout error occurred"))
 			}
 		case b := <-m.send:
 			if b != nil {

server/internal/cluster.go

@@ -20,6 +20,7 @@ import (
 	"encoding/gob"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net"
 	"net/rpc"
 	"sort"
@@ -225,21 +226,19 @@ func (n *ClusterNode) callAsync(proc string, msg, resp interface{}, done chan *r
 
 	myDone := make(chan *rpc.Call, 1)
 	go func() {
-		select {
-		case call := <-myDone:
-			if call.Error != nil {
-				n.lock.Lock()
-				if n.connected {
-					n.endpoint.Close()
-					n.connected = false
-					go n.reconnect()
-				}
-				n.lock.Unlock()
+		call := <-myDone
+		if call.Error != nil {
+			n.lock.Lock()
+			if n.connected {
+				n.endpoint.Close()
+				n.connected = false
+				go n.reconnect()
 			}
+			n.lock.Unlock()
+		}
 
-			if done != nil {
-				done <- call
-			}
+		if done != nil {
+			done <- call
 		}
 	}()
 
@@ -306,7 +305,7 @@ func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
 				return nil
 			}
 
-			log.Info("cluster.Master", "new connection request"+string(msg.Conn.ConnID))
+			log.Info("cluster.Master", "new connection request"+fmt.Sprint(msg.Conn.ConnID))
 			conn = Globals.Service.newRpcConn(node, msg.Conn.ConnID, msg.Conn.SessID, msg.Conn.ClientID)
 			go conn.rpcWriteLoop()
 		}
@@ -333,7 +332,7 @@ func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
 
 // Dispatch receives messages from the master node addressed to a specific local connection.
 func (Cluster) Proxy(resp *ClusterResp, unused *bool) error {
-	log.Info("cluster.Proxy", "response from Master for connection "+string(resp.FromConnID))
+	log.Info("cluster.Proxy", "response from Master for connection "+fmt.Sprint(resp.FromConnID))
 
 	// This cluster member received a response from topic owner to be forwarded to a connection
 	// Find appropriate connection, send the message to it
@@ -376,7 +375,7 @@ func (c *Cluster) isRemoteContract(contract string) bool {
 // Forward client message to the Master (cluster node which owns the topic)
 func (c *Cluster) routeToContract(msg lp.LineProtocol, topic *security.Topic, msgType uint8, m *message.Message, conn *_Conn) error {
 	// Find the cluster node which owns the topic, then forward to it.
-	n := c.nodeForContract(string(conn.clientID.Contract()))
+	n := c.nodeForContract(fmt.Sprint(conn.clientID.Contract()))
 	if n == nil {
 		return errors.New("cluster.routeToContract: attempt to route to non-existent node")
 	}
@@ -416,7 +415,7 @@ func (c *Cluster) routeToContract(msg lp.LineProtocol, topic *security.Topic, ms
 				//RemoteAddr: conn.(),
 				Proto:    conn.proto,
 				ConnID:   conn.connID,
-				SessID: conn.sessID,
+				SessID:   conn.sessID,
 				ClientID: conn.clientID}})
 }
 
@@ -620,9 +619,7 @@ func (c *Cluster) rehash(nodes []string) []string {
 		}
 		ringKeys = append(ringKeys, c.thisNodeName)
 	} else {
-		for _, name := range nodes {
-			ringKeys = append(ringKeys, name)
-		}
+		ringKeys = append(ringKeys, nodes...)
 	}
 	ring.Add(ringKeys...)
 

server/internal/conn.go

@@ -19,6 +19,7 @@ package internal
 import (
 	"encoding/binary"
 	"encoding/json"
+	"fmt"
 	"net"
 	"runtime/debug"
 	"strconv"
@@ -37,7 +38,6 @@ import (
 
 type _Conn struct {
 	sync.Mutex
-	tracked            uint32 // Whether the connection was already tracked or not.
 	proto              lp.Proto
 	adp                lp.ProtoAdapter
 	socket             net.Conn
@@ -187,17 +187,17 @@ func (c *_Conn) subscribe(msg lp.Subscribe, topic *security.Topic, sub *lp.Subsc
 			return err
 		}
 	}
-	if exists := c.subs.Exist(key); exists && !msg.IsForwarded && Globals.Cluster.isRemoteContract(string(c.clientID.Contract())) {
+	if exists := c.subs.Exist(key); exists && !msg.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
 		// The contract is handled by a remote node. Forward message to it.
 		if err := Globals.Cluster.routeToContract(&msg, topic, message.SUBSCRIBE, &message.Message{}, c); err != nil {
 			log.ErrLogger.Err(err).Str("context", "conn.subscribe").Int64("connid", int64(c.connID)).Msg("unable to subscribe to remote topic")
 			return err
 		}
-		// Add the subscription to Counters
 	} else {
 		messageId, err := store.Subscription.NewID()
 		if err != nil {
 			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
+			return err
 		}
 		if first := c.subs.Increment(topic.Topic[:topic.Size], key, messageId); first {
 			// Subscribe the subscriber
@@ -232,7 +232,7 @@ func (c *_Conn) unsubscribe(msg lp.Unsubscribe, topic *security.Topic) (err erro
 		// Decrement the subscription counter
 		c.service.meter.Subscriptions.Dec(1)
 	}
-	if !msg.IsForwarded && Globals.Cluster.isRemoteContract(string(c.clientID.Contract())) {
+	if !msg.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
 		// The topic is handled by a remote node. Forward message to it.
 		if err := Globals.Cluster.routeToContract(&msg, topic, message.UNSUBSCRIBE, &message.Message{}, c); err != nil {
 			log.ErrLogger.Err(err).Str("context", "conn.unsubscribe").Int64("connid", int64(c.connID)).Msg("unable to unsubscribe to remote topic")
@@ -252,6 +252,7 @@ func (c *_Conn) publish(pkt lp.Publish, topic *security.Topic, m *lp.PublishMess
 	subscriptions, err := store.Subscription.Get(c.clientID.Contract(), topic.Topic)
 	if err != nil {
 		log.ErrLogger.Err(err).Str("context", "conn.publish")
+		return err
 	}
 	pubMsg := &message.Message{
 		MessageID: pkt.MessageID,
@@ -300,12 +301,13 @@ func (c *_Conn) publish(pkt lp.Publish, topic *security.Topic, m *lp.PublishMess
 	c.service.meter.OutMsgs.Inc(int64(msgCount))
 	c.service.meter.OutBytes.Inc(pubMsg.Size() * int64(msgCount))
 
-	if !pkt.IsForwarded && Globals.Cluster.isRemoteContract(string(c.clientID.Contract())) {
+	if !pkt.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
 		if err = Globals.Cluster.routeToContract(&pkt, topic, message.PUBLISH, pubMsg, c); err != nil {
 			log.ErrLogger.Err(err).Str("context", "conn.publish").Int64("connid", int64(c.connID)).Msg("unable to publish to remote topic")
+			return err
 		}
 	}
-	return err
+	return nil
 }
 
 // Load all stored messages and resend them to ensure DeliveryMode > 1,2 even after an application crash.

server/internal/hdl_conn.go

@@ -22,6 +22,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"time"
 
 	"github.com/unit-io/unitdb/server/internal/message"
@@ -40,7 +41,7 @@ const (
 	requestKeygen   = 812942072  // hash("keygen")
 )
 
-func (c *_Conn) readLoop(ctx context.Context) error {
+func (c *_Conn) readLoop(ctx context.Context) (err error) {
 	defer func() {
 		log.Info("conn.Handler", "closing...")
 		c.close()
@@ -63,7 +64,8 @@ func (c *_Conn) readLoop(ctx context.Context) error {
 			}
 
 			// Message handler
-			if err := c.handler(pkt); err != nil {
+			if err = c.handler(pkt); err != nil {
+				fmt.Println("read:: handler error", err)
 				return err
 			}
 		}
@@ -318,7 +320,9 @@ func (c *_Conn) onSubscribe(sub lp.Subscribe, subsc *lp.Subscription) *types.Err
 		}
 	}
 
-	c.subscribe(sub, topic, subsc)
+	if err := c.subscribe(sub, topic, subsc); err != nil {
+		return types.ErrServerError
+	}
 
 	if subsc.Last != "" {
 		msgs, err := store.Message.Get(c.clientID.Contract(), topic.Topic, subsc.Last)
@@ -327,7 +331,7 @@ func (c *_Conn) onSubscribe(sub lp.Subscribe, subsc *lp.Subscription) *types.Err
 			return types.ErrServerError
 		}
 
-		// Range over the messages in the channel and forward them
+		// Range over the messages from the store and forward them to subscribers
 		for _, m := range msgs {
 			msg := m // Copy message
 			c.SendMessage(&msg)
@@ -355,7 +359,9 @@ func (c *_Conn) onUnsubscribe(unsub lp.Unsubscribe, subsc *lp.Subscription) *typ
 		}
 	}
 
-	c.unsubscribe(unsub, topic)
+	if err := c.unsubscribe(unsub, topic); err != nil {
+		return types.ErrServerError
+	}
 
 	return nil
 }