improve gatway and p2papp reconnect

core/common.go

@@ -20,7 +20,6 @@ import (
 const MinNodeNameLen = 8
 
 func getmac(ip string) string {
-	//get mac relative to the ip address which connected to the mq.
 	ifaces, err := net.Interfaces()
 	if err != nil {
 		return ""

core/daemon.go

@@ -6,7 +6,7 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/kardianos/service"
+	"github.com/openp2p-cn/service"
 )
 
 type daemon struct {
@@ -44,9 +44,9 @@ func (d *daemon) run() {
 	}
 	gLog.Println(LvINFO, mydir)
 	conf := &service.Config{
-		Name:        ProducnName,
-		DisplayName: ProducnName,
-		Description: ProducnName,
+		Name:        ProductName,
+		DisplayName: ProductName,
+		Description: ProductName,
 		Executable:  binPath,
 	}
 
@@ -95,9 +95,9 @@ func (d *daemon) run() {
 
 func (d *daemon) Control(ctrlComm string, exeAbsPath string, args []string) error {
 	svcConfig := &service.Config{
-		Name:        ProducnName,
-		DisplayName: ProducnName,
-		Description: ProducnName,
+		Name:        ProductName,
+		DisplayName: ProductName,
+		Description: ProductName,
 		Executable:  exeAbsPath,
 		Arguments:   args,
 	}

core/handlepush.go

@@ -101,7 +101,6 @@ func handlePush(pn *P2PNetwork, subType uint16, msg []byte) error {
 				msg := TunnelMsg{ID: t.id}
 				pn.push(r.From, MsgPushAddRelayTunnelRsp, msg)
 			}
-
 		}(req)
 	case MsgPushAPPKey:
 		req := APPKeySync{}

core/install.go

@@ -17,7 +17,7 @@ import (
 // ./openp2p install -node hhd1207-222 -token YOUR-TOKEN -sharebandwidth 0 -peernode hhdhome-n1 -dstip 127.0.0.1 -dstport 50022 -protocol tcp -srcport 22
 func install() {
 	gLog.Println(LvINFO, "openp2p start. version: ", OpenP2PVersion)
-	gLog.Println(LvINFO, "Contact: QQ: 16947733, Email: openp2p.cn@gmail.com")
+	gLog.Println(LvINFO, "Contact: QQ group 16947733, Email openp2p.cn@gmail.com")
 	gLog.Println(LvINFO, "install start")
 	defer gLog.Println(LvINFO, "install end")
 	// auto uninstall
@@ -74,6 +74,7 @@ func install() {
 	} else {
 		gLog.Println(LvINFO, "start openp2p service ok.")
 	}
+	gLog.Println(LvINFO, "Visit WebUI on https://console.openp2p.cn")
 }
 
 func installByFilename() {

core/openp2p.go

@@ -13,7 +13,7 @@ func Run() {
 	rand.Seed(time.Now().UnixNano())
 	baseDir := filepath.Dir(os.Args[0])
 	os.Chdir(baseDir) // for system service
-	gLog = NewLogger(baseDir, ProducnName, LvDEBUG, 1024*1024, LogFileAndConsole)
+	gLog = NewLogger(baseDir, ProductName, LvDEBUG, 1024*1024, LogFileAndConsole)
 	// TODO: install sub command, deamon process
 	if len(os.Args) > 1 {
 		switch os.Args[1] {
@@ -21,7 +21,7 @@ func Run() {
 			fmt.Println(OpenP2PVersion)
 			return
 		case "update":
-			gLog = NewLogger(baseDir, ProducnName, LvDEBUG, 1024*1024, LogFileAndConsole)
+			gLog = NewLogger(baseDir, ProductName, LvDEBUG, 1024*1024, LogFileAndConsole)
 			targetPath := filepath.Join(defaultInstallPath, defaultBinName)
 			d := daemon{}
 			err := d.Control("restart", targetPath, nil)
@@ -70,7 +70,7 @@ var network *P2PNetwork
 func RunAsModule(baseDir string, token string, bw int, logLevel int) *P2PNetwork {
 	rand.Seed(time.Now().UnixNano())
 	os.Chdir(baseDir) // for system service
-	gLog = NewLogger(baseDir, ProducnName, LvDEBUG, 1024*1024, LogFileAndConsole)
+	gLog = NewLogger(baseDir, ProductName, LvDEBUG, 1024*1024, LogFileAndConsole)
 
 	parseParams("")
 

core/p2papp.go

@@ -204,12 +204,15 @@ func (app *p2pApp) listen() error {
 	if app.rtid != 0 {
 		go app.relayHeartbeatLoop()
 	}
-	for app.tunnel.isRuning() && app.running {
+	for app.tunnel.isRuning() {
 		if app.config.Protocol == "udp" {
 			app.listenUDP()
 		} else {
 			app.listenTCP()
 		}
+		if !app.running {
+			break
+		}
 		time.Sleep(time.Second * 10)
 	}
 	return nil

core/p2pnetwork.go

@@ -7,7 +7,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math"
 	"math/rand"
 	"net/http"
 	"net/url"
@@ -24,15 +23,15 @@ var (
 )
 
 type P2PNetwork struct {
-	conn      *websocket.Conn
-	online    bool
-	running   bool
-	restartCh chan bool
-	wg        sync.WaitGroup
-	writeMtx  sync.Mutex
-	serverTs  int64
-	localTs   int64
-	hbTime    time.Time
+	conn        *websocket.Conn
+	online      bool
+	running     bool
+	restartCh   chan bool
+	wgReconnect sync.WaitGroup
+	writeMtx    sync.Mutex
+	serverTs    int64
+	localTs     int64
+	hbTime      time.Time
 	// msgMap    sync.Map
 	msgMap     map[uint64]chan []byte //key: nodeID
 	msgMapMtx  sync.Mutex
@@ -64,18 +63,16 @@ func P2PNetworkInstance(config *NetworkConfig) *P2PNetwork {
 }
 
 func (pn *P2PNetwork) run() {
-	go pn.readLoop()
 	go pn.autorunApp()
 	heartbeatTimer := time.NewTicker(NetworkHeartbeatTime)
 	for pn.running {
 		select {
-		case <-heartbeatTimer.C: // TODO: deal with connect failed, no send hb
+		case <-heartbeatTimer.C:
 			pn.write(MsgHeartbeat, 0, "")
 
 		case <-pn.restartCh:
 			pn.online = false
-			pn.wg.Wait() // wait read/write goroutine exited
-			time.Sleep(NetworkHeartbeatTime)
+			pn.wgReconnect.Wait() // wait read/write goroutine end
 			err := pn.init()
 			if err != nil {
 				gLog.Println(LvERROR, "P2PNetwork init error:", err)
@@ -130,14 +127,12 @@ func (pn *P2PNetwork) runAll() {
 		}
 		config.retryNum++
 		config.retryTime = time.Now()
-		increase := math.Pow(1.5, float64(config.retryNum)) // exponential increase retry time. 1.5^x
-		if increase > 900 {
-			increase = 900
+		if config.retryNum > 20 {
 			config.Enabled = 0
 			gLog.Printf(LvWARN, "app %s has stopped retry, manually enable it on Web console", config.AppName)
 			continue
 		}
-		config.nextRetryTime = time.Now().Add(time.Second * time.Duration(increase))
+		config.nextRetryTime = time.Now().Add(time.Second * 10)
 		config.connectTime = time.Now()
 		config.peerToken = pn.config.Token
 		gConf.mtx.Unlock() // AddApp will take a period of time
@@ -156,7 +151,6 @@ func (pn *P2PNetwork) autorunApp() {
 			continue
 		}
 		pn.runAll()
-		time.Sleep(time.Second * 10)
 	}
 	gLog.Println(LvINFO, "autorunApp end")
 }
@@ -372,7 +366,7 @@ func (pn *P2PNetwork) addDirectTunnel(config AppConfig, tid uint64) (*P2PTunnel,
 		gLog.Println(LvERROR, "init error:", initErr)
 		return nil, initErr
 	}
-	err := ErrorHandshake
+	var err error
 	// try TCP6
 	if IsIPv6(t.config.peerIPv6) && IsIPv6(t.pn.config.publicIPv6) {
 		gLog.Println(LvINFO, "try TCP6")
@@ -418,7 +412,7 @@ func (pn *P2PNetwork) addDirectTunnel(config AppConfig, tid uint64) (*P2PTunnel,
 			return t, nil
 		}
 	}
-	return nil, err
+	return nil, ErrorHandshake // only ErrorHandshake will try relay
 }
 
 func (pn *P2PNetwork) newTunnel(t *P2PTunnel, tid uint64, isClient bool) error {
@@ -441,6 +435,11 @@ func (pn *P2PNetwork) newTunnel(t *P2PTunnel, tid uint64, isClient bool) error {
 }
 func (pn *P2PNetwork) init() error {
 	gLog.Println(LvINFO, "init start")
+	go func() { //reconnect at least 5s
+		pn.wgReconnect.Add(1)
+		defer pn.wgReconnect.Done()
+		time.Sleep(NatTestTimeout)
+	}()
 	var err error
 	for {
 		// detect nat type
@@ -488,30 +487,31 @@ func (pn *P2PNetwork) init() error {
 			err = errors.New("get local ip failed")
 			break
 		}
-
+		go pn.readLoop()
 		pn.config.mac = getmac(pn.config.localIP)
 		pn.config.os = getOsName()
-
-		req := ReportBasic{
-			Mac:             pn.config.mac,
-			LanIP:           pn.config.localIP,
-			OS:              pn.config.os,
-			HasIPv4:         pn.config.hasIPv4,
-			HasUPNPorNATPMP: pn.config.hasUPNPorNATPMP,
-			Version:         OpenP2PVersion,
-		}
-		rsp := netInfo()
-		gLog.Println(LvDEBUG, "netinfo:", rsp)
-		if rsp != nil && rsp.Country != "" {
-			if IsIPv6(rsp.IP.String()) {
-				pn.config.publicIPv6 = rsp.IP.String()
+		go func() {
+			req := ReportBasic{
+				Mac:             pn.config.mac,
+				LanIP:           pn.config.localIP,
+				OS:              pn.config.os,
+				HasIPv4:         pn.config.hasIPv4,
+				HasUPNPorNATPMP: pn.config.hasUPNPorNATPMP,
+				Version:         OpenP2PVersion,
 			}
-			req.NetInfo = *rsp
-		} else {
-			pn.refreshIPv6(true)
-		}
-		req.IPv6 = pn.config.publicIPv6
-		pn.write(MsgReport, MsgReportBasic, &req)
+			rsp := netInfo()
+			gLog.Println(LvDEBUG, "netinfo:", rsp)
+			if rsp != nil && rsp.Country != "" {
+				if IsIPv6(rsp.IP.String()) {
+					pn.config.publicIPv6 = rsp.IP.String()
+				}
+				req.NetInfo = *rsp
+			} else {
+				pn.refreshIPv6(true)
+			}
+			req.IPv6 = pn.config.publicIPv6
+			pn.write(MsgReport, MsgReportBasic, &req)
+		}()
 		gLog.Println(LvDEBUG, "P2PNetwork init ok")
 		break
 	}
@@ -572,8 +572,8 @@ func (pn *P2PNetwork) handleMessage(t int, msg []byte) {
 
 func (pn *P2PNetwork) readLoop() {
 	gLog.Printf(LvDEBUG, "P2PNetwork readLoop start")
-	pn.wg.Add(1)
-	defer pn.wg.Done()
+	pn.wgReconnect.Add(1)
+	defer pn.wgReconnect.Done()
 	for pn.running {
 		pn.conn.SetReadDeadline(time.Now().Add(NetworkHeartbeatTime + 10*time.Second))
 		t, msg, err := pn.conn.ReadMessage()

core/p2ptunnel.go

@@ -106,7 +106,7 @@ func (t *P2PTunnel) connect() error {
 		AppKey:           appKey,
 		Version:          OpenP2PVersion,
 		LinkMode:         t.config.linkMode,
-		IsUnderlayServer: t.config.isUnderlayServer ^ 1,
+		IsUnderlayServer: t.config.isUnderlayServer ^ 1, // peer
 	}
 	if req.Token == 0 { // no relay token
 		req.Token = t.pn.config.Token
@@ -154,6 +154,9 @@ func (t *P2PTunnel) setRun(running bool) {
 }
 
 func (t *P2PTunnel) isActive() bool {
+	if !t.isRuning() {
+		return false
+	}
 	t.hbMtx.Lock()
 	defer t.hbMtx.Unlock()
 	return time.Now().Before(t.hbTime.Add(TunnelIdleTimeout))

core/protocol.go

@@ -10,8 +10,8 @@ import (
 	"time"
 )
 
-const OpenP2PVersion = "3.5.0"
-const ProducnName string = "openp2p"
+const OpenP2PVersion = "3.5.6"
+const ProductName string = "openp2p"
 const LeastSupportVersion = "3.0.0"
 
 const (
@@ -146,7 +146,7 @@ const (
 	MaxRetry                     = 10
 	RetryInterval                = time.Second * 30
 	PublicIPEchoTimeout          = time.Second * 1
-	NatTestTimeout               = time.Second * 10
+	NatTestTimeout               = time.Second * 5
 	ClientAPITimeout             = time.Second * 10
 	MaxDirectTry                 = 3
 )

core/util_windows.go

@@ -45,9 +45,9 @@ func setFirewall() {
 	}
 	if isXP {
 		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh firewall del allowedprogram "%s"`, fullPath)).Run()
-		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh firewall add allowedprogram "%s" "%s" ENABLE`, ProducnName, fullPath)).Run()
+		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh firewall add allowedprogram "%s" "%s" ENABLE`, ProductName, fullPath)).Run()
 	} else { // win7 or later
-		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh advfirewall firewall del rule name="%s"`, ProducnName)).Run()
-		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh advfirewall firewall add rule name="%s" dir=in action=allow program="%s" enable=yes`, ProducnName, fullPath)).Run()
+		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh advfirewall firewall del rule name="%s"`, ProductName)).Run()
+		exec.Command("cmd.exe", `/c`, fmt.Sprintf(`netsh advfirewall firewall add rule name="%s" dir=in action=allow program="%s" enable=yes`, ProductName, fullPath)).Run()
 	}
 }

go.mod

@@ -4,9 +4,9 @@ go 1.18
 
 require (
 	github.com/gorilla/websocket v1.4.2
-	github.com/kardianos/service v1.2.0
 	github.com/lucas-clemente/quic-go v0.27.0
 	github.com/openp2p-cn/go-reuseport v0.3.2
+	github.com/openp2p-cn/service v1.0.0
 	golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f
 )
 
@@ -14,13 +14,13 @@ require (
 	github.com/cheekybits/genny v1.0.0 // indirect
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
+	github.com/kardianos/service v1.2.2 // indirect
 	github.com/marten-seemann/qtls-go1-16 v0.1.5 // indirect
 	github.com/marten-seemann/qtls-go1-17 v0.1.1 // indirect
 	github.com/marten-seemann/qtls-go1-18 v0.1.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/onsi/ginkgo v1.16.4 // indirect
 	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect
-	golang.org/x/mobile v0.0.0-20221020085226-b36e6246172e // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/net v0.0.0-20220722155237-a158d28d115b // indirect
 	golang.org/x/tools v0.1.12 // indirect