clean up code for websocket connection

Author: Ivan Sukach

jsonrpc/http_json_handler.go

@@ -9,7 +9,7 @@ import (
 	"reflect"
 	"sort"
 
-	cmtjson "github.com/cometbft/cometbft/libs/json"
+	tmjson "github.com/cometbft/cometbft/libs/json"
 	"github.com/cometbft/cometbft/libs/log"
 	"github.com/cometbft/cometbft/rpc/jsonrpc/server"
 	"github.com/cometbft/cometbft/rpc/jsonrpc/types"
@@ -160,7 +160,7 @@ func mapParamsToArgs(
 
 		if p, ok := params[argName]; ok && p != nil && len(p) > 0 {
 			val := reflect.New(argType)
-			err := cmtjson.Unmarshal(p, val.Interface())
+			err := tmjson.Unmarshal(p, val.Interface())
 			if err != nil {
 				return nil, err
 			}
@@ -187,7 +187,7 @@ func arrayParamsToArgs(
 	for i, p := range params {
 		argType := rpcFunc.args[i+argsOffset]
 		val := reflect.New(argType)
-		err := cmtjson.Unmarshal(p, val.Interface())
+		err := tmjson.Unmarshal(p, val.Interface())
 		if err != nil {
 			return nil, err
 		}

vm/rpc.go

@@ -5,8 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"github.com/cometbft/cometbft/config"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
-	cmtquery "github.com/cometbft/cometbft/libs/pubsub/query"
+	"github.com/cometbft/cometbft/libs/pubsub"
 	"github.com/cometbft/cometbft/rpc/jsonrpc/server"
 	"sort"
 	"time"
@@ -82,7 +81,7 @@ func (rpc *RPC) Routes() map[string]*jsonrpc.RPCFunc {
 	}
 }
 
-func (rpc *RPC) CMTRoutes() map[string]*server.RPCFunc {
+func (rpc *RPC) TMRoutes() map[string]*server.RPCFunc {
 	return map[string]*server.RPCFunc{
 		//subscribe/unsubscribe are reserved for websocket events.
 		"subscribe":       server.NewWSRPCFunc(rpc.Subscribe, "query"),
@@ -868,7 +867,7 @@ func (rpc *RPC) Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSu
 
 	rpc.vm.logger.Info("Subscribe to query", "remote", addr, "query", query)
 
-	q, err := cmtquery.New(query)
+	q, err := tmquery.New(query)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse query: %w", err)
 	}
@@ -912,7 +911,7 @@ func (rpc *RPC) Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSu
 					}
 				}
 			case <-sub.Canceled():
-				if sub.Err() != cmtpubsub.ErrUnsubscribed {
+				if sub.Err() != pubsub.ErrUnsubscribed {
 					var reason string
 					if sub.Err() == nil {
 						reason = "CometBFT exited"
@@ -941,7 +940,7 @@ func (rpc *RPC) Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSu
 func (rpc *RPC) Unsubscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultUnsubscribe, error) {
 	addr := ctx.RemoteAddr()
 	rpc.vm.logger.Info("Unsubscribe from query", "remote", addr, "query", query)
-	q, err := cmtquery.New(query)
+	q, err := tmquery.New(query)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse query: %w", err)
 	}

vm/types/state/executor.go

@@ -11,7 +11,7 @@ import (
 	"github.com/cometbft/cometbft/libs/fail"
 	"github.com/cometbft/cometbft/libs/log"
 	"github.com/cometbft/cometbft/mempool"
-	cmtproto "github.com/cometbft/cometbft/proto/tendermint/types"
+	tmproto "github.com/cometbft/cometbft/proto/tendermint/types"
 	"github.com/cometbft/cometbft/proxy"
 	statetypes "github.com/cometbft/cometbft/state"
 	"github.com/cometbft/cometbft/types"
@@ -463,7 +463,7 @@ func BuildLastCommitInfo(block *types.Block, lastValSet *types.ValidatorSet, ini
 		commitSig := block.LastCommit.Signatures[i]
 		votes[i] = abci.VoteInfo{
 			Validator:   types.TM2PB.Validator(val),
-			BlockIdFlag: cmtproto.BlockIDFlag(commitSig.BlockIDFlag),
+			BlockIdFlag: tmproto.BlockIDFlag(commitSig.BlockIDFlag),
 		}
 	}
 
@@ -542,7 +542,7 @@ func BuildExtendedCommitInfo(ec *types.ExtendedCommit, valSet *types.ValidatorSe
 
 		votes[i] = abci.ExtendedVoteInfo{
 			Validator:          types.TM2PB.Validator(val),
-			BlockIdFlag:        cmtproto.BlockIDFlag(ecs.BlockIDFlag),
+			BlockIdFlag:        tmproto.BlockIDFlag(ecs.BlockIDFlag),
 			VoteExtension:      ecs.Extension,
 			ExtensionSignature: ecs.ExtensionSignature,
 		}

vm/types/state/utils.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/cometbft/cometbft/crypto"
 	"github.com/cometbft/cometbft/libs/json"
-	cmtproto "github.com/cometbft/cometbft/proto/tendermint/types"
+	tmproto "github.com/cometbft/cometbft/proto/tendermint/types"
 	"github.com/cometbft/cometbft/state"
 	"github.com/cometbft/cometbft/types"
 	cmttime "github.com/cometbft/cometbft/types/time"
@@ -58,7 +58,7 @@ func EncodeBlock(block *types.Block) ([]byte, error) {
 }
 
 func DecodeBlock(data []byte) (*types.Block, error) {
-	protoBlock := new(cmtproto.Block)
+	protoBlock := new(tmproto.Block)
 	if err := protoBlock.Unmarshal(data); err != nil {
 		return nil, err
 	}

vm/vm.go

@@ -19,7 +19,7 @@ import (
 	"github.com/cometbft/cometbft/crypto"
 	"github.com/cometbft/cometbft/crypto/secp256k1"
 	"github.com/cometbft/cometbft/libs/log"
-	cmtpubsub "github.com/cometbft/cometbft/libs/pubsub"
+	"github.com/cometbft/cometbft/libs/pubsub"
 	"github.com/cometbft/cometbft/mempool"
 	"github.com/cometbft/cometbft/node"
 	"github.com/cometbft/cometbft/proxy"
@@ -522,11 +522,11 @@ func (vm *LandslideVM) CreateHandlers(context.Context, *emptypb.Empty) (*vmpb.Cr
 	vm.serverCloser.Add(server)
 
 	mux := http2.NewServeMux()
-	cmtRPC := NewRPC(vm)
-	wm := rpcserver.NewWebsocketManager(cmtRPC.CMTRoutes(),
+	tmRPC := NewRPC(vm)
+	wm := rpcserver.NewWebsocketManager(tmRPC.TMRoutes(),
 		rpcserver.OnDisconnect(func(remoteAddr string) {
 			err := vm.eventBus.UnsubscribeAll(context.Background(), remoteAddr)
-			if err != nil && err != cmtpubsub.ErrSubscriptionNotFound {
+			if err != nil && err != pubsub.ErrSubscriptionNotFound {
 				vm.logger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err)
 			}
 		}),
@@ -536,7 +536,7 @@ func (vm *LandslideVM) CreateHandlers(context.Context, *emptypb.Empty) (*vmpb.Cr
 	wm.SetLogger(vm.logger)
 	mux.HandleFunc("/websocket", wm.WebsocketHandler)
 	mux.HandleFunc("/v1/websocket", wm.WebsocketHandler)
-	jsonrpc.RegisterRPCFuncs(mux, cmtRPC.Routes(), vm.logger)
+	jsonrpc.RegisterRPCFuncs(mux, tmRPC.Routes(), vm.logger)
 
 	httppb.RegisterHTTPServer(server, http.NewServer(mux))