Add UDP puncher test

contree/conn.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"net"
+	"sync"
+	"time"
+)
+
+// Reference: https://github.com/pion/sctp/blob/master/association_test.go
+// Since UDP is connectionless, as a server, it doesn't know how to reply
+// simply using the `Write` method. So, to make it work, `disconnectedPacketConn`
+// will infer the last packet that it reads as the reply address for `Write`
+
+type disconnectedPacketConn struct { // nolint: unused
+	mu    sync.RWMutex
+	rAddr net.Addr
+	pConn net.PacketConn
+}
+
+// Read
+func (c *disconnectedPacketConn) Read(p []byte) (int, error) {
+	i, rAddr, err := c.pConn.ReadFrom(p)
+	if err != nil {
+		return 0, err
+	}
+
+	c.mu.Lock()
+	c.rAddr = rAddr
+	c.mu.Unlock()
+
+	return i, err
+}
+
+// Write writes len(p) bytes from p to the DTLS connection
+func (c *disconnectedPacketConn) Write(p []byte) (n int, err error) {
+	return c.pConn.WriteTo(p, c.RemoteAddr())
+}
+
+// Close closes the conn and releases any Read calls
+func (c *disconnectedPacketConn) Close() error {
+	return c.pConn.Close()
+}
+
+// LocalAddr is a stub
+func (c *disconnectedPacketConn) LocalAddr() net.Addr {
+	if c.pConn != nil {
+		return c.pConn.LocalAddr()
+	}
+	return nil
+}
+
+// RemoteAddr is a stub
+func (c *disconnectedPacketConn) RemoteAddr() net.Addr {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	return c.rAddr
+}
+
+// SetDeadline is a stub
+func (c *disconnectedPacketConn) SetDeadline(t time.Time) error {
+	return nil
+}
+
+// SetReadDeadline is a stub
+func (c *disconnectedPacketConn) SetReadDeadline(t time.Time) error {
+	return nil
+}
+
+// SetWriteDeadline is a stub
+func (c *disconnectedPacketConn) SetWriteDeadline(t time.Time) error {
+	return nil
+}

contree/dialer.go

@@ -0,0 +1,86 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	"github.com/pion/logging"
+	"github.com/pion/sctp"
+)
+
+func dial(address string) {
+	// Dial the target.
+	dialConn, err := net.Dial("udp", address)
+	if err != nil {
+		log.Panic(err)
+	}
+	defer func() {
+		if closeErr := dialConn.Close(); closeErr != nil {
+			panic(err)
+		}
+	}()
+	fmt.Println("Dialed UDP")
+
+	// Create the client.
+	config := sctp.Config{
+		NetConn:       dialConn,
+		LoggerFactory: logging.NewDefaultLoggerFactory(),
+	}
+	client, err := sctp.Client(config)
+	if err != nil {
+		log.Panic(err)
+	}
+	defer func() {
+		if closeErr := client.Close(); closeErr != nil {
+			panic(err)
+		}
+	}()
+	fmt.Println("Created client")
+
+	// Create the stream.
+	stream, err := client.OpenStream(0, sctp.PayloadTypeWebRTCString)
+	if err != nil {
+		log.Panic(err)
+	}
+	defer func() {
+		if closeErr := stream.Close(); closeErr != nil {
+			panic(err)
+		}
+	}()
+	stream.SetReliabilityParams(false, sctp.ReliabilityTypeReliable, 10)
+	fmt.Println("Created stream")
+
+	// Writer
+	go func() {
+		enc := json.NewEncoder(stream)
+		msgNum := 1
+		for {
+			time.Sleep(2 * time.Second)
+			enc.Encode(Message{
+				seq: msgNum,
+				msg: "from dialer",
+			})
+			msgNum++
+		}
+	}()
+
+	// Reader
+	dec := json.NewDecoder(stream)
+	log.Println("dailer: Made a JSON stream")
+	for {
+		var msg Message
+		err := dec.Decode(&msg)
+		if err == nil {
+			fmt.Printf("dialer: Got msg %+v\n", msg)
+			if msg.msg == "bye" {
+				return
+			}
+		} else {
+			fmt.Printf("dialer: Got error %+v\n", err)
+		}
+	}
+
+}

contree/listener.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+
+	"github.com/pion/logging"
+	"github.com/pion/sctp"
+)
+
+func listen(ip net.IP, port int) {
+	addr := net.UDPAddr{
+		IP:   ip,
+		Port: port,
+	}
+
+	mainConn, err := net.ListenUDP("udp", &addr)
+	if err != nil {
+		log.Panic(err)
+	}
+	defer mainConn.Close()
+	fmt.Println("Created listener")
+
+	config := sctp.Config{
+		NetConn:       &disconnectedPacketConn{pConn: mainConn},
+		LoggerFactory: logging.NewDefaultLoggerFactory(),
+	}
+
+	for {
+		server, err := sctp.Server(config)
+		if err != nil {
+			log.Panic(err)
+		}
+		defer server.Close()
+		fmt.Println("Created server")
+		stream, err := server.AcceptStream()
+		if err != nil {
+			log.Panic(err)
+		}
+		stream.SetReliabilityParams(false, sctp.ReliabilityTypeReliable, 10)
+		go runListenStream(stream)
+	}
+}
+
+func runListenStream(stream *sctp.Stream) {
+	defer func() {
+		if closeErr := stream.Close(); closeErr != nil {
+			panic(closeErr)
+		}
+	}()
+
+	dec := json.NewDecoder(stream)
+	enc := json.NewEncoder(stream)
+	log.Println("Made a JSON stream")
+	msgNum := 1
+	for {
+		var msg Message
+		err := dec.Decode(&msg)
+		if err == nil {
+			fmt.Printf("Got msg %+v\n", msg)
+			if msg.msg == "bye" {
+				return
+			} else {
+				enc.Encode(Message{
+					seq: msgNum,
+					msg: fmt.Sprintf("from listener to msg %d", msg.seq),
+				})
+				msgNum++
+			}
+		} else {
+			fmt.Printf("Got error %+v\n", err)
+		}
+	}
+}

contree/main.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"net"
+)
+
+func main() {
+	var address string
+	var port int
+	var shouldDial bool
+
+	flag.StringVar(&address, "address", "127.0.0.1", "Address to dial or listen")
+	flag.IntVar(&port, "port", 10300, "Port to dial or listen")
+	flag.BoolVar(&shouldDial, "dial", false, "Whether to dial target address or not")
+	flag.Parse()
+
+	if shouldDial {
+		dial(fmt.Sprintf("%s:%d", address, port))
+	} else {
+		listen(net.ParseIP(address), port)
+	}
+
+}

contree/message.go

@@ -0,0 +1,6 @@
+package main
+
+type Message struct {
+	seq int
+	msg string
+}

udp-punch/handshakee.go

@@ -0,0 +1,73 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"strings"
+	"time"
+)
+
+func handshakee(handshakerAddress string, name, target string) {
+
+	// Get our address as a usable UDPAddr.
+	remoteAddr, _ := net.ResolveUDPAddr("udp", handshakerAddress)
+
+	// Get a random local port.
+	localAddr := &net.UDPAddr{}
+
+	// Start listening!
+	localConn, err := net.ListenUDP("udp", localAddr)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	go func() {
+		_, err := localConn.WriteTo([]byte(fmt.Sprintf("REG %s", name)), remoteAddr)
+		if err != nil {
+			panic(err)
+		}
+
+		if target != "" {
+			_, err := localConn.WriteTo([]byte(fmt.Sprintf("AWAIT %s", target)), remoteAddr)
+			if err != nil {
+				panic(err)
+			}
+		}
+	}()
+
+	listen(localConn, name)
+}
+
+func listen(conn *net.UDPConn, name string) {
+	fmt.Printf("Listening on %s\n", conn.LocalAddr().String())
+	for {
+		fmt.Println("listening")
+		buffer := make([]byte, 1024)
+		bytesRead, fromAddr, err := conn.ReadFromUDP(buffer)
+		if err != nil {
+			fmt.Println("[ERROR]", err)
+			continue
+		}
+		msg := string(buffer[0:bytesRead])
+		parts := strings.Split(msg, " ")
+		if parts[0] == "ARRIVED" {
+			otherAddr, _ := net.ResolveUDPAddr("udp", parts[1])
+			conn.WriteTo([]byte(fmt.Sprintf("HELLO %s", name)), otherAddr)
+			go chatWith(conn, name, parts[1])
+		} else if parts[0] == "HELLO" {
+			go chatWith(conn, name, fromAddr.String())
+		} else {
+			fmt.Printf("Unhandled %+v", parts)
+		}
+	}
+
+}
+
+func chatWith(conn *net.UDPConn, name string, otherClient string) {
+	otherAddr, _ := net.ResolveUDPAddr("udp", otherClient)
+	for {
+		conn.WriteTo([]byte(fmt.Sprintf("Hello from %s", name)), otherAddr)
+		time.Sleep(5 * time.Second)
+	}
+}

udp-punch/handshaker.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"strings"
+)
+
+// AddressKey is a key that represents an ip address and port.
+type AddressKey string
+
+type MessageBox struct {
+	name     string              // String for this messagebox
+	wavingAt map[string]struct{} // waving at other names
+}
+
+var clientsMap map[AddressKey]*MessageBox = make(map[AddressKey]*MessageBox)
+
+func IPToAddressKey(addr *net.UDPAddr) (a AddressKey) {
+	return AddressKey(addr.String())
+}
+
+func AddressKeyToIP(a AddressKey) *net.UDPAddr {
+	addr, _ := net.ResolveUDPAddr("udp", string(a))
+	return addr
+}
+
+func handshaker(address string) {
+	fmt.Println("Starting handshaker...")
+	addr, err := net.ResolveUDPAddr("udp", address)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	localConn, err := net.ListenUDP("udp", addr)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Begin the Eternal Listen (tm)
+	for {
+		buffer := make([]byte, 1024)
+		bytesRead, remoteAddr, err := localConn.ReadFromUDP(buffer)
+		if err != nil {
+			panic(err)
+		}
+
+		clientKey := IPToAddressKey(remoteAddr)
+
+		msg := string(buffer[0:bytesRead])
+		parts := strings.Split(msg, " ")
+
+		fmt.Println("[INCOMING]", msg)
+		//if incoming.
+		if parts[0] == "REG" {
+			if _, ok := clientsMap[clientKey]; !ok {
+				clientsMap[clientKey] = new(MessageBox)
+				clientsMap[clientKey].wavingAt = make(map[string]struct{})
+			}
+			clientsMap[clientKey].name = parts[1]
+			// Check if any clients are waiting this target and send arrival msg.
+			for otherClientKey, mbox := range clientsMap {
+				if _, ok := mbox.wavingAt[parts[1]]; ok {
+					targetAddress := AddressKeyToIP(otherClientKey)
+					localConn.WriteTo([]byte(fmt.Sprintf("ARRIVED %s", clientKey)), targetAddress)
+					delete(mbox.wavingAt, parts[1])
+				}
+			}
+		} else if parts[0] == "AWAIT" {
+			mbox, ok := clientsMap[clientKey]
+			if !ok {
+				continue
+			}
+			var matched = false
+			for otherClientKey, otherMbox := range clientsMap {
+				if otherMbox.name == parts[1] {
+					targetAddress := AddressKeyToIP(clientKey)
+					localConn.WriteTo([]byte(fmt.Sprintf("ARRIVED %s", otherClientKey)), targetAddress)
+					matched = true
+				}
+			}
+			if !matched {
+				if _, ok := mbox.wavingAt[parts[1]]; !ok {
+					mbox.wavingAt[parts[1]] = struct{}{}
+				}
+			}
+		}
+	}
+}

udp-punch/main.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+)
+
+func main() {
+	var address string
+	var name string
+	var await string
+	var port int
+	var shouldDial bool
+
+	flag.StringVar(&name, "name", "Alpha", "Our name")
+	flag.StringVar(&await, "await", "", "Target's name to await")
+	flag.StringVar(&address, "address", "127.0.0.1", "Address to dial or listen")
+	flag.IntVar(&port, "port", 10300, "Port to dial or listen")
+	flag.BoolVar(&shouldDial, "dial", false, "Whether to dial target address or not")
+	flag.Parse()
+
+	if shouldDial {
+		handshakee(fmt.Sprintf("%s:%d", address, port), name, await)
+	} else {
+		handshaker(fmt.Sprintf("%s:%d", address, port))
+	}
+
+}