Initial commit of stream reuse

cli/up.go

@@ -33,6 +33,8 @@ var (
 	// RevLookup allow quick lookups of an incoming stream
 	// for security before accepting or responding to any data.
 	RevLookup map[string]bool
+	// activeStreams is a map of active streams to a peer
+	activeStreams map[string]network.Stream
 )
 
 // Up creates and brings up a Hyprspace Interface.
@@ -180,23 +182,36 @@ func UpRun(r *cmd.Root, c *cmd.Sub) {
 
 	fmt.Println("[+] Network Setup Complete...Waiting on Node Discovery")
 	// Listen For New Packets on TUN Interface
-	packet := make([]byte, 1420)
+	activeStreams = make(map[string]network.Stream)
+	var packet = make([]byte, 1420)
 	var stream network.Stream
-	var header *ipv4.Header
+	var ok bool
 	var plen int
+	var dst string
 	for {
-		plen, err = tunDev.Iface.Read(packet)
-		checkErr(err)
-		header, _ = ipv4.ParseHeader(packet)
-		_, ok := cfg.Peers[header.Dst.String()]
+		plen, err = tunDev.Iface.Read([]byte(packet))
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		dst = net.IPv4(packet[16], packet[17], packet[18], packet[19]).String()
+		stream, ok = activeStreams[dst]
 		if ok {
-			stream, err = host.NewStream(ctx, peerTable[header.Dst.String()], p2p.Protocol)
+			_, err = stream.Write([]byte(packet[:plen]))
+			if err == nil {
+				continue
+			}
+			stream.Close()
+			ok = false
+		}
+		if !ok {
+			stream, err = host.NewStream(ctx, peerTable[dst])
 			if err != nil {
 				log.Println(err)
 				continue
 			}
-			stream.Write(packet[:plen])
-			stream.Close()
+			stream.Write([]byte(packet[:plen]))
+			activeStreams[dst] = stream
 		}
 	}
 }
@@ -236,9 +251,27 @@ func streamHandler(stream network.Stream) {
 	// If the remote node ID isn't in the list of known nodes don't respond.
 	if _, ok := RevLookup[stream.Conn().RemotePeer().Pretty()]; !ok {
 		stream.Reset()
+		return
+	}
+	headers := io.LimitReader(stream, 20)
+	var err error
+	var header *ipv4.Header
+	var packetHeader = make([]byte, 20)
+
+	for {
+		_, err = headers.Read([]byte(packetHeader))
+		if err != nil {
+			stream.Close()
+			return
+		}
+		header, err = ipv4.ParseHeader(packetHeader)
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		tunDev.Iface.Write(packetHeader)
+		io.CopyN(tunDev.Iface.ReadWriteCloser, stream, int64(header.TotalLen))
 	}
-	io.Copy(tunDev.Iface.ReadWriteCloser, stream)
-	stream.Close()
 }
 
 func prettyDiscovery(ctx context.Context, node host.Host, peerTable map[string]peer.ID) {

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/libp2p/go-libp2p-quic-transport v0.12.0
 	github.com/libp2p/go-tcp-transport v0.2.8
 	github.com/multiformats/go-multiaddr v0.3.3
+	github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091
 	github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8
 	github.com/tcnksm/go-latest v0.0.0-20170313132115-e3007ae9052e
 	github.com/vishvananda/netlink v1.1.0

go.sum

@@ -838,6 +838,8 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/smola/gocompat v0.2.0/go.mod h1:1B0MlxbmoZNo3h8guHp8HztB3BSYR5itql9qtVc0ypY=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
+github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091 h1:1zN6ImoqhSJhN8hGXFaJlSC8msLmIbX8bFqOfWLKw0w=
+github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091/go.mod h1:N20Z5Y8oye9a7HmytmZ+tr8Q2vlP0tAHP13kTHzwvQY=
 github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 h1:TG/diQgUe0pntT/2D9tmUCz4VNwm9MfrtPr0SU2qSX8=
 github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8/go.mod h1:P5HUIBuIWKbyjl083/loAegFkfbFNx5i2qEP4CNbm7E=
 github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=