Appevent module implementation.

.travis.yml

@@ -18,7 +18,7 @@ before_install:
 
 install:
   - go get -u github.com/FiloSottile/vendorcheck
-  - curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $GOPATH/bin v1.22.2
+  - curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $GOPATH/bin v1.27.0
 
 before_script:
   - ci_scripts/create-ip-aliases.sh

cmd/apps/helloworld/helloworld.go

@@ -4,84 +4,167 @@ simple client server app for skywire visor testing
 package main
 
 import (
-	"os"
+	"flag"
+	"fmt"
+	"net"
+	"time"
 
 	"github.com/SkycoinProject/dmsg/cipher"
 	"github.com/sirupsen/logrus"
 
 	"github.com/SkycoinProject/skywire-mainnet/pkg/app"
+	"github.com/SkycoinProject/skywire-mainnet/pkg/app/appevent"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/app/appnet"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/routing"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/util/buildinfo"
 )
 
 const (
-	netType = appnet.TypeSkynet
+	modeServer = "server"
+	modeClient = "client"
+)
+
+var (
+	mode    = flag.String("mode", modeServer, fmt.Sprintf("mode of operation: %v", []string{modeServer, modeClient}))
+	network = flag.String("net", string(appnet.TypeSkynet), fmt.Sprintf("network: %v", []appnet.Type{appnet.TypeSkynet, appnet.TypeDmsg}))
+	remote  = flag.String("remote", "", "remote public key to dial to (client mode only)")
+	port    = flag.Uint("port", 1024, "port to either dial to (client mode), or listen from (server mode)")
 )
 
 var log = logrus.New()
 
 func main() {
-	appC := app.NewClient()
+	flag.Parse()
+
+	subs := prepareSubscriptions()
+	appC := app.NewClient(subs)
 	defer appC.Close()
 
 	if _, err := buildinfo.Get().WriteTo(log.Writer()); err != nil {
-		log.Printf("Failed to output build info: %v", err)
+		log.WithError(err).Info("Failed to output build info.")
 	}
 
-	if len(os.Args) == 1 {
-		port := routing.Port(1024)
-		l, err := appC.Listen(netType, port)
-		if err != nil {
-			log.Fatalf("Error listening network %v on port %d: %v\n", netType, port, err)
-		}
-
-		log.Println("listening for incoming connections")
-		for {
-			conn, err := l.Accept()
-			if err != nil {
-				log.Fatalf("Failed to accept conn: %v\n", err)
-			}
-
-			log.Printf("got new connection from: %v\n", conn.RemoteAddr())
-			go func() {
-				buf := make([]byte, 4)
-				if _, err := conn.Read(buf); err != nil {
-					log.Printf("Failed to read remote data: %v\n", err)
-					// TODO: close conn
-				}
-
-				log.Printf("Message from %s: %s\n", conn.RemoteAddr().String(), string(buf))
-				if _, err := conn.Write([]byte("pong")); err != nil {
-					log.Printf("Failed to write to a remote visor: %v\n", err)
-					// TODO: close conn
-				}
-			}()
-		}
+	switch *mode {
+	case modeServer:
+		runServer(appC)
+	case modeClient:
+		runClient(appC)
+	default:
+		log.WithField("mode", *mode).Fatal("Invalid mode.")
 	}
+}
 
-	remotePK := cipher.PubKey{}
-	if err := remotePK.UnmarshalText([]byte(os.Args[1])); err != nil {
-		log.Fatal("Failed to construct PubKey: ", err, os.Args[1])
-	}
+func prepareSubscriptions() *appevent.Subscriber {
+	subs := appevent.NewSubscriber()
+
+	subs.TCPDial(func(data appevent.TCPDialData) {
+		log.WithField("event_type", data.Type()).
+			WithField("event_data", data).
+			Info("Received event.")
+	})
 
-	conn, err := appC.Dial(appnet.Addr{
-		Net:    netType,
-		PubKey: remotePK,
-		Port:   10,
+	subs.TCPClose(func(data appevent.TCPCloseData) {
+		log.WithField("event_type", data.Type()).
+			WithField("event_data", data).
+			Info("Received event.")
 	})
+
+	return subs
+}
+
+func runServer(appC *app.Client) {
+	log := log.
+		WithField("network", *network).
+		WithField("port", *port)
+
+	lis, err := appC.Listen(appnet.Type(*network), routing.Port(*port))
 	if err != nil {
-		log.Fatalf("Failed to open remote conn: %v\n", err)
+		log.WithError(err).Fatal("Failed to listen.")
+	}
+	log.Info("Listening for incoming connections.")
+
+	for {
+		conn, err := lis.Accept()
+		if err != nil {
+			log.WithError(err).Fatal("Failed to accept connection.")
+		}
+		go handleServerConn(log, conn)
 	}
+}
+
+func handleServerConn(log logrus.FieldLogger, conn net.Conn) {
+	log = log.WithField("remote_addr", conn.RemoteAddr())
+	log.Info("Serving connection.")
+	defer func() {
+		log.WithError(conn.Close()).Debug("Closed connection.")
+	}()
 
-	if _, err := conn.Write([]byte("ping")); err != nil {
-		log.Fatalf("Failed to write to a remote visor: %v\n", err)
+	for {
+		buf := make([]byte, 1024)
+		n, err := conn.Read(buf)
+		if err != nil {
+			log.WithField("n", n).WithError(err).
+				Error("Failed to read from connection.")
+			return
+		}
+		msg := string(buf[:n])
+		log.WithField("n", n).WithField("data", msg).Info("Read from connection.")
+
+		n, err = conn.Write([]byte(fmt.Sprintf("I've got your message: %s", msg)))
+		if err != nil {
+			log.WithField("n", n).WithError(err).
+				Error("Failed to write to connection.")
+			return
+		}
+		log.WithField("n", n).Info("Wrote response message.")
 	}
+}
 
-	buf := make([]byte, 4)
-	if _, err = conn.Read(buf); err != nil {
-		log.Fatalf("Failed to read remote data: %v\n", err)
+func runClient(appC *app.Client) {
+
+	var remotePK cipher.PubKey
+	if err := remotePK.UnmarshalText([]byte(*remote)); err != nil {
+		log.WithError(err).Fatal("Invalid remote public key.")
 	}
 
-	log.Printf("Message from %s: %s", conn.RemoteAddr().String(), string(buf))
+	var conn net.Conn
+
+	for i := 0; true; i++ {
+
+		time.Sleep(time.Second * 2)
+
+		if conn != nil {
+			log.WithError(conn.Close()).Debug("Connection closed.")
+			conn = nil
+		}
+
+		var err error
+		conn, err = appC.Dial(appnet.Addr{
+			Net:    appnet.Type(*network),
+			PubKey: remotePK,
+			Port:   routing.Port(*port),
+		})
+		if err != nil {
+			log.WithError(err).Error("Failed to dial.")
+			time.Sleep(time.Second)
+			continue
+		}
+
+		n, err := conn.Write([]byte(fmt.Sprintf("Hello world! %d", i)))
+		if err != nil {
+			log.WithField("n", n).WithError(err).
+				Error("Failed to write to connection.")
+			continue
+		}
+
+		buf := make([]byte, 1024)
+		n, err = conn.Read(buf)
+		if err != nil {
+			log.WithField("n", n).WithError(err).
+				Error("Failed to read from connection.")
+			continue
+		}
+		msg := string(buf[:n])
+		log.WithField("n", n).WithField("data", msg).Info("Read reply from connection.")
+	}
 }

cmd/apps/skychat/chat.go

@@ -42,7 +42,7 @@ var (
 )
 
 func main() {
-	appC = app.NewClient()
+	appC = app.NewClient(nil)
 	defer appC.Close()
 
 	if _, err := buildinfo.Get().WriteTo(os.Stdout); err != nil {

cmd/apps/skysocks-client/skysocks-client.go

@@ -49,7 +49,7 @@ func dialServer(appCl *app.Client, pk cipher.PubKey) (net.Conn, error) {
 }
 
 func main() {
-	appC := app.NewClient()
+	appC := app.NewClient(nil)
 	defer appC.Close()
 
 	skysocks.Log = log

cmd/apps/skysocks/skysocks.go

@@ -25,7 +25,7 @@ const (
 var log = logrus.New()
 
 func main() {
-	appC := app.NewClient()
+	appC := app.NewClient(nil)
 	defer appC.Close()
 
 	skysocks.Log = log

cmd/apps/vpn-client/vpn-client.go

@@ -81,7 +81,7 @@ func main() {
 		noiseCreds = vpn.NewNoiseCredentials(localSK, localPK)
 	}
 
-	appClient := app.NewClient()
+	appClient := app.NewClient(nil)
 	defer appClient.Close()
 
 	log.Infof("Connecting to VPN server %s", serverPK.String())

cmd/apps/vpn-server/vpn-server.go

@@ -59,7 +59,7 @@ func main() {
 		noiseCreds = vpn.NewNoiseCredentials(localSK, localPK)
 	}
 
-	appClient := app.NewClient()
+	appClient := app.NewClient(nil)
 	defer appClient.Close()
 
 	osSigs := make(chan os.Signal, 2)

go.mod

@@ -3,7 +3,7 @@ module github.com/SkycoinProject/skywire-mainnet
 go 1.13
 
 require (
-	github.com/SkycoinProject/dmsg v0.1.1-0.20200420091742-8c1a3d828a49
+	github.com/SkycoinProject/dmsg v0.1.1-0.20200523194607-be73f083a729
 	github.com/SkycoinProject/skycoin v0.27.0
 	github.com/SkycoinProject/yamux v0.0.0-20191213015001-a36efeefbf6a
 	github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5

go.sum

@@ -2,8 +2,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/SkycoinProject/dmsg v0.0.0-20200306152741-acee74fa4514/go.mod h1:DzykXMLlx6Fx0fGjZsCIRas/MIvxW8DZpmDA6f2nCRk=
-github.com/SkycoinProject/dmsg v0.1.1-0.20200420091742-8c1a3d828a49 h1:rYqmvSRA+rq6LTne/Ge34T0i4yjSHSwkhk0ER6relWU=
-github.com/SkycoinProject/dmsg v0.1.1-0.20200420091742-8c1a3d828a49/go.mod h1:MiX+UG/6fl3g+9rS13/fq7BwUQ2eOlg1yOBOnNf6J6A=
+github.com/SkycoinProject/dmsg v0.1.1-0.20200523194607-be73f083a729 h1:Edgnt4ido4MGfNTEJUYqNeXt0AlJ4EHlFCWBrKYPvT4=
+github.com/SkycoinProject/dmsg v0.1.1-0.20200523194607-be73f083a729/go.mod h1:MiX+UG/6fl3g+9rS13/fq7BwUQ2eOlg1yOBOnNf6J6A=
 github.com/SkycoinProject/skycoin v0.26.0/go.mod h1:xqPLOKh5B6GBZlGA7B5IJfQmCy7mwimD9NlqxR3gMXo=
 github.com/SkycoinProject/skycoin v0.27.0 h1:N3IHxj8ossHOcsxLYOYugT+OaELLncYHJHxbbYLPPmY=
 github.com/SkycoinProject/skycoin v0.27.0/go.mod h1:xqPLOKh5B6GBZlGA7B5IJfQmCy7mwimD9NlqxR3gMXo=
@@ -59,8 +59,11 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-redis/redis v6.15.6+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee h1:s+21KNqlpePfkah2I+gwHF8xmJWRjooY+5248k6m4A0=
 github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee/go.mod h1:L0fX3K22YWvt/FAX9NnzrNzcI4wNYi9Yku4O0LKYflo=
+github.com/gobwas/pool v0.2.0 h1:QEmUOlnSjWtnpRGHF3SauEiOsy82Cup83Vf2LcMlnc8=
 github.com/gobwas/pool v0.2.0/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
+github.com/gobwas/ws v1.0.2 h1:CoAavW/wd/kulfZmSIBt6p24n4j7tHgNVCjsfHVNUbo=
 github.com/gobwas/ws v1.0.2/go.mod h1:szmBTxLgaFppYjEmNtny/v3w89xOydFnnZMcgRRu/EM=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
@@ -80,6 +83,7 @@ github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
@@ -90,6 +94,7 @@ github.com/gorilla/handlers v1.4.2/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/
 github.com/gorilla/securecookie v1.1.1 h1:miw7JPhV+b/lAHSXz4qd/nN9jRiAFV5FwjeKyCS8BvQ=
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
+github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
 github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -118,8 +123,10 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJ
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
@@ -279,6 +286,7 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20191024005414-555d28b269f0 h1:/5xXl8Y5W96D+TtHSlonuFqGHIWVuyCkGJLwGh9JJFs=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -287,6 +295,7 @@ golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384 h1:TFlARGu6Czu1z7q93HTxcP1P+/ZFC/IKythI5RzrnRg=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.zx2c4.com/wireguard v0.0.20200320 h1:1vE6zVeO7fix9cJX1Z9ZQ+ikPIIx7vIyU0o0tLDD88g=
 golang.zx2c4.com/wireguard v0.0.20200320/go.mod h1:lDian4Sw4poJ04SgHh35nzMVwGSYlPumkdnHcucAQoY=
@@ -298,6 +307,7 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQ
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=

internal/vpn/os.go

@@ -16,6 +16,7 @@ func parseCIDR(ipCIDR string) (ipStr, netmask string, err error) {
 	return ip.String(), fmt.Sprintf("%d.%d.%d.%d", net.Mask[0], net.Mask[1], net.Mask[2], net.Mask[3]), nil
 }
 
+//nolint:unparam
 func run(bin string, args ...string) error {
 	cmd := exec.Command(bin, args...) //nolint:gosec
 

internal/vpn/os_server_linux.go

@@ -75,6 +75,7 @@ func EnableIPv6Forwarding() error {
 // EnableIPMasquerading enables IP masquerading for the interface with name `ifcName`.
 func EnableIPMasquerading(ifcName string) error {
 	cmd := fmt.Sprintf(enableIPMasqueradingCMDFmt, ifcName)
+	//nolint:gosec
 	if err := exec.Command("sh", "-c", cmd).Run(); err != nil {
 		return fmt.Errorf("error running command %s: %w", cmd, err)
 	}
@@ -85,6 +86,7 @@ func EnableIPMasquerading(ifcName string) error {
 // DisableIPMasquerading disables IP masquerading for the interface with name `ifcName`.
 func DisableIPMasquerading(ifcName string) error {
 	cmd := fmt.Sprintf(disableIPMasqueradingCMDFmt, ifcName)
+	//nolint:gosec
 	if err := exec.Command("sh", "-c", cmd).Run(); err != nil {
 		return fmt.Errorf("error running command %s: %w", cmd, err)
 	}