revert to closing streams immediately but fix last packets sent back

This patch reverts the change done on v2.4.1, so that streams are closed immediately as it used to be (instead of relying on CheckStreamsTimeout) but fixes last packets sent back to sender, by doing basically what was already done before but on proper order to avoid the issue.

Relying on CheckStreamsTimeout actually introduces a problem: transmitting client gets "mute" for 1.6 sec. after each transmission, this means that if a 2nd client starts transmitting within this time frame then the "mute" client can't listen the beginning of the 2nd client transmission, the problem is even worst (i.e. critical) for DExtra because the "mute" client will not be able to get the header packet of the 2nd client transmission and this means that whole transmission is lost - this renders DExtra almost unusable on v2.4.1 unless all clients take special care about spaces between transmissions.

src/cprotocol.cpp

@@ -139,11 +139,11 @@ void CProtocol::OnDvFramePacketIn(CDvFramePacket *Frame, const CIp *Ip)
     // find the stream
     CPacketStream *stream = GetStream(Frame->GetStreamId(), Ip);
     if ( stream == NULL )
-	{
+    {
-		// std::cout << "Deleting oprhaned Last Frame Packet with StreamId " << Frame->GetStreamId() << " from " << *Ip << std::endl;
+        // std::cout << "Deleting oprhaned Frame Packet with StreamId " << Frame->GetStreamId() << " from " << *Ip << std::endl;
-		delete Frame;
+        delete Frame;
-	}
+    }
-	else
+    else
     {
         //std::cout << "DV frame" << "from "  << *Ip << std::endl;
         // and push
@@ -158,22 +158,40 @@ void CProtocol::OnDvLastFramePacketIn(CDvLastFramePacket *Frame, const CIp *Ip)
     // find the stream
     CPacketStream *stream = GetStream(Frame->GetStreamId(), Ip);
     if ( stream == NULL )
- 	{
+    {
-		// std::cout << "Deleting oprhaned Last Frame Packet with StreamId " << Frame->GetStreamId() << " from " << *Ip << std::endl;
+        // std::cout << "Deleting oprhaned Last Frame Packet with StreamId " << Frame->GetStreamId() << " from " << *Ip << std::endl;
-		delete Frame;
+        delete Frame;
-	}
+    }
-	else
+    else
     {
         // push
         stream->Lock();
         stream->Push(Frame);
         stream->Unlock();
+
+        // wait stream queue is empty, same as done in CloseStream(), but we need it before HandleQueue()
+        bool bEmpty = false;
+        do
+        {
+            stream->Lock();
+            // do not use stream->IsEmpty() has this "may" never succeed
+            // and anyway, the DvLastFramPacket short-circuit the transcoder
+            // loop queues
+            bEmpty = stream->empty();
+            stream->Unlock();
+            if ( !bEmpty )
+            {
+                // wait a bit
+                CTimePoint::TaskSleepFor(10);
+            }
+        } while (!bEmpty);
+
+        // handle queue from reflector a bit earlier, before closing the stream,
+        // this avoid last packets to be sent back to transmitting client (master)
+        HandleQueue();
         
-        // and don't close the stream yet but rely on CheckStreamsTimeout
+        // and close the stream
-        // mechanism, so the stream will be closed after the queues have
+        g_Reflector.CloseStream(stream);
-        // been sinked out. This avoid last packets to be send back
-        // to transmitting client (master)
-        // g_Reflector.CloseStream(stream);
     }
 }