Merge pull request #557 from galkinvv/fix-loop-tcp-disconnection

Fix infinite loop on fast TCP disconnection

src/lib/net/SecureSocket.cpp

@@ -761,7 +761,7 @@ MultiplexerJobStatus SecureSocket::serviceConnect(ISocketMultiplexerJob* job,
     // If status > 0, success
     if (status > 0) {
         sendEvent(m_events->forIDataSocket().secureConnected());
-        return {true, newJob()};
+        return newJobOrStopServicing();
     }
 
     // Retry case
@@ -793,7 +793,7 @@ MultiplexerJobStatus SecureSocket::serviceAccept(ISocketMultiplexerJob* job,
     // If status > 0, success
     if (status > 0) {
         sendEvent(m_events->forClientListener().accepted());
-        return {true, newJob()};
+        return newJobOrStopServicing();
     }
 
     // Retry case

src/lib/net/TCPSocket.cpp

@@ -403,6 +403,15 @@ void TCPSocket::setJob(std::unique_ptr<ISocketMultiplexerJob>&& job)
     }
 }
 
+MultiplexerJobStatus TCPSocket::newJobOrStopServicing()
+{
+    auto new_job = newJob();
+    if (new_job)
+        return {true, std::move(new_job)};
+    else
+        return {false, {}};
+}
+
 std::unique_ptr<ISocketMultiplexerJob> TCPSocket::newJob()
 {
     // note -- must have m_mutex locked on entry
@@ -519,22 +528,14 @@ MultiplexerJobStatus TCPSocket::serviceConnecting(ISocketMultiplexerJob* job, bo
         catch (XArchNetwork& e) {
             sendConnectionFailedEvent(e.what());
             onDisconnected();
-            auto new_job = newJob();
-            if (new_job)
-                return {true, std::move(new_job)};
-            else
-                return {false, {}};
+            return newJobOrStopServicing();
         }
     }
 
     if (write) {
         sendEvent(m_events->forIDataSocket().connected());
         onConnected();
-        auto new_job = newJob();
-        if (new_job)
-            return {true, std::move(new_job)};
-        else
-            return {false, {}};
+        return newJobOrStopServicing();
     }
 
     return {true, {}};
@@ -548,7 +549,7 @@ MultiplexerJobStatus TCPSocket::serviceConnected(ISocketMultiplexerJob* job,
     if (error) {
         sendEvent(m_events->forISocket().disconnected());
         onDisconnected();
-        return {true, newJob()};
+        return newJobOrStopServicing();
     }
 
     EJobResult writeResult = kRetry;
@@ -603,7 +604,7 @@ MultiplexerJobStatus TCPSocket::serviceConnected(ISocketMultiplexerJob* job,
     if (writeResult == kBreak || readResult == kBreak) {
         return {false, {}};
     } else if (writeResult == kNew || readResult == kNew) {
-        return {true, newJob()};
+        return newJobOrStopServicing();
     } else {
         return {true, {}};
     }

src/lib/net/TCPSocket.h

@@ -76,7 +76,8 @@ protected:
 
     void removeJob();
     void setJob(std::unique_ptr<ISocketMultiplexerJob>&& job);
-    
+    MultiplexerJobStatus newJobOrStopServicing();
+
     bool                isReadable() { return m_readable; }
     bool                isWritable() { return m_writable; }