WIP

peer/remote.go

@@ -6,7 +6,14 @@ import (
 	"net/netip"
 	"strings"
 	"sync/atomic"
+<<<<<<< HEAD
 	"vppn/m"
+=======
+	"time"
+	"vppn/m"
+
+	"git.crumpington.com/lib/go/ratelimiter"
+>>>>>>> 69f2536 (WIP)
 )
 
 // ----------------------------------------------------------------------------
@@ -37,6 +44,10 @@ type Remote struct {
 	Globals
 	RemotePeerIP byte // Immutable.
 
+<<<<<<< HEAD
+=======
+	limiter     *ratelimiter.Limiter
+>>>>>>> 69f2536 (WIP)
 	dupCheck    *dupCheck
 	sendCounter uint64 // init to startupCount << 48. Atomic access only.
 
@@ -49,9 +60,19 @@ func newRemote(g Globals, remotePeerIP byte) *Remote {
 	r := &Remote{
 		Globals:      g,
 		RemotePeerIP: remotePeerIP,
+<<<<<<< HEAD
 		dupCheck:     newDupCheck(0),
 		sendCounter:  (uint64(g.StartupCount) << 48) + 1,
 		messages:     make(chan any, 8),
+=======
+		limiter: ratelimiter.New(ratelimiter.Config{
+			FillPeriod:   20 * time.Millisecond,
+			MaxWaitCount: 1,
+		}),
+		dupCheck:    newDupCheck(0),
+		sendCounter: uint64(time.Now().Unix()<<30) + 1,
+		messages:    make(chan any, 8),
+>>>>>>> 69f2536 (WIP)
 	}
 	r.config.Store(&remoteConfig{})
 	return r
@@ -79,19 +100,36 @@ func (r *Remote) updateConf(conf remoteConfig) {
 // ----------------------------------------------------------------------------
 
 func (r *Remote) sendUDP(b []byte, addr netip.AddrPort) {
+<<<<<<< HEAD
 	if _, err := r.SendUDP(b, addr); err != nil {
 		r.logf("Failed to send UDP packet: %v", err)
+=======
+	if err := r.limiter.Limit(); err != nil {
+		r.logf("Rate limiter")
+		return
+	}
+	if _, err := r.SendUDP(b, addr); err != nil {
+		r.logf("Failed to send URP packet: %v", err)
+>>>>>>> 69f2536 (WIP)
 	}
 }
 
 // ----------------------------------------------------------------------------
 
+<<<<<<< HEAD
 func (r *Remote) encryptData(conf remoteConfig, destIP byte, packet []byte) []byte {
+=======
+func (r *Remote) encryptData(conf remoteConfig, packet []byte) []byte {
+>>>>>>> 69f2536 (WIP)
 	h := Header{
 		StreamID: dataStreamID,
 		Counter:  atomic.AddUint64(&r.sendCounter, 1),
 		SourceIP: r.Globals.LocalPeerIP,
+<<<<<<< HEAD
 		DestIP:   destIP,
+=======
+		DestIP:   r.RemotePeerIP,
+>>>>>>> 69f2536 (WIP)
 	}
 	return conf.DataCipher.Encrypt(h, packet, packet[len(packet):cap(packet)])
 }
@@ -116,6 +154,7 @@ func (r *Remote) SendDataTo(data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	// Direct:
 
 	if conf.Direct {
@@ -124,6 +163,17 @@ func (r *Remote) SendDataTo(data []byte) {
 	}
 
 	// Relayed:
+=======
+	if conf.Direct {
+		r.sendDataDirect(conf, data)
+	} else {
+		r.sendDataRelayed(conf, data)
+	}
+}
+
+// sendDataRelayed sends data to the remote via the relay.
+func (r *Remote) sendDataRelayed(conf remoteConfig, data []byte) {
+>>>>>>> 69f2536 (WIP)
 	relay := r.RelayHandler.Load()
 
 	if relay == nil {
@@ -131,15 +181,29 @@ func (r *Remote) SendDataTo(data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	relay.relayData(conf.Peer.PeerIP, r.encryptData(conf, conf.Peer.PeerIP, data))
 }
 
 func (r *Remote) relayData(toIP byte, enc []byte) {
+=======
+	relay.relayData(r.encryptData(conf, data))
+}
+
+// sendDataDirect sends data to the remote directly.
+func (r *Remote) sendDataDirect(conf remoteConfig, data []byte) {
+	r.logf("Sending data direct...")
+	r.sendUDP(r.encryptData(conf, data), conf.DirectAddr)
+}
+
+func (r *Remote) relayData(enc []byte) {
+>>>>>>> 69f2536 (WIP)
 	conf := r.conf()
 	if !conf.Up || !conf.Direct {
 		r.logf("Cannot relay: not up or not a direct connection")
 		return
 	}
+<<<<<<< HEAD
 	r.sendUDP(r.encryptData(conf, toIP, enc), conf.DirectAddr)
 }
 
@@ -154,6 +218,32 @@ func (r *Remote) sendControl(conf remoteConfig, data []byte) {
 
 	// Relayed:
 
+=======
+	r.sendDataDirect(conf, enc)
+}
+
+func (r *Remote) sendControl(conf remoteConfig, data []byte) {
+	if conf.Direct {
+		r.sendControlDirect(conf, data)
+	} else {
+		r.sendControlRelayed(conf, data)
+	}
+}
+
+func (r *Remote) sendControlToAddr(buf []byte, addr netip.AddrPort) {
+	enc := r.encryptControl(r.conf(), buf)
+	r.sendUDP(enc, addr)
+}
+
+func (r *Remote) sendControlDirect(conf remoteConfig, data []byte) {
+	r.logf("Sending control direct...")
+	enc := r.encryptControl(conf, data)
+	r.sendUDP(enc, conf.DirectAddr)
+}
+
+func (r *Remote) sendControlRelayed(conf remoteConfig, data []byte) {
+	r.logf("Sending control relayed...")
+>>>>>>> 69f2536 (WIP)
 	relay := r.RelayHandler.Load()
 
 	if relay == nil {
@@ -161,12 +251,16 @@ func (r *Remote) sendControl(conf remoteConfig, data []byte) {
 		return
 	}
 
+<<<<<<< HEAD
 	relay.relayData(conf.Peer.PeerIP, r.encryptControl(conf, data))
 }
 
 func (r *Remote) sendControlToAddr(buf []byte, addr netip.AddrPort) {
 	enc := r.encryptControl(r.conf(), buf)
 	r.sendUDP(enc, addr)
+=======
+	relay.relayData(r.encryptControl(conf, data))
+>>>>>>> 69f2536 (WIP)
 }
 
 func (r *Remote) forwardPacket(data []byte) {
@@ -244,9 +338,13 @@ func (r *Remote) handleDataPacket(h Header, data []byte) {
 	// For local.
 	if h.DestIP == r.LocalPeerIP {
 		if _, err := r.IFace.Write(dec); err != nil {
+<<<<<<< HEAD
 			// This could be a malformed packet from a peer, so we don't crash if it
 			// happens.
 			r.logf("Failed to write to interface: %v", err)
+=======
+			log.Fatalf("Failed to write to interface: %v", err)
+>>>>>>> 69f2536 (WIP)
 		}
 		return
 	}
@@ -275,6 +373,10 @@ func (r *Remote) HandleLocalDiscoveryPacket(h Header, srcAddr netip.AddrPort, da
 		SrcIP:   h.SourceIP,
 		SrcAddr: srcAddr,
 	}
+<<<<<<< HEAD
+=======
+	r.logf("Got local discovery packet from %v.", srcAddr)
+>>>>>>> 69f2536 (WIP)
 
 	select {
 	case r.messages <- msg: