call Router::PumpLL after every batch of packets from a link session

llarp/iwp/iwp.cpp

@@ -12,10 +12,11 @@ namespace llarp
                    LinkMessageHandler h, SignBufferFunc sign,
                    SessionEstablishedHandler est,
                    SessionRenegotiateHandler reneg, TimeoutHandler timeout,
-                   SessionClosedHandler closed)
+                   SessionClosedHandler closed, PumpDoneHandler pumpDone)
     {
       return std::make_shared< LinkLayer >(routerEncSecret, getrc, h, sign, est,
-                                           reneg, timeout, closed, true);
+                                           reneg, timeout, closed, pumpDone,
+                                           true);
     }
 
     LinkLayer_ptr
@@ -23,10 +24,11 @@ namespace llarp
                     LinkMessageHandler h, SignBufferFunc sign,
                     SessionEstablishedHandler est,
                     SessionRenegotiateHandler reneg, TimeoutHandler timeout,
-                    SessionClosedHandler closed)
+                    SessionClosedHandler closed, PumpDoneHandler pumpDone)
     {
       return std::make_shared< LinkLayer >(routerEncSecret, getrc, h, sign, est,
-                                           reneg, timeout, closed, false);
+                                           reneg, timeout, closed, pumpDone,
+                                           false);
     }
   }  // namespace iwp
 }  // namespace llarp

llarp/iwp/iwp.hpp

@@ -14,13 +14,13 @@ namespace llarp
                    LinkMessageHandler h, SignBufferFunc sign,
                    SessionEstablishedHandler est,
                    SessionRenegotiateHandler reneg, TimeoutHandler timeout,
-                   SessionClosedHandler closed);
+                   SessionClosedHandler closed, PumpDoneHandler pumpDone);
     LinkLayer_ptr
     NewOutboundLink(const SecretKey& routerEncSecret, GetRCFunc getrc,
                     LinkMessageHandler h, SignBufferFunc sign,
                     SessionEstablishedHandler est,
                     SessionRenegotiateHandler reneg, TimeoutHandler timeout,
-                    SessionClosedHandler closed);
+                    SessionClosedHandler closed, PumpDoneHandler pumpDone);
 
   }  // namespace iwp
 }  // namespace llarp

llarp/iwp/linklayer.cpp

@@ -11,9 +11,9 @@ namespace llarp
                          SessionEstablishedHandler est,
                          SessionRenegotiateHandler reneg,
                          TimeoutHandler timeout, SessionClosedHandler closed,
-                         bool allowInbound)
+                         PumpDoneHandler pumpDone, bool allowInbound)
         : ILinkLayer(routerEncSecret, getrc, h, sign, est, reneg, timeout,
-                     closed)
+                     closed, pumpDone)
         , permitInbound{allowInbound}
     {
     }

llarp/iwp/linklayer.hpp

@@ -18,7 +18,7 @@ namespace llarp
                 LinkMessageHandler h, SignBufferFunc sign,
                 SessionEstablishedHandler est, SessionRenegotiateHandler reneg,
                 TimeoutHandler timeout, SessionClosedHandler closed,
-                bool permitInbound);
+                PumpDoneHandler pumpDone, bool permitInbound);
 
       ~LinkLayer() override;
 

llarp/iwp/session.cpp

@@ -596,14 +596,7 @@ namespace llarp
         }
       }
       SendMACK();
-      if(m_EncryptNext && !m_EncryptNext->empty())
-      {
-        m_Parent->QueueWork(
-            [self = shared_from_this(), data = std::move(m_EncryptNext)] {
-              self->EncryptWorker(data);
-            });
-        m_EncryptNext = nullptr;
-      }
+      m_Parent->PumpDone();
     }
 
     void

llarp/link/factory.hpp

@@ -22,7 +22,7 @@ namespace llarp
     using Factory = std::function< LinkLayer_ptr(
         const SecretKey&, GetRCFunc, LinkMessageHandler, SignBufferFunc,
         SessionEstablishedHandler, SessionRenegotiateHandler, TimeoutHandler,
-        SessionClosedHandler) >;
+        SessionClosedHandler, PumpDoneHandler) >;
 
     /// get link type by name string
     /// if invalid returns eLinkUnspec

llarp/link/server.hpp

@@ -45,13 +45,17 @@ namespace llarp
   /// handles close of all sessions with pubkey
   using SessionClosedHandler = std::function< void(llarp::RouterID) >;
 
+  /// notifies router that a link session has ended its pump and we should flush
+  /// messages to upper layers
+  using PumpDoneHandler = std::function< void(void) >;
+
   struct ILinkLayer
   {
     ILinkLayer(const SecretKey& routerEncSecret, GetRCFunc getrc,
                LinkMessageHandler handler, SignBufferFunc signFunc,
                SessionEstablishedHandler sessionEstablish,
                SessionRenegotiateHandler renegotiate, TimeoutHandler timeout,
-               SessionClosedHandler closed);
+               SessionClosedHandler closed, PumpDoneHandler pumpDone);
     virtual ~ILinkLayer();
 
     /// get current time via event loop
@@ -182,6 +186,7 @@ namespace llarp
     SessionEstablishedHandler SessionEstablished;
     SessionClosedHandler SessionClosed;
     SessionRenegotiateHandler SessionRenegotiate;
+    PumpDoneHandler PumpDone;
 
     std::shared_ptr< Logic >
     logic()

llarp/router/router.cpp

@@ -539,7 +539,8 @@ namespace llarp
           util::memFn(&AbstractRouter::CheckRenegotiateValid, this),
           util::memFn(&IOutboundSessionMaker::OnConnectTimeout,
                       &_outboundSessionMaker),
-          util::memFn(&AbstractRouter::SessionClosed, this));
+          util::memFn(&AbstractRouter::SessionClosed, this),
+          util::memFn(&AbstractRouter::PumpLL, this));
 
       if(!server->EnsureKeys(transport_keyfile.string().c_str()))
       {

test/link/test_llarp_link.cpp

@@ -227,7 +227,9 @@ TEST_F(LinkLayerTest, TestIWP)
         ASSERT_FALSE(session->IsEstablished());
         Stop();
       },
-      [&](RouterID router) { ASSERT_EQ(router, Bob.GetRouterID()); });
+      [&](RouterID router) { ASSERT_EQ(router, Alice.GetRouterID()); },
+      []() {})
+      ;
 
   auto sendDiscardMessage = [](ILinkSession* s) -> bool {
     // send discard message in reply to complete unit test
@@ -267,7 +269,9 @@ TEST_F(LinkLayerTest, TestIWP)
         return true;
       },
       [&](ILinkSession* session) { ASSERT_FALSE(session->IsEstablished()); },
-      [&](RouterID router) { ASSERT_EQ(router, Alice.GetRouterID()); });
+      [&](RouterID router) { ASSERT_EQ(router, Alice.GetRouterID()); },
+      []() {})
+      ;
 
   ASSERT_TRUE(Alice.Start(m_logic, netLoop, AlicePort));
   ASSERT_TRUE(Bob.Start(m_logic, netLoop, BobPort));