dissertation-2-code/udp/listener.go

package udp

import (
	"context"
	"log"
	"mpbl3p/proxy"
	"net"
	"time"
)

type ComparableUdpAddress struct {
	IP   [16]byte
	Port int
	Zone string
}

func fromUdpAddress(address net.UDPAddr) ComparableUdpAddress {
	var ip [16]byte
	for i, b := range []byte(address.IP) {
		ip[i] = b
	}

	return ComparableUdpAddress{
		IP:   ip,
		Port: address.Port,
		Zone: address.Zone,
	}
}

func NewListener(ctx context.Context, p *proxy.Proxy, local string, v func() proxy.MacVerifier, g func() proxy.MacGenerator, c func() Congestion) error {
	laddr, err := net.ResolveUDPAddr("udp", local)
	if err != nil {
		return err
	}

	pconn, err := net.ListenUDP("udp", laddr)
	if err != nil {
		return err
	}

	err = pconn.SetWriteBuffer(0)
	if err != nil {
		panic(err)
	}

	receivedConnections := make(map[ComparableUdpAddress]*Flow)

	go func() {
		for ctx.Err() == nil {
			buf := make([]byte, 6000)

			if err := pconn.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
				panic(err)
			}
			n, addr, err := pconn.ReadFromUDP(buf)
			if err != nil {
				if e, ok := err.(net.Error); ok && e.Timeout() {
					continue
				}
				panic(err)
			}

			raddr := fromUdpAddress(*addr)
			if f, exists := receivedConnections[raddr]; exists {
				log.Println("existing flow. queuing...")
				if err := f.queueDatagram(ctx, buf[:n]); err != nil {

				}
				log.Println("queued")
				continue
			}

			v := v()
			g := g()

			f := newFlow(c(), v)

			f.writer = pconn
			f.raddr = addr
			f.isAlive = true

			log.Printf("received new udp connection: %v\n", f)

			go f.earlyUpdateLoop(ctx, g, 0)

			receivedConnections[raddr] = &f

			p.AddConsumer(ctx, &f, g)
			p.AddProducer(ctx, &f, v)

			log.Println("handling...")
			if err := f.queueDatagram(ctx, buf[:n]); err != nil {
				return
			}
			log.Println("handled")
		}
	}()

	return nil
}