Remove disconnected sockets

2 files changed, 21 insertions, 0 deletions

diff --git a/include/sibs/DirectConnection.hpp b/include/sibs/DirectConnection.hpp
index 10359f9..9786c64 100644
--- a/include/sibs/DirectConnection.hpp
+++ b/include/sibs/DirectConnection.hpp
@@ -56,6 +56,7 @@ namespace sibs
         int createSocket(const Ipv4 &addressToBind, bool rendezvous, bool reuseAddr, bool bind = true);
     private:
         void connect(const Ipv4 &address, bool rendezvous, bool reuseAddr, PubSubConnectCallback connectCallbackFunc, PubSubReceiveDataCallback receiveDataCallbackFunc);
+        void removeDisconnectedPeers();
         void receiveData();
         int receiveDataFromPeer(const int socket, char *output, usize *receivedTotalSize);
         bool removePeer(int peerSocket);
diff --git a/src/DirectConnection.cpp b/src/DirectConnection.cpp
index ce63eaf..e6b5645 100644
--- a/src/DirectConnection.cpp
+++ b/src/DirectConnection.cpp
@@ -158,6 +158,25 @@ namespace sibs
         return wasRemoved;
     }
     
+    void DirectConnections::removeDisconnectedPeers()
+    {
+        peersMutex.lock();
+        for(std::unordered_map<int, std::shared_ptr<DirectConnectionPeer>>::iterator it = peers.begin(); it != peers.end(); )
+        {
+            UDTSTATUS peerSocketStatus = UDT::getsockstate(it->first);
+            if(peerSocketStatus == UDTSTATUS::BROKEN || peerSocketStatus == UDTSTATUS::CLOSING || peerSocketStatus == UDTSTATUS::CLOSED || peerSocketStatus == UDTSTATUS::NONEXIST)
+            {
+                int socket = it->first;
+                Log::debug("UDT: Connection was broken to socket %d (peer most likely disconnected), removing peer", socket);
+                it = peers.erase(it);
+                Log::debug("UDT: Removed peer socket %d", socket);
+            }
+            else
+                ++it;
+        }
+        peersMutex.unlock();
+    }
+    
     void DirectConnections::receiveData()
     {
         std::vector<char> data;
@@ -167,6 +186,7 @@ namespace sibs
         std::set<UDTSOCKET> readfds;
         while(alive)
         {
+            removeDisconnectedPeers();
             int numfsReady = UDT::epoll_wait(eid, &readfds, nullptr, 1000);
             if(numfsReady == 0)
             {