# HG changeset patch
# User Colin Caughie <ccaughie@openeye.net>
# Date 1364858232 25200
# Node ID bc96a22825a0e57979826bd48849e3f4e144779d
# Parent  364dea5ec68ef972b6adafa4c1e4986d12be38fd
Read all available data on TCP socket before returning to task scheduler

This vastly improves efficiency when running several clients in one thread.

diff --git a/live/liveMedia/RTPInterface.cpp b/live/liveMedia/RTPInterface.cpp
--- a/live/liveMedia/RTPInterface.cpp
+++ b/live/liveMedia/RTPInterface.cpp
@@ -62,7 +62,7 @@
 
 private:
   static void tcpReadHandler(SocketDescriptor*, int mask);
-  void tcpReadHandler1(int mask);
+  bool tcpReadHandler1(int mask);
 
 private:
   UsageEnvironment& fEnv;
@@ -393,10 +393,13 @@
 }
 
 void SocketDescriptor::tcpReadHandler(SocketDescriptor* socketDescriptor, int mask) {
-  socketDescriptor->tcpReadHandler1(mask);
+  // Call the read handler until it returns false, with a limit to avoid
+  // starving other sockets
+  int count = 2000;
+  while (socketDescriptor->tcpReadHandler1(mask) && --count);
 }
 
-void SocketDescriptor::tcpReadHandler1(int mask) {
+bool SocketDescriptor::tcpReadHandler1(int mask) {
   // We expect the following data over the TCP channel:
   //   optional RTSP command or response bytes (before the first '$' character)
   //   a '$' character
@@ -409,16 +412,20 @@
   struct sockaddr_in fromAddress;
   if (fTCPReadingState != AWAITING_PACKET_DATA) {
     int result = readSocket(fEnv, fOurSocketNum, &c, 1, fromAddress);
-    if (result != 1) { // error reading TCP socket, so we will no longer handle it
+    if (result == 0) {
+      // No more data to read
+      return false;
+    } else if (result != 1) { // error reading TCP socket, so we will no longer handle it
 #ifdef DEBUG_RECEIVE
       fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): readSocket(1 byte) returned %d (error)\n", fOurSocketNum, result);
 #endif
       fReadErrorOccurred = True;
       delete this;
-      return;
+      return false;
     }
   }
-  
+
+  bool callAgain = true;
   switch (fTCPReadingState) {
     case AWAITING_DOLLAR: {
       if (c == '$') {
@@ -470,6 +477,7 @@
       break;
     }
     case AWAITING_PACKET_DATA: {
+      callAgain = false;
       fTCPReadingState = AWAITING_DOLLAR; // the next state, unless we end up having to read more data in the current state
       // Call the appropriate read handler to get the packet data from the TCP stream:
       RTPInterface* rtpInterface = lookupRTPInterface(fStreamChannelId);
@@ -489,24 +497,29 @@
 	  fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): No handler proc for \"rtpInterface\" for channel %d; need to skip %d remaining bytes\n", fOurSocketNum, fStreamChannelId, rtpInterface->fNextTCPReadSize);
 #endif
 	  int result = readSocket(fEnv, fOurSocketNum, &c, 1, fromAddress);
-	  if (result != 1) { // error reading TCP socket, so we will no longer handle it
+	  if (result < 0) { // error reading TCP socket, so we will no longer handle it
 #ifdef DEBUG_RECEIVE
 	    fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): readSocket(1 byte) returned %d (error)\n", fOurSocketNum, result);
 #endif
 	    fReadErrorOccurred = True;
 	    delete this;
-	    return;
+	    return false;
+	  } else {
+	    fTCPReadingState = AWAITING_PACKET_DATA;
+	    if (result == 1) {
+	      --rtpInterface->fNextTCPReadSize;
+	      callAgain = true;
+	    }
 	  }
-	  --rtpInterface->fNextTCPReadSize;
-	  fTCPReadingState = AWAITING_PACKET_DATA;
 	}
       }
 #ifdef DEBUG_RECEIVE
       else fprintf(stderr, "SocketDescriptor(socket %d)::tcpReadHandler(): No \"rtpInterface\" for channel %d\n", fOurSocketNum, fStreamChannelId);
 #endif
-      return;
     }
   }
+
+  return callAgain;
 }