[WIP] Feature: TCPFactory/TCPTransport instead of dmsg

Makefile

@@ -98,6 +98,10 @@ bin: ## Build `skywire-visor`, `skywire-cli`, `hypervisor`, `SSH-cli`
 	${OPTS} go build ${BUILD_OPTS} -o ./hypervisor ./cmd/hypervisor
 	${OPTS} go build ${BUILD_OPTS} -o ./SSH-cli ./cmd/therealssh-cli
 
+minibuild:
+	${OPTS} go build ${BUILD_OPTS} -o ./apps/skychat.v1.0 ./cmd/apps/skychat	
+	${OPTS} go build ${BUILD_OPTS} -o ./skywire-visor ./cmd/skywire-visor
+
 
 release: ## Build `skywire-visor`, `skywire-cli`, `hypervisor`, `SSH-cli` and apps without -race flag
 	${OPTS} go build -o ./skywire-visor ./cmd/skywire-visor

cmd/apps/skychat/chat.go

@@ -13,35 +13,46 @@ import (
 	"net"
 	"net/http"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/skycoin/dmsg/cipher"
+	"github.com/skycoin/skycoin/src/util/logging"
 
 	"github.com/skycoin/skywire/internal/netutil"
+	th "github.com/skycoin/skywire/internal/testhelpers"
 	"github.com/skycoin/skywire/pkg/app"
 	"github.com/skycoin/skywire/pkg/routing"
+	"github.com/skycoin/skywire/pkg/util/env"
 )
 
-var addr = flag.String("addr", ":8000", "address to bind")
-var r = netutil.NewRetrier(50*time.Millisecond, 5, 2)
-
 var (
+	addr    = flag.String("addr", ":8000", "address to bind")
+	logger  = logging.NewMasterLogger().PackageLogger("chat")
+	retrier = netutil.NewRetrier(
+		env.Duration("SKYWIRE_CHAT_RETRY_BACKOFF", 50*time.Millisecond),
+		env.UInt32("SKYWIRE_CHAT_RETRY_TIMES", 5),
+		env.UInt32("SKYWIRE_CHAT_RETRY_FACTOR", 2))
 	chatApp   *app.App
 	clientCh  chan string
 	chatConns map[cipher.PubKey]net.Conn
 	connsMu   sync.Mutex
+
+	trcLog   = logger.WithField("_module", th.GetCallerN(3))
+	trStart  = func() error { trcLog.Info("ENTER"); return nil }
+	trFinish = func(_ error) { trcLog.Info(th.Trace("EXIT")) }
 )
 
 func main() {
 	flag.Parse()
 
 	a, err := app.Setup(&app.Config{AppName: "skychat", AppVersion: "1.0", ProtocolVersion: "0.0.1"})
 	if err != nil {
-		log.Fatal("Setup failure: ", err)
+		trcLog.Fatal("Setup failure", err)
 	}
 	defer func() {
 		if err := a.Close(); err != nil {
-			log.Println("Failed to close app:", err)
+			trcLog.Info("Failed to close app:", err)
 		}
 	}()
 
@@ -57,8 +68,8 @@ func main() {
 	http.HandleFunc("/message", messageHandler)
 	http.HandleFunc("/sse", sseHandler)
 
-	log.Println("Serving HTTP on", *addr)
-	log.Fatal(http.ListenAndServe(*addr, nil))
+	trcLog.Info("Serving HTTP on", *addr)
+	trcLog.Info(http.ListenAndServe(*addr, nil))
 }
 
 func listenLoop() {
@@ -79,29 +90,39 @@ func listenLoop() {
 }
 
 func handleConn(conn net.Conn) {
+	// defer trFinish(trStart())
+
+	var cntr uint64
 	raddr := conn.RemoteAddr().(routing.Addr)
+
 	for {
+		atomic.AddUint64(&cntr, 1)
+
+		trcLog.Debugf("CYCLE %03d START", cntr)
 		buf := make([]byte, 32*1024)
 		n, err := conn.Read(buf)
 		if err != nil {
-			log.Println("failed to read packet:", err)
+			trcLog.Info("failed to read packet:", err)
 			return
 		}
 
 		clientMsg, err := json.Marshal(map[string]string{"sender": raddr.PubKey.Hex(), "message": string(buf[:n])})
 		if err != nil {
-			log.Printf("Failed to marshal json: %v", err)
+			trcLog.Info("Failed to marshal json: ", err)
 		}
 		select {
 		case clientCh <- string(clientMsg):
-			log.Printf("received and sent to ui: %s\n", clientMsg)
+			trcLog.Debugf("received and sent to ui: %s\n", clientMsg)
 		default:
-			log.Printf("received and trashed: %s\n", clientMsg)
+			trcLog.Debugf("received and trashed: %s\n", clientMsg)
 		}
+		trcLog.Debugf("CYCLE %03d END", cntr)
 	}
 }
 
 func messageHandler(w http.ResponseWriter, req *http.Request) {
+	// defer trFinish(trStart())
+
 	data := map[string]string{}
 	if err := json.NewDecoder(req.Body).Decode(&data); err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
@@ -115,17 +136,25 @@ func messageHandler(w http.ResponseWriter, req *http.Request) {
 	}
 
 	addr := routing.Addr{PubKey: pk, Port: 1}
+	trcLog.Info("addr: ", addr)
+
 	connsMu.Lock()
 	conn, ok := chatConns[pk]
 	connsMu.Unlock()
+	trcLog.Debugf("chatConn: %v  pk:%v\n", chatConns, pk)
+
+	var cntr uint64
 
 	if !ok {
 		var err error
-		err = r.Do(func() error {
+		err = retrier.Do(func() error {
+			atomic.AddUint64(&cntr, 1)
+			trcLog.Debugf("dial %v  addr:%v\n", cntr, addr)
 			conn, err = chatApp.Dial(addr)
 			return err
 		})
 		if err != nil {
+			trcLog.Info("err: ", err)
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
@@ -149,6 +178,8 @@ func messageHandler(w http.ResponseWriter, req *http.Request) {
 }
 
 func sseHandler(w http.ResponseWriter, req *http.Request) {
+	// defer trFinish(trStart())
+
 	f, ok := w.(http.Flusher)
 	if !ok {
 		http.Error(w, "Streaming unsupported!", http.StatusBadRequest)

cmd/apps/therealproxy-client/therealproxy-client.go

@@ -19,7 +19,7 @@ import (
 
 const socksPort = 3
 
-var r = netutil.NewRetrier(time.Second, 0, 1)
+var r = netutil.NewRetrier(time.Millisecond*50, 5, 2)
 
 func main() {
 	var addr = flag.String("addr", ":1080", "Client address to listen on")

cmd/skywire-cli/commands/node/gen-config.go

@@ -100,10 +100,12 @@ func defaultConfig() *visor.Config {
 	conf.Routing.RouteFinder = "https://routefinder.skywire.skycoin.net/"
 
 	const defaultSetupNodePK = "0324579f003e6b4048bae2def4365e634d8e0e3054a20fc7af49daf2a179658557"
+
 	sPK := cipher.PubKey{}
 	if err := sPK.UnmarshalText([]byte(defaultSetupNodePK)); err != nil {
 		log.WithError(err).Warnf("Failed to unmarshal default setup node public key %s", defaultSetupNodePK)
 	}
+
 	conf.Routing.SetupNodes = []cipher.PubKey{sPK}
 	conf.Routing.Table.Type = "boltdb"
 	conf.Routing.Table.Location = "./skywire/routing.db"
@@ -120,5 +122,8 @@ func defaultConfig() *visor.Config {
 
 	conf.Interfaces.RPCAddress = "localhost:3435"
 
+	conf.TransportType = "dmsg"
+	conf.PubKeysFile = "./local/pubkeys"
+
 	return conf
 }

go.mod

@@ -3,7 +3,10 @@ module github.com/skycoin/skywire
 go 1.12
 
 require (
+	github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1
 	github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
+	github.com/blang/semver v3.5.1+incompatible // indirect
+	github.com/boltdb/bolt v1.3.1 // indirect
 	github.com/go-chi/chi v4.0.2+incompatible
 	github.com/google/uuid v1.1.1
 	github.com/gorilla/handlers v1.4.0
@@ -14,6 +17,7 @@ require (
 	github.com/pkg/profile v1.3.0
 	github.com/prometheus/client_golang v1.0.0
 	github.com/prometheus/common v0.4.1
+	github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 // indirect
 	github.com/sirupsen/logrus v1.4.2
 	github.com/skycoin/dmsg v0.0.0-20190805065636-70f4c32a994f
 	github.com/skycoin/skycoin v0.26.0

go.sum

@@ -1,4 +1,6 @@
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1 h1:GDQdwm/gAcJcLAKQQZGOJ4knlw+7rfEQQcmwTbt4p5E=
+github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1/go.mod h1:xTS7Pm1pD1mvyM075QCDSRqH6qRLXylzS24ZTpRiSzQ=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY=
@@ -9,6 +11,10 @@ github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkY
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
+github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
+github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
+github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
+github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -83,6 +89,8 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNGfs=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
+github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 h1:pntxY8Ary0t43dCZ5dqY4YTJCObLY1kIXl0uzMv+7DE=
+github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=

integration/check-delays.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+
+MSGD=messaging.discovery.skywire.skycoin.net
+MSGD_GET="https://"$MSGD"/messaging-discovery/available_servers" 
+
+echo -e "\nTCP delays. Measuring by ping:"
+ping $MSGD -c 10 -q
+
+if type mtr > /dev/null; then
+    echo -e "\nTCP delays. Measuring by mtr:"
+    mtr -y 2 --report  --report-cycles=5   $MSGD > /tmp/msgd-out.txt
+
+    cat /tmp/msgd-out.txt
+else
+    echo -e "\nTCP delays. mtr not found. Install for detailed stats"
+fi
+
+if type vegeta > /dev/null; then
+    echo -e "\nHTTP delays. Measuring by vegeta:"
+    echo "GET "$MSGD_GET  \
+        | vegeta attack -duration=10s |tee results.bin |vegeta report 
+else
+    echo -e "\nHTTP delays.vegeta not found\n. Install with \ngo get -u github.com/tsenart/vegeta\n for detailed stats"        
+fi
+
+echo -e "\nHTTP delays. Measuring by curl:"
+curl $MSGD_GET >/dev/null

integration/tcp-tr/env-vars.sh

@@ -0,0 +1,13 @@
+# intended to be sourced  `source ./integration/tcp-tr/env-vars.sh`
+
+export RPC_A=192.168.1.2:3435
+export RPC_C=192.168.1.3:3435
+
+alias CLI_A='./skywire-cli --rpc $RPC_A'
+alias CLI_C='./skywire-cli --rpc $RPC_C'
+
+export PK_A=$(./skywire-cli --rpc $RPC_A node pk)
+export PK_C=$(./skywire-cli --rpc $RPC_C node pk)
+
+export CHAT_A=http://192.168.1.2:8001/message
+export CHAT_C=http://192.168.1.3:8001/message