rewrite the QUIC proxy used in the integrationtests

The new QUIC proxy allows to listen on “:0”, which allows us to get rid
of all “address already in use” errors. The constructor now takes an
Opts struct, which makes configuring it easier.

integrationtests/drop_test.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"os/exec"
 	"strconv"
-	"time"
 
 	_ "github.com/lucas-clemente/quic-clients" // download clients
 	"github.com/lucas-clemente/quic-go/integrationtests/proxy"
@@ -21,21 +20,21 @@ var _ = Describe("Drop Proxy", func() {
 		dataMan.GenerateData(dataLen)
 	})
 
-	var dropproxy *proxy.UDPProxy
+	var proxy *quicproxy.QuicProxy
 
-	runDropTest := func(incomingPacketDropper, outgoingPacketDropper proxy.DropCallback, version protocol.VersionNumber) {
-		proxyPort := 12345
-
-		iPort, _ := strconv.Atoi(port)
+	runDropTest := func(dropCallback quicproxy.DropCallback, version protocol.VersionNumber) {
 		var err error
-		dropproxy, err = proxy.NewUDPProxy(proxyPort, "localhost", iPort, incomingPacketDropper, outgoingPacketDropper, 0, 0)
+		proxy, err = quicproxy.NewQuicProxy("localhost:0", quicproxy.Opts{
+			RemoteAddr: "localhost:" + port,
+			DropPacket: dropCallback,
+		})
 		Expect(err).ToNot(HaveOccurred())
 
 		command := exec.Command(
 			clientPath,
 			"--quic-version="+strconv.Itoa(int(version)),
 			"--host=127.0.0.1",
-			"--port="+strconv.Itoa(proxyPort),
+			"--port="+strconv.Itoa(proxy.LocalPort()),
 			"https://quic.clemente.io/data",
 		)
 
@@ -47,8 +46,7 @@ var _ = Describe("Drop Proxy", func() {
 	}
 
 	AfterEach(func() {
-		dropproxy.Stop()
-		time.Sleep(time.Millisecond)
+		Expect(proxy.Close()).To(Succeed())
 	})
 
 	for i := range protocol.SupportedVersions {
@@ -64,11 +62,15 @@ var _ = Describe("Drop Proxy", func() {
 				}
 
 				It("gets a file when many outgoing packets are dropped", func() {
-					runDropTest(nil, dropper, version)
+					runDropTest(func(d quicproxy.Direction, p protocol.PacketNumber) bool {
+						return d == quicproxy.DirectionOutgoing && dropper(p)
+					}, version)
 				})
 
 				It("gets a file when many incoming packets are dropped", func() {
-					runDropTest(dropper, nil, version)
+					runDropTest(func(d quicproxy.Direction, p protocol.PacketNumber) bool {
+						return d == quicproxy.DirectionIncoming && dropper(p)
+					}, version)
 				})
 			})
 		})

integrationtests/proxy/proxy.go

@@ -0,0 +1,227 @@
+package quicproxy
+
+import (
+	"bytes"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/lucas-clemente/quic-go"
+	"github.com/lucas-clemente/quic-go/protocol"
+)
+
+// Connection is a UDP connection
+type connection struct {
+	ClientAddr *net.UDPAddr // Address of the client
+	ServerConn *net.UDPConn // UDP connection to server
+
+	incomingPacketCounter uint64
+	outgoingPacketCounter uint64
+}
+
+// Direction is the direction a packet is sent.
+type Direction int
+
+const (
+	// DirectionIncoming is the direction from the client to the server.
+	DirectionIncoming Direction = iota
+	// DirectionOutgoing is the direction from the server to the client.
+	DirectionOutgoing
+)
+
+// DropCallback is a callback that determines which packet gets dropped.
+type DropCallback func(Direction, protocol.PacketNumber) bool
+
+// NoDropper doesn't drop packets.
+var NoDropper DropCallback = func(Direction, protocol.PacketNumber) bool {
+	return false
+}
+
+// DelayCallback is a callback that determines how much delay to apply to a packet.
+type DelayCallback func(Direction, protocol.PacketNumber) time.Duration
+
+// NoDelay doesn't apply a delay.
+var NoDelay DelayCallback = func(Direction, protocol.PacketNumber) time.Duration {
+	return 0
+}
+
+// Opts are proxy options.
+type Opts struct {
+	// The address this proxy proxies packets to.
+	RemoteAddr string
+	// DropPacket determines whether a packet gets dropped.
+	DropPacket DropCallback
+	// DelayPacket determines how long a packet gets delayed. This allows
+	// simulating a connection with non-zero RTTs.
+	// Note that the RTT is the sum of the delay for the incoming and the outgoing packet.
+	DelayPacket DelayCallback
+}
+
+// QuicProxy is a QUIC proxy that can drop and delay packets.
+type QuicProxy struct {
+	mutex sync.Mutex
+
+	conn       *net.UDPConn
+	serverAddr *net.UDPAddr
+
+	dropPacket  DropCallback
+	delayPacket DelayCallback
+
+	// Mapping from client addresses (as host:port) to connection
+	clientDict map[string]*connection
+}
+
+// NewQuicProxy creates a new UDP proxy
+func NewQuicProxy(local string, opts Opts) (*QuicProxy, error) {
+	laddr, err := net.ResolveUDPAddr("udp", local)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := net.ListenUDP("udp", laddr)
+	if err != nil {
+		return nil, err
+	}
+	raddr, err := net.ResolveUDPAddr("udp", opts.RemoteAddr)
+	if err != nil {
+		return nil, err
+	}
+
+	packetDropper := NoDropper
+	if opts.DropPacket != nil {
+		packetDropper = opts.DropPacket
+	}
+
+	packetDelayer := NoDelay
+	if opts.DelayPacket != nil {
+		packetDelayer = opts.DelayPacket
+	}
+
+	p := QuicProxy{
+		clientDict:  make(map[string]*connection),
+		conn:        conn,
+		serverAddr:  raddr,
+		dropPacket:  packetDropper,
+		delayPacket: packetDelayer,
+	}
+
+	go p.runProxy()
+	return &p, nil
+}
+
+// Close stops the UDP Proxy
+func (p *QuicProxy) Close() error {
+	return p.conn.Close()
+}
+
+// LocalAddr is the address the proxy is listening on.
+func (p *QuicProxy) LocalAddr() net.Addr {
+	return p.conn.LocalAddr()
+}
+
+func (p *QuicProxy) LocalPort() int {
+	return p.conn.LocalAddr().(*net.UDPAddr).Port
+}
+
+func (p *QuicProxy) newConnection(cliAddr *net.UDPAddr) (*connection, error) {
+	srvudp, err := net.DialUDP("udp", nil, p.serverAddr)
+	if err != nil {
+		return nil, err
+	}
+	return &connection{
+		ClientAddr: cliAddr,
+		ServerConn: srvudp,
+	}, nil
+}
+
+// runProxy listens on the proxy address and handles incoming packets.
+func (p *QuicProxy) runProxy() error {
+	for {
+		buffer := make([]byte, protocol.MaxPacketSize)
+		n, cliaddr, err := p.conn.ReadFromUDP(buffer)
+		if err != nil {
+			return err
+		}
+		raw := buffer[0:n]
+
+		saddr := cliaddr.String()
+		p.mutex.Lock()
+		conn, ok := p.clientDict[saddr]
+
+		if !ok {
+			conn, err = p.newConnection(cliaddr)
+			if err != nil {
+				p.mutex.Unlock()
+				return err
+			}
+			p.clientDict[saddr] = conn
+			go p.runConnection(conn)
+		}
+		p.mutex.Unlock()
+
+		atomic.AddUint64(&conn.incomingPacketCounter, 1)
+
+		r := bytes.NewReader(raw)
+		hdr, err := quic.ParsePublicHeader(r, protocol.PerspectiveClient)
+		if err != nil {
+			return err
+		}
+
+		if p.dropPacket(DirectionIncoming, hdr.PacketNumber) {
+			continue
+		}
+
+		// Send the packet to the server
+		delay := p.delayPacket(DirectionIncoming, hdr.PacketNumber)
+		if delay != 0 {
+			time.AfterFunc(delay, func() {
+				// TODO: handle error
+				_, _ = conn.ServerConn.Write(raw)
+			})
+		} else {
+			_, err := conn.ServerConn.Write(raw)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+// runConnection handles packets from server to a single client
+func (p *QuicProxy) runConnection(conn *connection) error {
+	for {
+		buffer := make([]byte, protocol.MaxPacketSize)
+		n, err := conn.ServerConn.Read(buffer)
+		if err != nil {
+			return err
+		}
+		raw := buffer[0:n]
+
+		// TODO: Switch back to using the public header once Chrome properly sets the type byte.
+		// r := bytes.NewReader(raw)
+		// , err := quic.ParsePublicHeader(r, protocol.PerspectiveServer)
+		// if err != nil {
+		// return err
+		// }
+
+		v := atomic.AddUint64(&conn.outgoingPacketCounter, 1)
+
+		packetNumber := protocol.PacketNumber(v)
+		if p.dropPacket(DirectionOutgoing, packetNumber) {
+			continue
+		}
+
+		delay := p.delayPacket(DirectionOutgoing, packetNumber)
+		if delay != 0 {
+			time.AfterFunc(delay, func() {
+				// TODO: handle error
+				_, _ = p.conn.WriteToUDP(raw, conn.ClientAddr)
+			})
+		} else {
+			_, err := p.conn.WriteToUDP(raw, conn.ClientAddr)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}

integrationtests/proxy/proxy_suite_test.go

@@ -1,4 +1,4 @@
-package proxy
+package quicproxy
 
 import (
 	. "github.com/onsi/ginkgo"
@@ -9,5 +9,5 @@ import (
 
 func TestQuicGo(t *testing.T) {
 	RegisterFailHandler(Fail)
-	RunSpecs(t, "UDP Proxy")
+	RunSpecs(t, "QUIC Proxy")
 }

integrationtests/proxy/proxy_test.go

@@ -0,0 +1,337 @@
+package quicproxy
+
+import (
+	"bytes"
+	"net"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/lucas-clemente/quic-go"
+	"github.com/lucas-clemente/quic-go/protocol"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+type packetData []byte
+
+var _ = Describe("QUIC Proxy", func() {
+	var serverAddr string
+
+	makePacket := func(p protocol.PacketNumber, payload []byte) []byte {
+		b := &bytes.Buffer{}
+		hdr := quic.PublicHeader{
+			PacketNumber:         p,
+			PacketNumberLen:      protocol.PacketNumberLen6,
+			ConnectionID:         1337,
+			TruncateConnectionID: false,
+		}
+		hdr.Write(b, protocol.VersionWhatever, protocol.PerspectiveServer)
+		raw := b.Bytes()
+		raw = append(raw, payload...)
+		return raw
+	}
+
+	BeforeEach(func() {
+		serverAddr = "localhost:7331"
+	})
+
+	Context("Proxy setup and teardown", func() {
+		It("sets up the UDPProxy", func() {
+			proxy, err := NewQuicProxy("localhost:13370", Opts{RemoteAddr: serverAddr})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(proxy.clientDict).To(HaveLen(0))
+
+			// check that port 13370 is in use
+			addr, err := net.ResolveUDPAddr("udp", "localhost:13370")
+			Expect(err).ToNot(HaveOccurred())
+			_, err = net.ListenUDP("udp", addr)
+			Expect(err).To(MatchError("listen udp 127.0.0.1:13370: bind: address already in use"))
+
+			err = proxy.Close() // stopping is tested in the next test
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("stops the UDPProxy", func() {
+			proxy, err := NewQuicProxy("localhost:13371", Opts{RemoteAddr: serverAddr})
+			Expect(err).ToNot(HaveOccurred())
+			err = proxy.Close()
+			Expect(err).ToNot(HaveOccurred())
+
+			// check that port 13371 is not in use anymore
+			addr, err := net.ResolveUDPAddr("udp", "localhost:13371")
+			Expect(err).ToNot(HaveOccurred())
+			ln, err := net.ListenUDP("udp", addr)
+			Expect(err).ToNot(HaveOccurred())
+			err = ln.Close()
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		It("has the correct LocalAddr and LocalPort", func() {
+			proxy, err := NewQuicProxy("localhost:13372", Opts{RemoteAddr: serverAddr})
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(proxy.LocalAddr().String()).To(Equal("127.0.0.1:13372"))
+			Expect(proxy.LocalPort()).To(Equal(13372))
+
+			err = proxy.Close()
+			Expect(err).ToNot(HaveOccurred())
+		})
+	})
+
+	Context("Proxy tests", func() {
+		var (
+			serverConn            *net.UDPConn
+			serverReceivedPackets []packetData
+			serverNumPacketsSent  int
+			clientConn            *net.UDPConn
+			proxy                 *QuicProxy
+		)
+
+		// start the proxy on port 10001
+		startProxy := func(opts Opts) {
+			var err error
+			proxy, err = NewQuicProxy("localhost:10001", opts)
+			Expect(err).ToNot(HaveOccurred())
+			clientConn, err = net.DialUDP("udp", nil, proxy.LocalAddr().(*net.UDPAddr))
+			Expect(err).ToNot(HaveOccurred())
+		}
+
+		BeforeEach(func() {
+			serverReceivedPackets = serverReceivedPackets[:0]
+			serverNumPacketsSent = 0
+
+			// setup a dump UDP server on port 7331
+			// in production this would be a QUIC server
+			raddr, err := net.ResolveUDPAddr("udp", serverAddr)
+			Expect(err).ToNot(HaveOccurred())
+			serverConn, err = net.ListenUDP("udp", raddr)
+			Expect(err).ToNot(HaveOccurred())
+
+			go func() {
+				for {
+					buf := make([]byte, protocol.MaxPacketSize)
+					// the ReadFromUDP will error as soon as the UDP conn is closed
+					n, addr, err2 := serverConn.ReadFromUDP(buf)
+					if err2 != nil {
+						return
+					}
+					data := buf[0:n]
+					serverReceivedPackets = append(serverReceivedPackets, packetData(data))
+					// echo the packet
+					serverConn.WriteToUDP(data, addr)
+					serverNumPacketsSent++
+				}
+			}()
+		})
+
+		AfterEach(func() {
+			err := proxy.Close()
+			Expect(err).ToNot(HaveOccurred())
+			err = serverConn.Close()
+			Expect(err).ToNot(HaveOccurred())
+			err = clientConn.Close()
+			Expect(err).ToNot(HaveOccurred())
+			time.Sleep(200 * time.Millisecond)
+		})
+
+		Context("no packet drop", func() {
+			It("relays packets from the client to the server", func() {
+				startProxy(Opts{RemoteAddr: serverAddr})
+				// send the first packet
+				_, err := clientConn.Write(makePacket(1, []byte("foobar")))
+				Expect(err).ToNot(HaveOccurred())
+
+				Eventually(func() map[string]*connection { return proxy.clientDict }).Should(HaveLen(1))
+				var conn *connection
+				for _, conn = range proxy.clientDict {
+					Expect(atomic.LoadUint64(&conn.incomingPacketCounter)).To(Equal(uint64(1)))
+				}
+
+				// send the second packet
+				_, err = clientConn.Write(makePacket(2, []byte("decafbad")))
+				Expect(err).ToNot(HaveOccurred())
+
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(2))
+				Expect(proxy.clientDict).To(HaveLen(1))
+				Expect(string(serverReceivedPackets[0])).To(ContainSubstring("foobar"))
+				Expect(string(serverReceivedPackets[1])).To(ContainSubstring("decafbad"))
+			})
+
+			It("relays packets from the server to the client", func() {
+				startProxy(Opts{RemoteAddr: serverAddr})
+				// send the first packet
+				_, err := clientConn.Write(makePacket(1, []byte("foobar")))
+				Expect(err).ToNot(HaveOccurred())
+
+				Eventually(func() map[string]*connection { return proxy.clientDict }).Should(HaveLen(1))
+				var key string
+				var conn *connection
+				for key, conn = range proxy.clientDict {
+					Eventually(func() uint64 { return atomic.LoadUint64(&conn.outgoingPacketCounter) }).Should(Equal(uint64(1)))
+				}
+
+				// send the second packet
+				_, err = clientConn.Write(makePacket(2, []byte("decafbad")))
+				Expect(err).ToNot(HaveOccurred())
+
+				Expect(proxy.clientDict).To(HaveLen(1))
+				Eventually(func() uint64 { return proxy.clientDict[key].outgoingPacketCounter }).Should(BeEquivalentTo(2))
+
+				var clientReceivedPackets []packetData
+				// receive the packets echoed by the server on client side
+				go func() {
+					for {
+						buf := make([]byte, protocol.MaxPacketSize)
+						// the ReadFromUDP will error as soon as the UDP conn is closed
+						n, _, err2 := clientConn.ReadFromUDP(buf)
+						if err2 != nil {
+							return
+						}
+						data := buf[0:n]
+						clientReceivedPackets = append(clientReceivedPackets, packetData(data))
+					}
+				}()
+
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(2))
+				Expect(serverReceivedPackets).To(HaveLen(2))
+				Expect(serverNumPacketsSent).To(Equal(2))
+				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(2))
+				Expect(string(clientReceivedPackets[0])).To(ContainSubstring("foobar"))
+				Expect(string(clientReceivedPackets[1])).To(ContainSubstring("decafbad"))
+			})
+		})
+
+		Context("Drop Callbacks", func() {
+			It("drops incoming packets", func() {
+				opts := Opts{
+					RemoteAddr: serverAddr,
+					DropPacket: func(d Direction, p protocol.PacketNumber) bool {
+						return d == DirectionIncoming && p%2 == 0
+					},
+				}
+				startProxy(opts)
+
+				// send 6 packets
+				for i := 1; i <= 6; i++ {
+					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
+					Expect(err).ToNot(HaveOccurred())
+				}
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
+				Consistently(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
+			})
+
+			It("drops outgoing packets", func() {
+				opts := Opts{
+					RemoteAddr: serverAddr,
+					DropPacket: func(d Direction, p protocol.PacketNumber) bool {
+						return d == DirectionOutgoing && p%2 == 0
+					},
+				}
+				startProxy(opts)
+
+				var clientReceivedPackets []packetData
+				// receive the packets echoed by the server on client side
+				go func() {
+					for {
+						buf := make([]byte, protocol.MaxPacketSize)
+						// the ReadFromUDP will error as soon as the UDP conn is closed
+						n, _, err2 := clientConn.ReadFromUDP(buf)
+						if err2 != nil {
+							return
+						}
+						data := buf[0:n]
+						clientReceivedPackets = append(clientReceivedPackets, packetData(data))
+					}
+				}()
+
+				// send 6 packets
+				for i := 1; i <= 6; i++ {
+					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
+					Expect(err).ToNot(HaveOccurred())
+				}
+
+				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(3))
+				Consistently(func() []packetData { return clientReceivedPackets }).Should(HaveLen(3))
+			})
+		})
+
+		Context("Delay Callback", func() {
+			It("delays incoming packets", func() {
+				opts := Opts{
+					RemoteAddr: serverAddr,
+					// delay packet 1 by 200 ms
+					// delay packet 2 by 400 ms
+					// ...
+					DelayPacket: func(d Direction, p protocol.PacketNumber) time.Duration {
+						if d == DirectionOutgoing {
+							return 0
+						}
+						return time.Duration(200*p) * time.Millisecond
+					},
+				}
+				startProxy(opts)
+
+				// send 3 packets
+				start := time.Now()
+				for i := 1; i <= 3; i++ {
+					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
+					Expect(err).ToNot(HaveOccurred())
+				}
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(1))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(200*time.Millisecond), 50*time.Millisecond))
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(2))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(400*time.Millisecond), 50*time.Millisecond))
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(600*time.Millisecond), 50*time.Millisecond))
+			})
+
+			It("delays outgoing packets", func() {
+				opts := Opts{
+					RemoteAddr: serverAddr,
+					// delay packet 1 by 200 ms
+					// delay packet 2 by 400 ms
+					// ...
+					DelayPacket: func(d Direction, p protocol.PacketNumber) time.Duration {
+						if d == DirectionIncoming {
+							return 0
+						}
+						return time.Duration(200*p) * time.Millisecond
+					},
+				}
+				startProxy(opts)
+
+				var clientReceivedPackets []packetData
+				// receive the packets echoed by the server on client side
+				go func() {
+					for {
+						buf := make([]byte, protocol.MaxPacketSize)
+						// the ReadFromUDP will error as soon as the UDP conn is closed
+						n, _, err2 := clientConn.ReadFromUDP(buf)
+						if err2 != nil {
+							return
+						}
+						data := buf[0:n]
+						clientReceivedPackets = append(clientReceivedPackets, packetData(data))
+					}
+				}()
+
+				// send 3 packets
+				start := time.Now()
+				for i := 1; i <= 3; i++ {
+					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
+					Expect(err).ToNot(HaveOccurred())
+				}
+				// the packets should have arrived immediately at the server
+				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
+				Expect(time.Now()).To(BeTemporally("~", start, 50*time.Millisecond))
+				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(1))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(200*time.Millisecond), 50*time.Millisecond))
+				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(2))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(400*time.Millisecond), 50*time.Millisecond))
+				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(3))
+				Expect(time.Now()).To(BeTemporally("~", start.Add(600*time.Millisecond), 50*time.Millisecond))
+			})
+		})
+	})
+})

integrationtests/proxy/rtt_generator.go

@@ -1,31 +0,0 @@
-package proxy
-
-import (
-	"math/rand"
-	"time"
-)
-
-type rttGenerator struct {
-	min time.Duration
-	max time.Duration
-}
-
-func newRttGenerator(min, max time.Duration) rttGenerator {
-	rand.Seed(time.Now().UnixNano())
-	return rttGenerator{
-		min: min,
-		max: max,
-	}
-}
-
-func (s *rttGenerator) getRTT() time.Duration {
-	if s.min == s.max {
-		return s.min
-	}
-
-	minns := s.min.Nanoseconds()
-	maxns := s.max.Nanoseconds()
-	rttns := rand.Int63n(maxns-minns) + minns
-
-	return time.Duration(rttns) * time.Nanosecond
-}

integrationtests/proxy/rtt_generator_test.go

@@ -1,56 +0,0 @@
-package proxy
-
-import (
-	"time"
-
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-)
-
-var _ = Describe("RTT settings", func() {
-	Context("no variance", func() {
-		It("always gets the same value", func() {
-			rttGen := newRttGenerator(10*time.Millisecond, 10*time.Millisecond)
-			for i := 0; i < 100; i++ {
-				Expect(rttGen.getRTT()).To(Equal(10 * time.Millisecond))
-			}
-		})
-	})
-
-	Context("random RTT", func() {
-		var rttGen rttGenerator
-
-		BeforeEach(func() {
-			rttGen = newRttGenerator(10*time.Millisecond, 30*time.Millisecond)
-		})
-
-		It("has the right mean value", func() {
-			var rttSum time.Duration
-			rep := 1000
-			for i := 0; i < rep; i++ {
-				rttSum += rttGen.getRTT()
-			}
-			averageRTT := rttSum.Nanoseconds() / 1000 / int64(rep) // in microseconds
-			Expect(averageRTT).To(BeNumerically("~", 20000, 1000)) // between 19 and 21 microseconds
-		})
-
-		It("covers the whole interval", func() {
-			var max time.Duration
-			min := time.Hour
-
-			rep := 1000
-			for i := 0; i < rep; i++ {
-				rtt := rttGen.getRTT()
-				if rtt > max {
-					max = rtt
-				}
-				if rtt < min {
-					min = rtt
-				}
-			}
-
-			Expect(min.Nanoseconds() / 1000).To(BeNumerically("<", 11000))
-			Expect(max.Nanoseconds() / 1000).To(BeNumerically(">", 29000))
-		})
-	})
-})

integrationtests/proxy/udp_proxy.go

@@ -1,172 +0,0 @@
-package proxy
-
-import (
-	"bytes"
-	"net"
-	"strconv"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/lucas-clemente/quic-go"
-	"github.com/lucas-clemente/quic-go/protocol"
-)
-
-// Connection is a UDP connection
-type connection struct {
-	ClientAddr *net.UDPAddr // Address of the client
-	ServerConn *net.UDPConn // UDP connection to server
-
-	incomingPacketCounter uint64
-	outgoingPacketCounter uint64
-}
-
-// DropCallback is a callback that determines which packet gets dropped
-type DropCallback func(protocol.PacketNumber) bool
-
-// UDPProxy is a UDP proxy
-type UDPProxy struct {
-	serverAddr *net.UDPAddr
-	mutex      sync.Mutex
-
-	proxyConn          *net.UDPConn
-	dropIncomingPacket DropCallback
-	dropOutgoingPacket DropCallback
-	rttGen             rttGenerator
-
-	// Mapping from client addresses (as host:port) to connection
-	clientDict map[string]*connection
-}
-
-// NewUDPProxy creates a new UDP proxy
-func NewUDPProxy(proxyPort int, serverAddress string, serverPort int, dropIncomingPacket, dropOutgoingPacket DropCallback, rttMin time.Duration, rttMax time.Duration) (*UDPProxy, error) {
-	dontDrop := func(p protocol.PacketNumber) bool {
-		return false
-	}
-
-	if dropIncomingPacket == nil {
-		dropIncomingPacket = dontDrop
-	}
-	if dropOutgoingPacket == nil {
-		dropOutgoingPacket = dontDrop
-	}
-
-	p := UDPProxy{
-		clientDict:         make(map[string]*connection),
-		dropIncomingPacket: dropIncomingPacket,
-		dropOutgoingPacket: dropOutgoingPacket,
-		rttGen:             newRttGenerator(rttMin, rttMax),
-	}
-
-	saddr, err := net.ResolveUDPAddr("udp", ":"+strconv.Itoa(proxyPort))
-	if err != nil {
-		return nil, err
-	}
-	pudp, err := net.ListenUDP("udp", saddr)
-	if err != nil {
-		return nil, err
-	}
-	p.proxyConn = pudp
-
-	srvaddr, err := net.ResolveUDPAddr("udp", serverAddress+":"+strconv.Itoa(serverPort))
-	if err != nil {
-		return nil, err
-	}
-	p.serverAddr = srvaddr
-
-	go p.runProxy()
-
-	return &p, nil
-}
-
-// Stop stops the UDP Proxy
-func (p *UDPProxy) Stop() {
-	p.proxyConn.Close()
-}
-
-func (p *UDPProxy) newConnection(cliAddr *net.UDPAddr) (*connection, error) {
-	var conn connection
-	conn.ClientAddr = cliAddr
-	srvudp, err := net.DialUDP("udp", nil, p.serverAddr)
-	if err != nil {
-		return nil, err
-	}
-	conn.ServerConn = srvudp
-	return &conn, nil
-}
-
-// runProxy handles inputs to Proxy port
-func (p *UDPProxy) runProxy() error {
-	for {
-		buffer := make([]byte, 1500)
-		n, cliaddr, err := p.proxyConn.ReadFromUDP(buffer[0:])
-		if err != nil {
-			return err
-		}
-
-		saddr := cliaddr.String()
-
-		p.mutex.Lock()
-		conn, ok := p.clientDict[saddr]
-
-		if !ok {
-			conn, err = p.newConnection(cliaddr)
-			if err != nil {
-				p.mutex.Unlock()
-				return err
-			}
-			p.clientDict[saddr] = conn
-			p.mutex.Unlock()
-			go p.runConnection(conn)
-		} else {
-			p.mutex.Unlock()
-		}
-
-		atomic.AddUint64(&conn.incomingPacketCounter, 1)
-
-		raw := buffer[0:n]
-		r := bytes.NewReader(raw)
-		hdr, err := quic.ParsePublicHeader(r, protocol.PerspectiveClient)
-		if err != nil {
-			return err
-		}
-
-		if !p.dropIncomingPacket(hdr.PacketNumber) {
-			// Relay to server
-			go func() {
-				time.Sleep(p.rttGen.getRTT() / 2)
-				conn.ServerConn.Write(raw)
-			}()
-		}
-	}
-}
-
-// runConnection handles packets from server to a single client
-func (p *UDPProxy) runConnection(conn *connection) error {
-	for {
-		buffer := make([]byte, 1500)
-		n, err := conn.ServerConn.Read(buffer[0:])
-		if err != nil {
-			return err
-		}
-
-		raw := buffer[0:n]
-
-		// TODO: Switch back to using the public header once Chrome properly sets the type byte.
-		// r := bytes.NewReader(raw)
-		// , err := quic.ParsePublicHeader(r, protocol.PerspectiveServer)
-		// if err != nil {
-		// return err
-		// }
-
-		v := atomic.AddUint64(&conn.outgoingPacketCounter, 1)
-
-		if !p.dropOutgoingPacket(protocol.PacketNumber(v)) {
-			// Relay it to client
-			go func() {
-				time.Sleep(p.rttGen.getRTT() / 2)
-				p.proxyConn.WriteToUDP(raw, conn.ClientAddr)
-			}()
-		}
-	}
-}

integrationtests/proxy/udp_proxy_test.go

@@ -1,239 +0,0 @@
-package proxy
-
-import (
-	"bytes"
-	"net"
-	"strconv"
-	"sync/atomic"
-	"time"
-
-	"github.com/lucas-clemente/quic-go"
-	"github.com/lucas-clemente/quic-go/protocol"
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-)
-
-type packetData []byte
-
-var _ = Describe("UDP Proxy", func() {
-	var serverPort int
-
-	makePacket := func(p protocol.PacketNumber, payload []byte) []byte {
-		b := &bytes.Buffer{}
-		hdr := quic.PublicHeader{
-			PacketNumber:         p,
-			PacketNumberLen:      protocol.PacketNumberLen6,
-			ConnectionID:         1337,
-			TruncateConnectionID: false,
-		}
-		hdr.Write(b, protocol.VersionWhatever, protocol.PerspectiveServer)
-		raw := b.Bytes()
-		raw = append(raw, payload...)
-		return raw
-	}
-
-	BeforeEach(func() {
-		serverPort = 7331
-	})
-
-	It("sets up the UDPProxy", func() {
-		proxy, err := NewUDPProxy(13370, "localhost", serverPort, nil, nil, 0, 0)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(proxy.clientDict).To(HaveLen(0))
-
-		// check that port 13370 is in use
-		addr, err := net.ResolveUDPAddr("udp", ":13370")
-		Expect(err).ToNot(HaveOccurred())
-		_, err = net.ListenUDP("udp", addr)
-		Expect(err).To(HaveOccurred())
-		Expect(err).To(MatchError("listen udp :13370: bind: address already in use"))
-
-		proxy.Stop() // stopping is tested in the next test
-	})
-
-	It("stops the UDPProxy", func() {
-		proxy, err := NewUDPProxy(13371, "localhost", serverPort, nil, nil, 0, 0)
-		Expect(err).ToNot(HaveOccurred())
-		proxy.Stop()
-
-		// check that port 13370 is not in use anymore
-		addr, err := net.ResolveUDPAddr("udp", ":13371")
-		Expect(err).ToNot(HaveOccurred())
-		ln, err := net.ListenUDP("udp", addr)
-		Expect(err).ToNot(HaveOccurred())
-		ln.Close()
-	})
-
-	Context("Proxy tests", func() {
-		var serverConn *net.UDPConn
-		var serverReceivedPackets []packetData
-		var serverNumPacketsSent int
-		var clientConn *net.UDPConn
-		var proxy *UDPProxy
-
-		BeforeEach(func() {
-			var err error
-			serverReceivedPackets = serverReceivedPackets[:0]
-			serverNumPacketsSent = 0
-
-			// setup a UDP server on port 7331
-			serverAddr, err := net.ResolveUDPAddr("udp", ":"+strconv.Itoa(serverPort))
-			Expect(err).ToNot(HaveOccurred())
-			serverConn, err = net.ListenUDP("udp", serverAddr)
-			Expect(err).ToNot(HaveOccurred())
-
-			proxyAddr, err := net.ResolveUDPAddr("udp", ":10001")
-			Expect(err).ToNot(HaveOccurred())
-
-			go func() {
-				defer GinkgoRecover()
-
-				for {
-					buf := make([]byte, 1500)
-					n, addr, err2 := serverConn.ReadFromUDP(buf)
-					if err2 != nil {
-						return
-					}
-					data := buf[0:n]
-					serverReceivedPackets = append(serverReceivedPackets, packetData(data))
-
-					// echo each packet received back to the client
-					serverConn.WriteToUDP(data, addr)
-					serverNumPacketsSent++
-				}
-			}()
-
-			// setup the client
-			localAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
-			Expect(err).ToNot(HaveOccurred())
-			clientConn, err = net.DialUDP("udp", localAddr, proxyAddr)
-			Expect(err).ToNot(HaveOccurred())
-		})
-
-		AfterEach(func() {
-			proxy.Stop()
-			serverConn.Close()
-			time.Sleep(time.Millisecond)
-		})
-
-		Context("no packet drop", func() {
-			BeforeEach(func() {
-				// setup the proxy
-				var err error
-				proxy, err = NewUDPProxy(10001, "localhost", serverPort, nil, nil, 0, 0)
-				Expect(err).ToNot(HaveOccurred())
-			})
-
-			It("relays packets from the client to the server", func() {
-				_, err := clientConn.Write(makePacket(1, []byte("foobar")))
-				Expect(err).ToNot(HaveOccurred())
-
-				Eventually(func() map[string]*connection { return proxy.clientDict }).Should(HaveLen(1))
-				var conn *connection
-				for _, conn = range proxy.clientDict {
-					Expect(atomic.LoadUint64(&conn.incomingPacketCounter)).To(Equal(uint64(1)))
-				}
-				_, err = clientConn.Write(makePacket(2, []byte("decafbad")))
-				Expect(err).ToNot(HaveOccurred())
-				Expect(proxy.clientDict).To(HaveLen(1))
-				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(2))
-				Expect(string(serverReceivedPackets[0])).To(ContainSubstring("foobar"))
-				Expect(string(serverReceivedPackets[1])).To(ContainSubstring("decafbad"))
-			})
-
-			It("relays packets from the server to the client", func() {
-				_, err := clientConn.Write(makePacket(1, []byte("foobar")))
-				Expect(err).ToNot(HaveOccurred())
-				Eventually(func() map[string]*connection { return proxy.clientDict }).Should(HaveLen(1))
-				var key string
-				var conn *connection
-				for key, conn = range proxy.clientDict {
-					Eventually(func() uint64 { return atomic.LoadUint64(&conn.outgoingPacketCounter) }).Should(Equal(uint64(1)))
-				}
-				_, err = clientConn.Write(makePacket(2, []byte("decafbad")))
-				Expect(err).ToNot(HaveOccurred())
-
-				Expect(proxy.clientDict).To(HaveLen(1))
-				Eventually(func() uint64 { return proxy.clientDict[key].outgoingPacketCounter }).Should(Equal(uint64(2)))
-
-				var clientReceivedPackets []packetData
-
-				// receive the packets echoed by the server on client side
-				go func() {
-					defer GinkgoRecover()
-
-					for {
-						buf := make([]byte, 1500)
-						n, _, err2 := clientConn.ReadFromUDP(buf)
-						if err2 != nil {
-							return
-						}
-						data := buf[0:n]
-						clientReceivedPackets = append(clientReceivedPackets, packetData(data))
-					}
-				}()
-
-				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(2))
-				Expect(serverReceivedPackets).To(HaveLen(2))
-				Expect(serverNumPacketsSent).To(Equal(2))
-				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(2))
-				Expect(string(clientReceivedPackets[0])).To(ContainSubstring("foobar"))
-				Expect(string(clientReceivedPackets[1])).To(ContainSubstring("decafbad"))
-			})
-		})
-
-		Context("Drop Callbacks", func() {
-			It("drops incoming packets", func() {
-				dropper := func(p protocol.PacketNumber) bool {
-					return p%2 == 0
-				}
-
-				var err error
-				proxy, err = NewUDPProxy(10001, "localhost", serverPort, dropper, nil, 0, 0)
-				Expect(err).ToNot(HaveOccurred())
-
-				for i := 1; i <= 6; i++ {
-					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
-					Expect(err).ToNot(HaveOccurred())
-				}
-				Eventually(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
-				Consistently(func() []packetData { return serverReceivedPackets }).Should(HaveLen(3))
-			})
-
-			It("drops outgoing packets", func() {
-				dropper := func(p protocol.PacketNumber) bool {
-					return p%2 == 0
-				}
-
-				var err error
-				proxy, err = NewUDPProxy(10001, "localhost", serverPort, nil, dropper, 0, 0)
-				Expect(err).ToNot(HaveOccurred())
-
-				for i := 1; i <= 6; i++ {
-					_, err := clientConn.Write(makePacket(protocol.PacketNumber(i), []byte("foobar"+strconv.Itoa(i))))
-					Expect(err).ToNot(HaveOccurred())
-				}
-
-				var clientReceivedPackets []packetData
-
-				// receive the packets echoed by the server on client side
-				go func() {
-					defer GinkgoRecover()
-
-					for {
-						buf := make([]byte, 1500)
-						n, _, err2 := clientConn.ReadFromUDP(buf)
-						if err2 != nil {
-							return
-						}
-						data := buf[0:n]
-						clientReceivedPackets = append(clientReceivedPackets, packetData(data))
-					}
-				}()
-
-				Eventually(func() []packetData { return clientReceivedPackets }).Should(HaveLen(3))
-				Consistently(func() []packetData { return clientReceivedPackets }).Should(HaveLen(3))
-			})
-		})
-	})
-})

integrationtests/random_rtt_test.go

@@ -3,6 +3,7 @@ package integrationtests
 import (
 	"bytes"
 	"fmt"
+	"math/rand"
 	"os/exec"
 	"strconv"
 	"time"
@@ -16,26 +17,62 @@ import (
 	. "github.com/onsi/gomega/gexec"
 )
 
+// get a random duration between min and max
+func getRandomDuration(min, max time.Duration) time.Duration {
+	return min + time.Duration(rand.Int63n(int64(max-min)))
+}
+
+var _ = Describe("Random Duration Generator", func() {
+	rand.Seed(time.Now().UnixNano())
+	It("gets a random RTT", func() {
+		var min time.Duration = time.Hour
+		var max time.Duration
+
+		var sum time.Duration
+		rep := 10000
+		for i := 0; i < rep; i++ {
+			val := getRandomDuration(100*time.Millisecond, 500*time.Millisecond)
+			sum += val
+			if val < min {
+				min = val
+			}
+			if val > max {
+				max = val
+			}
+		}
+		avg := sum / time.Duration(rep)
+		fmt.Println(avg)
+		Expect(avg).To(BeNumerically("~", 300*time.Millisecond, 5*time.Millisecond))
+		Expect(min).To(BeNumerically(">=", 100*time.Millisecond))
+		Expect(min).To(BeNumerically("<", 105*time.Millisecond))
+		Expect(max).To(BeNumerically(">", 495*time.Millisecond))
+		Expect(max).To(BeNumerically("<=", 500*time.Millisecond))
+	})
+})
+
 var _ = Describe("Random RTT", func() {
 	BeforeEach(func() {
 		dataMan.GenerateData(dataLen)
 	})
 
-	var rttProxy *proxy.UDPProxy
+	var proxy *quicproxy.QuicProxy
 
 	runRTTTest := func(minRtt, maxRtt time.Duration, version protocol.VersionNumber) {
-		proxyPort := 12345
-
-		iPort, _ := strconv.Atoi(port)
+		rand.Seed(time.Now().UnixNano())
 		var err error
-		rttProxy, err = proxy.NewUDPProxy(proxyPort, "localhost", iPort, nil, nil, minRtt, maxRtt)
+		proxy, err = quicproxy.NewQuicProxy("localhost:", quicproxy.Opts{
+			RemoteAddr: "localhost:" + port,
+			DelayPacket: func(_ quicproxy.Direction, _ protocol.PacketNumber) time.Duration {
+				return getRandomDuration(minRtt, maxRtt)
+			},
+		})
 		Expect(err).ToNot(HaveOccurred())
 
 		command := exec.Command(
 			clientPath,
 			"--quic-version="+strconv.Itoa(int(version)),
 			"--host=127.0.0.1",
-			"--port="+strconv.Itoa(proxyPort),
+			"--port="+strconv.Itoa(proxy.LocalPort()),
 			"https://quic.clemente.io/data",
 		)
 
@@ -47,7 +84,8 @@ var _ = Describe("Random RTT", func() {
 	}
 
 	AfterEach(func() {
-		rttProxy.Stop()
+		err := proxy.Close()
+		Expect(err).ToNot(HaveOccurred())
 		time.Sleep(time.Millisecond)
 	})
 

integrationtests/rtt_test.go

@@ -21,21 +21,23 @@ var _ = Describe("non-zero RTT", func() {
 		dataMan.GenerateData(dataLen)
 	})
 
-	var rttProxy *proxy.UDPProxy
+	var proxy *quicproxy.QuicProxy
 
 	runRTTTest := func(rtt time.Duration, version protocol.VersionNumber) {
-		proxyPort := 12345
-
-		iPort, _ := strconv.Atoi(port)
 		var err error
-		rttProxy, err = proxy.NewUDPProxy(proxyPort, "localhost", iPort, nil, nil, rtt, rtt)
+		proxy, err = quicproxy.NewQuicProxy("localhost:", quicproxy.Opts{
+			RemoteAddr: "localhost:" + port,
+			DelayPacket: func(_ quicproxy.Direction, _ protocol.PacketNumber) time.Duration {
+				return rtt / 2
+			},
+		})
 		Expect(err).ToNot(HaveOccurred())
 
 		command := exec.Command(
 			clientPath,
 			"--quic-version="+strconv.Itoa(int(version)),
 			"--host=127.0.0.1",
-			"--port="+strconv.Itoa(proxyPort),
+			"--port="+strconv.Itoa(proxy.LocalPort()),
 			"https://quic.clemente.io/data",
 		)
 
@@ -47,7 +49,8 @@ var _ = Describe("non-zero RTT", func() {
 	}
 
 	AfterEach(func() {
-		rttProxy.Stop()
+		err := proxy.Close()
+		Expect(err).ToNot(HaveOccurred())
 		time.Sleep(time.Millisecond)
 	})