Fix rule expiration

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.4.0
 	go.etcd.io/bbolt v1.3.3
-	golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72
+	golang.org/x/crypto v0.0.0-20200208060501-ecb85df21340
 	golang.org/x/net v0.0.0-20191204025024-5ee1b9f4859a
 	golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5 // indirect
 )

go.sum

@@ -164,6 +164,8 @@ golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d h1:9FCpayM9Egr1baVnV1SX0H
 golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72 h1:+ELyKg6m8UBf0nPFSqD0mi7zUfwPyXo23HNjMnXPz7w=
 golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200208060501-ecb85df21340 h1:KOcEaR10tFr7gdJV2GCKw8Os5yED1u1aOqHjOAb6d2Y=
+golang.org/x/crypto v0.0.0-20200208060501-ecb85df21340/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=

pkg/router/route_group.go

@@ -76,8 +76,9 @@ type RouteGroup struct {
 	// - fwd/tps should have the same number of elements.
 	// - the corresponding element of tps should have tpID of the corresponding rule in fwd.
 	// - fwd references 'ForwardRule' rules for writes.
-	fwd []routing.Rule // forward rules (for writing)
-	rvs []routing.Rule // reverse rules (for reading)
+	fwd                  []routing.Rule // forward rules (for writing)
+	rvs                  []routing.Rule // reverse rules (for reading)
+	rvsRouteLastActivity time.Time
 
 	lastSent int64
 
@@ -93,9 +94,10 @@ type RouteGroup struct {
 	writeDeadline deadline.PipeDeadline
 
 	// used as a bool to indicate if this particular route group initiated close loop
-	closeInitiated int32
-	remoteClosed   chan struct{}
-	closed         chan struct{}
+	closeInitiated   int32
+	remoteClosedOnce sync.Once
+	remoteClosed     chan struct{}
+	closed           chan struct{}
 	// used to wait for all the `Close` packets to run through the loop and come back
 	closeDone sync.WaitGroup
 }
@@ -418,7 +420,8 @@ func (rg *RouteGroup) close(code routing.CloseCode) error {
 			close(rg.closed)
 		}
 
-		close(rg.remoteClosed)
+		rg.setRemoteClosed()
+
 		rg.readChMu.Lock()
 		close(rg.readCh)
 		rg.readChMu.Unlock()
@@ -427,6 +430,33 @@ func (rg *RouteGroup) close(code routing.CloseCode) error {
 	return nil
 }
 
+func (rg *RouteGroup) handlePacket(packet routing.Packet) error {
+	rg.mu.Lock()
+	// no need to check rule expiry, since router won't allow packet in
+	// in case it's expired, so simply update the activity
+	rg.rvsRouteLastActivity = time.Now()
+	rg.mu.Unlock()
+
+	switch packet.Type() {
+	case routing.ClosePacket:
+		return rg.handleClosePacket(routing.CloseCode(packet.Payload()[0]))
+	case routing.DataPacket:
+		return rg.handleDataPacket(packet)
+	}
+
+	return nil
+}
+
+func (rg *RouteGroup) handleDataPacket(packet routing.Packet) error {
+	select {
+	case <-rg.closed:
+		return io.ErrClosedPipe
+	case rg.readCh <- packet.Payload():
+	}
+
+	return nil
+}
+
 func (rg *RouteGroup) handleClosePacket(code routing.CloseCode) error {
 	rg.logger.Infof("Got close packet with code %d", code)
 
@@ -475,19 +505,23 @@ func (rg *RouteGroup) isCloseInitiator() bool {
 	return atomic.LoadInt32(&rg.closeInitiated) == 1
 }
 
-func (rg *RouteGroup) isRemoteClosed() bool {
-	select {
-	case <-rg.remoteClosed:
-		return true
-	default:
-	}
+func (rg *RouteGroup) setRemoteClosed() {
+	rg.remoteClosedOnce.Do(func() {
+		close(rg.remoteClosed)
+	})
+}
 
-	return false
+func (rg *RouteGroup) isRemoteClosed() bool {
+	return chanClosed(rg.remoteClosed)
 }
 
 func (rg *RouteGroup) isClosed() bool {
+	return chanClosed(rg.closed)
+}
+
+func chanClosed(ch chan struct{}) bool {
 	select {
-	case <-rg.closed:
+	case <-ch:
 		return true
 	default:
 	}

pkg/router/route_group_test.go

@@ -589,7 +589,7 @@ func safeSend(ctx context.Context, to *RouteGroup, payload []byte) (keepSending
 }
 
 func createRouteGroup(cfg *RouteGroupConfig) *RouteGroup {
-	rt := routing.NewTable(routing.DefaultConfig())
+	rt := routing.NewTable()
 
 	pk1, _ := cipher.GenerateKeyPair()
 	pk2, _ := cipher.GenerateKeyPair()