ipv6 p2p

Author: matteriot

chans/chans.go

@@ -0,0 +1,6 @@
+package chans
+
+import "github.com/OpenIoTHub/gateway-go/v2/models"
+
+// ClientTaskChan 传入访问者的ipv6监听ip+端口，任务从本chan接受配置创建客户端
+var ClientTaskChan = make(chan models.Ipv6ClientHandleTask)

client/lib.go

@@ -6,6 +6,7 @@ import (
 	"github.com/OpenIoTHub/gateway-go/v2/config"
 	"github.com/OpenIoTHub/gateway-go/v2/info"
 	"github.com/OpenIoTHub/gateway-go/v2/services"
+	"github.com/OpenIoTHub/gateway-go/v2/tasks"
 	"github.com/OpenIoTHub/openiothub_grpc_api/pb-go/proto/gateway"
 	"github.com/OpenIoTHub/utils/v2/models"
 	"github.com/gin-gonic/gin"
@@ -34,6 +35,7 @@ func Run() {
 }
 
 func start() {
+	tasks.RunTasks()
 	//启动http服务
 	go func() {
 		if HttpPort == 0 {

config/global.go

@@ -0,0 +1,6 @@
+package config
+
+// Ipv6ListenTcpHandlePort ipv6监听访问端直接的连接，
+// 处理测试链接，测试链接显示可以连通后面有请求直接使用，当连接不通则使用其他连接
+// 这个端口在启动时开启，开启后将实际端口号存到这里，当接到访问者请求之后将该端口和ipv6地址发送到访问者
+var Ipv6ListenTcpHandlePort int = 0

models/chansData.go

@@ -0,0 +1,7 @@
+package models
+
+type Ipv6ClientHandleTask struct {
+	RunId        string `json:"RunId"`
+	Ipv6AddrIp   string `json:"Ipv6AddrIp"`
+	Ipv6AddrPort int    `json:"Ipv6AddrPort"`
+}

netservice/services/connect/service/getIPv6Addr.go

@@ -2,24 +2,38 @@ package service
 
 import (
 	"encoding/json"
+	"github.com/OpenIoTHub/gateway-go/v2/chans"
+	"github.com/OpenIoTHub/gateway-go/v2/config"
+	models2 "github.com/OpenIoTHub/gateway-go/v2/models"
 	"github.com/OpenIoTHub/getip/v2/iputils"
 	"github.com/OpenIoTHub/utils/v2/models"
 	"github.com/OpenIoTHub/utils/v2/msg"
 	"log"
 	"net"
 )
 
-// GetIPv6Addr 传过来一个ipv6Addr+port返回一个ipv6Addr+port
+// GetIPv6Addr 传过来一个ipv6Addr+port返回一个ipv6Addr+port，传过来的ipv6Addr+port用于这边通过channel通知连接，
+// 访问端获取的ipv6Addr+port用于先校验连通性再有连接需求的时候使用连接
 func GetIPv6Addr(stream net.Conn, service *models.NewService) error {
-	ipv6Map := make(map[string]interface{})
-
-	// 获取磁盘信息
+	//从service读取访问者监听ipv6端口号，这里将ipv6地址+port通过chan发送到任务创建连接并处理请求
+	var remoteIpv6ServerConfig models2.Ipv6ClientHandleTask
+	err := json.Unmarshal([]byte(service.Config), &remoteIpv6ServerConfig)
+	if err != nil {
+		log.Println("json.Unmarshal([]byte(service.Config), &config):" + err.Error())
+		return err
+	}
+	chans.ClientTaskChan <- remoteIpv6ServerConfig
+	// 获取ipv6公网地址，可能为空字符串代表没有或者没获取到
 	var ipv6Addr = iputils.GetMyPublicIpv6()
-	ipv6Map["Ipv6Addr"] = ipv6Addr
-	rstByte, err := json.Marshal(ipv6Map)
+	ipv6Info := models2.Ipv6ClientHandleTask{}
+	//访问者只要保存提供服务的ipv6地址+端口，有连接请求时创建连接
+	ipv6Info.Ipv6AddrIp = ipv6Addr
+	ipv6Info.Ipv6AddrPort = config.Ipv6ListenTcpHandlePort
+	rstByte, err := json.Marshal(ipv6Info)
 	if err != nil {
 		log.Println("json.Marshal(ipv6Map)：")
 		log.Println(err.Error())
+		return err
 	}
 	//log.Println(string(rstByte))
 	err = msg.WriteMsg(stream, &models.JsonResponse{Code: 0, Msg: "Success", Result: string(rstByte)})

tasks/ipv6ClientServer.go

@@ -0,0 +1,90 @@
+package tasks
+
+import (
+	"github.com/OpenIoTHub/gateway-go/v2/chans"
+	"github.com/OpenIoTHub/gateway-go/v2/config"
+	"github.com/OpenIoTHub/gateway-go/v2/netservice/handle"
+	"github.com/OpenIoTHub/utils/v2/models"
+	"github.com/OpenIoTHub/utils/v2/msg"
+	"github.com/libp2p/go-yamux"
+	"log"
+	"net"
+	"time"
+)
+
+func RunTasks() {
+	go ipv6ServerTask()
+	go ipv6ClientTask()
+}
+
+// Ipv6ClientTask 接收配置创建新的Client handle
+func ipv6ClientTask() {
+	//	主动连接访问者的APP
+	for remoteIpv6Server := range chans.ClientTaskChan {
+		ip := remoteIpv6Server.Ipv6AddrIp
+		port := remoteIpv6Server.Ipv6AddrPort
+		runId := remoteIpv6Server.RunId
+		//	使用配置创建连接，并且发送带RunId的凭证给访问者
+		ipv6conn, err := net.DialTCP("tcp", nil, &net.TCPAddr{
+			IP:   net.ParseIP(ip),
+			Port: port,
+		})
+		if err != nil {
+			continue
+		}
+		//TODO 发送凭证
+		runIdMsg := &models.Msg{
+			MsgType:    "RunId",
+			MsgContent: runId,
+		}
+		err = msg.WriteMsg(ipv6conn, runIdMsg)
+		if err != nil {
+			ipv6conn.Close()
+			return
+		}
+		//创建session，session handle
+		yamuxConfig := yamux.DefaultConfig()
+		//remoteIpv6Server.EnableKeepAlive = false
+		session, err := yamux.Server(ipv6conn, yamuxConfig)
+		if err != nil {
+			ipv6conn.Close()
+			return
+		}
+		log.Printf("ipv6 p2p client login OK!")
+		go handle.HandleSession(session, "")
+	}
+}
+
+func ipv6ServerTask() {
+	listener, err := net.ListenTCP("tcp6", &net.TCPAddr{})
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	listenerPort := listener.Addr().(*net.TCPAddr).Port
+	log.Println("ipv6 server listening on", listenerPort)
+	config.Ipv6ListenTcpHandlePort = listenerPort
+	//接受验证连通性，接受连接和服务请求
+	for {
+		conn, err := listener.AcceptTCP()
+		if err != nil {
+			time.Sleep(100 * time.Millisecond)
+			continue
+		}
+		//	验证token，回复
+		go ipv6ClientHandle(conn)
+	}
+}
+
+func ipv6ClientHandle(conn net.Conn) {
+	rawMsg, err := msg.ReadMsg(conn)
+	if err != nil {
+		log.Println(err.Error() + "从stream读取数据错误")
+		conn.Close()
+		return
+	}
+	// TODO 验证token,RunId
+	_ = rawMsg
+	// Token为空
+	handle.HandleStream(conn, "")
+}