wip

node/README.md

@@ -2,11 +2,10 @@
 
 ## Refactoring for Testability
 
-* [ ] connWriter
-  * [ ] Separate send/relay calls
+* [x] connWriter
 * [x] mcWriter
 * [x] ifWriter
-* [ ] ifReader
+* [ ] ifReader (testing)
 * [ ] connReader
 * [ ] mcReader
 * [ ] hubPoller

node/connwriter.go

@@ -68,18 +68,18 @@ func newConnWriter(conn udpAddrPortWriter, localIP byte) *connWriter {
 
 // Not safe for concurrent use. Should only be called by supervisor.
 func (w *connWriter) SendControlPacket(pkt marshaller, route *peerRoute) {
-	buf := pkt.Marshal(w.cBuf1)
-	h := header{
-		StreamID: controlStreamID,
-		Counter:  atomic.AddUint64(&w.counters[route.IP], 1),
-		SourceIP: w.localIP,
-		DestIP:   route.IP,
-	}
-	buf = route.ControlCipher.Encrypt(h, buf, w.cBuf2)
+	buf := w.encryptControlPacket(pkt, route)
 	w.writeTo(buf, route.RemoteAddr)
 }
 
+// Relay control packet. Routes must not be nil.
 func (w *connWriter) RelayControlPacket(pkt marshaller, route, relay *peerRoute) {
+	buf := w.encryptControlPacket(pkt, route)
+	w.relayPacket(buf, w.cBuf1, route, relay)
+}
+
+// Encrypted packet will occupy cBuf2.
+func (w *connWriter) encryptControlPacket(pkt marshaller, route *peerRoute) []byte {
 	buf := pkt.Marshal(w.cBuf1)
 	h := header{
 		StreamID: controlStreamID,
@@ -87,12 +87,11 @@ func (w *connWriter) RelayControlPacket(pkt marshaller, route, relay *peerRoute)
 		SourceIP: w.localIP,
 		DestIP:   route.IP,
 	}
-	buf = route.ControlCipher.Encrypt(h, buf, w.cBuf2)
-	w.relayPacket(buf, w.cBuf1, route, relay)
+	return route.ControlCipher.Encrypt(h, buf, w.cBuf2)
 }
 
 // Not safe for concurrent use. Should only be called by ifReader.
-func (w *connWriter) SendDataPacket(pkt []byte, route, relay *peerRoute) {
+func (w *connWriter) SendDataPacket(pkt []byte, route *peerRoute) {
 	h := header{
 		StreamID: dataStreamID,
 		Counter:  atomic.AddUint64(&w.counters[route.IP], 1),
@@ -101,16 +100,21 @@ func (w *connWriter) SendDataPacket(pkt []byte, route, relay *peerRoute) {
 	}
 
 	enc := route.DataCipher.Encrypt(h, pkt, w.dBuf1)
-
-	if route.Direct {
-		w.writeTo(enc, route.RemoteAddr)
-		return
-	}
-
-	w.relayPacket(enc, w.dBuf2, route, relay)
+	w.writeTo(enc, route.RemoteAddr)
 }
 
-// TODO: RelayDataPacket
+// Relay a data packet. Routes must not be nil.
+func (w *connWriter) RelayDataPacket(pkt []byte, route, relay *peerRoute) {
+	h := header{
+		StreamID: dataStreamID,
+		Counter:  atomic.AddUint64(&w.counters[route.IP], 1),
+		SourceIP: w.localIP,
+		DestIP:   route.IP,
+	}
+
+	enc := route.DataCipher.Encrypt(h, pkt, w.dBuf1)
+	w.relayPacket(enc, w.dBuf2, route, relay)
+}
 
 // Safe for concurrent use. Should only be called by connReader.
 //
@@ -122,10 +126,6 @@ func (w *connWriter) SendEncryptedDataPacket(pkt []byte, route *peerRoute) {
 }
 
 func (w *connWriter) relayPacket(data, buf []byte, route, relay *peerRoute) {
-	if relay == nil || !relay.Up {
-		return
-	}
-
 	h := header{
 		StreamID: dataStreamID,
 		Counter:  atomic.AddUint64(&w.counters[relay.IP], 1),

node/connwriter_test.go

@@ -126,7 +126,7 @@ func TestConnWriter_SendControlPacket_direct(t *testing.T) {
 }
 
 // Testing if we can relay a packet via an intermediary.
-func TestConnWriter_SendControlPacket_relay(t *testing.T) {
+func TestConnWriter_RelayControlPacket_relay(t *testing.T) {
 	route, rRoute, relay, rRelay := testConnWriter_getTestRoutes()
 
 	writer := &testUDPAddrPortWriter{}
@@ -159,40 +159,6 @@ func TestConnWriter_SendControlPacket_relay(t *testing.T) {
 	}
 }
 
-// Testing that a nil relay doesn't cause an issue.
-func TestConnWriter_SendControlPacket_relay_relayNil(t *testing.T) {
-	route, rRoute, _, _ := testConnWriter_getTestRoutes()
-
-	writer := &testUDPAddrPortWriter{}
-	w := newConnWriter(writer, rRoute.IP)
-	in := testPacket("hello world!")
-
-	w.RelayControlPacket(in, route, nil)
-
-	out := writer.Written()
-	if len(out) != 0 {
-		t.Fatal(out)
-	}
-
-}
-
-// Testing that we don't send anything if the relay isn't up.
-func TestConnWriter_SendControlPacket_relay_relayNotUp(t *testing.T) {
-	route, rRoute, relay, _ := testConnWriter_getTestRoutes()
-	relay.Up = false
-
-	writer := &testUDPAddrPortWriter{}
-	w := newConnWriter(writer, rRoute.IP)
-	in := testPacket("hello world!")
-
-	w.RelayControlPacket(in, route, relay)
-
-	out := writer.Written()
-	if len(out) != 0 {
-		t.Fatal(out)
-	}
-}
-
 // Testing that we can send a data packet directly to a remote route.
 func TestConnWriter_SendDataPacket_direct(t *testing.T) {
 	route, rRoute, _, _ := testConnWriter_getTestRoutes()
@@ -202,7 +168,7 @@ func TestConnWriter_SendDataPacket_direct(t *testing.T) {
 	w := newConnWriter(writer, rRoute.IP)
 
 	in := []byte("hello world!")
-	w.SendDataPacket(in, route, nil)
+	w.SendDataPacket(in, route)
 
 	out := writer.Written()
 	if len(out) != 1 {
@@ -224,14 +190,14 @@ func TestConnWriter_SendDataPacket_direct(t *testing.T) {
 }
 
 // Testing that we can relay a data packet via a relay.
-func TestConnWriter_SendDataPacket_relay(t *testing.T) {
+func TestConnWriter_RelayDataPacket_relay(t *testing.T) {
 	route, rRoute, relay, rRelay := testConnWriter_getTestRoutes()
 
 	writer := &testUDPAddrPortWriter{}
 	w := newConnWriter(writer, rRoute.IP)
 	in := []byte("Hello world!")
 
-	w.SendDataPacket(in, route, relay)
+	w.RelayDataPacket(in, route, relay)
 
 	out := writer.Written()
 	if len(out) != 1 {
@@ -257,35 +223,26 @@ func TestConnWriter_SendDataPacket_relay(t *testing.T) {
 	}
 }
 
-// Testing that we don't attempt to relay if the relay is nil.
-func TestConnWriter_SendDataPacket_relay_relayNil(t *testing.T) {
+// Testing that we can send an already encrypted packet.
+func TestConnWriter_SendEncryptedDataPacket(t *testing.T) {
 	route, rRoute, _, _ := testConnWriter_getTestRoutes()
 
 	writer := &testUDPAddrPortWriter{}
 	w := newConnWriter(writer, rRoute.IP)
 	in := []byte("Hello world!")
 
-	w.SendDataPacket(in, route, nil)
+	w.SendEncryptedDataPacket(in, route)
 
 	out := writer.Written()
-	if len(out) != 0 {
+	if len(out) != 1 {
 		t.Fatal(out)
 	}
-}
 
-// Testing that we don't attempt to relay if the relay isn't up.
-func TestConnWriter_SendDataPacket_relay_relayNotUp(t *testing.T) {
-	route, rRoute, relay, _ := testConnWriter_getTestRoutes()
-	relay.Up = false
+	if out[0].Addr != route.RemoteAddr {
+		t.Fatal(out[0])
+	}
 
-	writer := &testUDPAddrPortWriter{}
-	w := newConnWriter(writer, rRoute.IP)
-	in := []byte("Hello world!")
-
-	w.SendDataPacket(in, route, relay)
-
-	out := writer.Written()
-	if len(out) != 0 {
-		t.Fatal(out)
+	if !bytes.Equal(out[0].Data, in) {
+		t.Fatal(out[0])
 	}
 }

node/dupcheck.go

@@ -38,14 +38,14 @@ func (dc *dupCheck) IsDup(counter uint64) bool {
 	delta := counter - dc.tailCounter
 
 	// Full clear.
-	if delta >= bitSetSize {
+	if delta >= bitSetSize-1 {
 		dc.ClearAll()
 		dc.Set(0)
 
 		dc.tail = 1
 		dc.head = 2
 		dc.tailCounter = counter + 1
-		dc.headCounter = dc.tailCounter - bitSetSize
+		dc.headCounter = dc.tailCounter - bitSetSize + 1
 
 		return false
 	}

node/header.go

@@ -20,6 +20,18 @@ type header struct {
 	Counter  uint64 // Init with time.Now().Unix << 30 to ensure monotonic.
 }
 
+func parseHeader(b []byte) (h header, ok bool) {
+	if len(b) < headerSize {
+		return
+	}
+	h.Version = b[0]
+	h.StreamID = b[1]
+	h.SourceIP = b[2]
+	h.DestIP = b[3]
+	h.Counter = *(*uint64)(unsafe.Pointer(&b[4]))
+	return h, true
+}
+
 func (h *header) Parse(b []byte) {
 	h.Version = b[0]
 	h.StreamID = b[1]

node/ifreader.go

@@ -57,7 +57,7 @@ func (r *ifReader) sendPacket(pkt []byte, remoteIP byte) {
 		return
 	}
 
-	if relay := r.relay.Load(); relay != nil {
+	if relay := r.relay.Load(); relay != nil && relay.Up {
 		r.relayDataPacket(pkt, route, relay)
 	}
 }