tracker websorkcet to grpc refactor

config/cfg.go

@@ -23,6 +23,7 @@ var (
 	HtlcMaxFee  = 0.01
 
 	TrackerHost = "127.0.0.1:60060"
+	TrackerHostGrpc = "127.0.0.1:60061"
 
 	ChainNodeType = "regtest"
 	//P2P
@@ -91,4 +92,5 @@ func Init() {
 		return
 	}
 	TrackerHost = tracker.Key("host").MustString("localhost:60060")
+	TrackerHostGrpc = tracker.Key("grpc").MustString("localhost:60061")
 }

config/conf.ini

@@ -32,4 +32,6 @@ port = 4001
 ;https://omnilaboratory.github.io/obd/#/nodes-in-testnet
 ;
 host = 62.234.216.108:60060
-;host = 127.0.0.1:60060
+grpc = 62.234.216.108:60061
+;host = 127.0.0.1:60060
+;grpc = 127.0.0.1:60061

go.mod

@@ -15,8 +15,9 @@ require (
 	github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect
 	github.com/gin-gonic/gin v1.6.3
 	github.com/go-ini/ini v1.56.0
-	github.com/golang/protobuf v1.4.0
+	github.com/golang/protobuf v1.5.2
 	github.com/gorilla/websocket v1.4.2
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.3
 	github.com/happierall/l v0.0.0-20190729144513-5eb32176fb02
 	github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 // indirect
 	github.com/jonboulle/clockwork v0.1.0 // indirect
@@ -33,6 +34,7 @@ require (
 	github.com/satori/go.uuid v1.2.0
 	github.com/shopspring/decimal v1.2.0
 	github.com/smartystreets/goconvey v1.6.4 // indirect
+	github.com/stretchr/testify v1.7.0
 	github.com/tebeka/strftime v0.1.4 // indirect
 	github.com/tidwall/gjson v1.6.0
 	github.com/tyler-smith/go-bip32 v0.0.0-20170922074101-2c9cfd177564
@@ -42,10 +44,11 @@ require (
 	github.com/vmihailenco/msgpack v4.0.4+incompatible // indirect
 	go.etcd.io/bbolt v1.3.4 // indirect
 	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9
-	golang.org/x/sys v0.0.0-20201024232916-9f70ab9862d5 // indirect
 	golang.org/x/tools v0.0.0-20201023174141-c8cfbd0f21e6 // indirect
-	google.golang.org/grpc v1.31.1
+	google.golang.org/grpc v1.43.0
+	google.golang.org/protobuf v1.27.1
 	gopkg.in/ini.v1 v1.56.0 // indirect
-	gopkg.in/yaml.v2 v2.3.0 // indirect
+	gorm.io/driver/sqlite v1.2.6
+	gorm.io/gorm v1.22.4
 	launchpad.net/gocheck v0.0.0-20140225173054-000000000087 // indirect
 )

lightclient/connect_tracker.go

@@ -1,42 +1,24 @@
 package lightclient
 
 import (
-	"encoding/json"
+	"context"
 	"github.com/asdine/storm"
 	"github.com/asdine/storm/q"
-	"github.com/gorilla/websocket"
 	"github.com/omnilaboratory/obd/bean"
-	"github.com/omnilaboratory/obd/bean/enum"
 	"github.com/omnilaboratory/obd/config"
 	"github.com/omnilaboratory/obd/conn"
 	"github.com/omnilaboratory/obd/dao"
 	"github.com/omnilaboratory/obd/service"
 	"github.com/omnilaboratory/obd/tool"
-	trackerBean "github.com/omnilaboratory/obd/tracker/bean"
-	"github.com/tidwall/gjson"
+	"github.com/omnilaboratory/obd/tracker/tkrpc"
 	"io/ioutil"
 	"log"
-	"net/url"
-	"reflect"
 	"strings"
 	"time"
 )
 
-var conn *websocket.Conn
-var ticker3m *time.Ticker
-var isReset = true
-
 func ConnectToTracker() (err error) {
 
-	u := url.URL{Scheme: "ws", Host: config.TrackerHost, Path: "/ws"}
-	log.Printf("begin to connect to tracker: %s", u.String())
-
-	conn, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
-	if err != nil {
-		log.Println("fail to dial tracker:", err)
-		return err
-	}
-
 	chainNodeType, trackerP2pAddress, err := conn2tracker.GetChainNodeType()
 	if err != nil {
 		return err
@@ -48,157 +30,55 @@ func ConnectToTracker() (err error) {
 	if err != nil {
 		return err
 	}
+	 SynData()
+	//HeartBeat will set obd-node online info.
+	go HeartBeat()
 
-	if service.TrackerChan == nil {
-		service.TrackerChan = make(chan []byte)
-	}
-
-	if isReset {
-		go readDataFromWs()
-	}
-
-	if ticker3m == nil {
-		startSchedule()
-	}
 	return nil
 }
 
-func readDataFromWs() {
-	isReset = false
-	ticker := time.NewTicker(time.Minute * 2)
-	defer ticker.Stop()
-
-	defer func(ticker *time.Ticker) {
-		if r := recover(); r != nil {
-			log.Println("tracker goroutine recover")
-			ticker.Stop()
-			conn = nil
-			isReset = true
-		}
-	}(ticker)
-	defer conn.Close()
-
-	// read message
-	go func() {
-		for {
-			if conn == nil {
-				isReset = true
-				return
-			}
-
-			_, message, err := conn.ReadMessage()
-			if err != nil {
-				isReset = true
-				log.Println("socket to tracker get err:", err)
-				conn = nil
-				return
-			}
-
-			//log.Println("get data from tracker", string(message))
-			replyMessage := bean.ReplyMessage{}
-			err = json.Unmarshal(message, &replyMessage)
-			if err == nil {
-				switch replyMessage.Type {
-				case enum.MsgType_Tracker_GetHtlcPath_351:
-					v := reflect.ValueOf(replyMessage.Result)
-					requestMessage := bean.RequestMessage{}
-					requestMessage.Type = replyMessage.Type
-					requestMessage.Data = ""
-					if v.Kind() == reflect.Map {
-						dataMap := replyMessage.Result.(map[string]interface{})
-						requestMessage.RecipientUserPeerId = dataMap["senderPeerId"].(string)
-						requestMessage.Data = dataMap["h"].(string) + "_" + dataMap["path"].(string) + "_" + tool.FloatToString(dataMap["amount"].(float64), 8)
-						//requestMessage.Data = dataMap["h"].(string) + "_" + dataMap["path"].(string)
-					}
-					htlcTrackerDealModule(requestMessage)
-				case enum.MsgType_Tracker_Connect_301:
-					config.ChainNodeType = replyMessage.Result.(string)
-					go SynData()
-				}
-			}
-		}
-	}()
-
-	// heartbeat and check whether
-	for {
-		select {
-		case t := <-ticker.C:
-			if conn != nil {
-				info := make(map[string]interface{})
-				info["type"] = enum.MsgType_Tracker_HeartBeat_302
-				info["data"] = t.String()
-				bytes, err := json.Marshal(info)
-				err = conn.WriteMessage(websocket.TextMessage, bytes)
-				if err != nil {
-					log.Println("HeartBeat:", err)
-					isReset = true
-					log.Println("socket to tracker get err:", err)
-					conn = nil
-					return
-				}
-			} else {
-				return
-			}
-		}
-	}
-}
-
+var ITclient=service.ITclient
 func SynData() {
 	log.Println("synData to tracker")
-	updateP2pAddressLogin()
-	sycUserInfos()
-	sycChannelInfos()
-}
-
-func updateP2pAddressLogin() {
-	info := make(map[string]interface{})
-	info["type"] = enum.MsgType_Tracker_NodeLogin_303
-	nodeLoginInfo := &bean.ObdNodeLoginRequest{}
-	nodeLoginInfo.NodeId = tool.GetObdNodeId()
-	nodeLoginInfo.P2PAddress = localServerDest
-	info["data"] = nodeLoginInfo
-	bytes, err := json.Marshal(info)
-	if err != nil {
-		log.Println(err)
-	} else {
-		sendMsgToTracker(bytes)
-	}
-}
 
-func sycUserInfos() {
-
-	nodes := make([]bean.ObdNodeUserLoginRequest, 0)
+	//sycUserInfos()
+	ureq:=&tkrpc.UpdateUserInfosReq{ }
 	for userId, _ := range GlobalWsClientManager.OnlineClientMap {
-		user := bean.ObdNodeUserLoginRequest{}
-		user.UserId = userId
-		nodes = append(nodes, user)
+		req:=&tkrpc.UpdateUserInfoReq{UserId: userId}
+		ureq.UpdateUserInfoReqs=append(ureq.UpdateUserInfoReqs,req)
 	}
-	if len(nodes) > 0 {
-		log.Println("syn UserInfo data to tracker", nodes)
-		info := make(map[string]interface{})
-		info["type"] = enum.MsgType_Tracker_UpdateUserInfo_353
-		info["data"] = nodes
-		bytes, err := json.Marshal(&info)
-		if err == nil {
-			sendMsgToTracker(bytes)
-		}
+	_,err:= ITclient.UpdateUserInfos(todo,ureq)
+	if err != nil {
+		panic(err)
+	}
+	log.Println("synced userinfos")
+
+	//sycChannelInfos()
+	creq:=&tkrpc.UpdateChannelInfosReq{}
+	channels := getChannelInfos()
+	for _, channel := range channels {
+		cInfo:= &tkrpc.ChannelInfo{}
+		cInfo.ChannelId  =channel.ChannelId
+		cInfo.PropertyId =channel.PropertyId
+		cInfo.CurrState  =tkrpc.ChannelState(int(channel.CurrState))
+		cInfo.PeerIda    =channel.PeerIdA
+		cInfo.PeerIdb    =channel.PeerIdB
+		cInfo.AmountA    =channel.AmountA
+		cInfo.AmountB    =channel.AmountB
+		cInfo.IsAlice    =channel.IsAlice
+		cInfo.NodeId     =tool.GetObdNodeId()
+		creq.ChannelInfos=append(creq.ChannelInfos,cInfo)
+	}
+	_,err= ITclient.UpdateChannelInfos(todo,creq)
+	if err != nil {
+		panic(err)
 	}
-}
+	log.Println("synced ChannelInfos")
 
-//同步通道信息
-func sycChannelInfos() {
-	nodes := getChannelInfos()
-	if len(nodes) > 0 {
-		info := make(map[string]interface{})
-		info["type"] = enum.MsgType_Tracker_UpdateChannelInfo_350
-		info["data"] = nodes
-		bytes, err := json.Marshal(info)
-		if err == nil {
-			sendMsgToTracker(bytes)
-		}
-	}
 }
 
+var todo=context.TODO()
+
 func getChannelInfos() []bean.ChannelInfoRequest {
 	_dir := config.DataDirectory + "/" + config.ChainNodeType
 	files, _ := ioutil.ReadDir(_dir)
@@ -334,55 +214,22 @@ func checkChannel(userId string, db storm.Node, nodes []bean.ChannelInfoRequest)
 	return nodes
 }
 
-func sendMsgToTracker(msg []byte) {
-	//log.Println("send to tracker", string(msg))
-	if conn == nil {
-		isReset = true
-		err := ConnectToTracker()
-		if err != nil {
-			log.Println(err)
-			return
-		}
-	}
-	err := conn.WriteMessage(websocket.TextMessage, msg)
-	if err != nil {
-		log.Println("write:", err)
-		return
-	}
-}
-
-func startSchedule() {
-	go func() {
-		for {
-			select {
-			case msg := <-service.TrackerChan:
-				sendMsgToTracker(msg)
-				msgType := gjson.Parse(string(msg)).Get("type").Int()
-				if msgType == 350 {
-					message := &trackerBean.RequestMessage{}
-					err := json.Unmarshal(msg, message)
-					if err == nil {
-						marshal, err := json.Marshal(message.Data)
-						if err == nil {
-							sendChannelInfoToIndirectTracker(string(marshal))
-						}
-					}
+func HeartBeat() {
+	ninfo := &tkrpc.UpdateNodeInfoReq{NodeId: tool.GetObdNodeId(), P2PAddress: localServerDest, IsOnline: 1}
+	for {
+		cliStream, err := ITclient.HeartBeat(context.TODO(), nil)
+		if err == nil {
+			for {
+				err = cliStream.Send(ninfo)
+				if err != nil {
+					break
 				}
+				time.Sleep(1 * time.Minute)
 			}
 		}
-	}()
-
-	go func() {
-		ticker3m = time.NewTicker(1 * time.Minute)
-		defer ticker3m.Stop()
-		for {
-			select {
-			case t := <-ticker3m.C:
-				if conn == nil {
-					log.Println("reconnect tracker ", t)
-					_ = ConnectToTracker()
-				}
-			}
+		if err != nil {
+			log.Println("heartBeat error", err)
 		}
-	}()
+		time.Sleep(2 * time.Second)
+	}
 }