121 lines
2.2 KiB
Go
121 lines
2.2 KiB
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
|
|
copy(ip[:], address.IP)
|
|
|
|
return ComparableUdpAddress{
|
|
IP: ip,
|
|
Port: address.Port,
|
|
Zone: address.Zone,
|
|
}
|
|
}
|
|
|
|
func NewListener(
|
|
ctx context.Context,
|
|
p *proxy.Proxy,
|
|
local string,
|
|
vs []func() proxy.MacVerifier,
|
|
gs []func() proxy.MacGenerator,
|
|
c func() Congestion,
|
|
enableConsumers bool,
|
|
enableProducers bool,
|
|
) error {
|
|
laddr, err := net.ResolveUDPAddr("udp", local)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
pconn, err := net.ListenUDP("udp", laddr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
receivedConnections := make(map[ComparableUdpAddress]*InboundFlow)
|
|
|
|
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 fi, exists := receivedConnections[raddr]; exists {
|
|
log.Println("existing flow. queuing...")
|
|
if err := fi.queueDatagram(ctx, buf[:n]); err != nil {
|
|
log.Println("error")
|
|
continue
|
|
}
|
|
log.Println("queued")
|
|
continue
|
|
}
|
|
|
|
var verifiers = make([]proxy.MacVerifier, len(vs))
|
|
for i, v := range vs {
|
|
verifiers[i] = v()
|
|
}
|
|
var generators = make([]proxy.MacGenerator, len(gs))
|
|
for i, g := range gs {
|
|
generators[i] = g()
|
|
}
|
|
|
|
f := newFlow(c(), verifiers, generators)
|
|
|
|
f.writer = pconn
|
|
f.raddr = addr
|
|
f.isAlive = true
|
|
|
|
fi, err := newInboundFlow(f)
|
|
if err != nil {
|
|
log.Println(err)
|
|
continue
|
|
}
|
|
|
|
log.Printf("received new udp connection: %v\n", f)
|
|
|
|
go fi.processPackets(ctx)
|
|
go fi.earlyUpdateLoop(ctx, 0)
|
|
|
|
receivedConnections[raddr] = fi
|
|
|
|
if enableConsumers {
|
|
p.AddConsumer(ctx, fi)
|
|
}
|
|
if enableProducers {
|
|
p.AddProducer(ctx, fi)
|
|
}
|
|
|
|
log.Println("handling...")
|
|
if err := fi.queueDatagram(ctx, buf[:n]); err != nil {
|
|
return
|
|
}
|
|
log.Println("handled")
|
|
}
|
|
}()
|
|
|
|
return nil
|
|
}
|