bug 225:

Fixed short reads for non-UDP input.

LCS/Stream/include/Stream/SocketStream.h

@@ -53,11 +53,12 @@ class SocketStream : public FileDescriptorBasedStream
     void    reaccept(time_t timeout = 0); // only for TCP server socket
     void    setReadBufferSize(size_t size);
 
+    const Protocol protocol;
+    const Mode mode;
+
   private:
     const char *hostname;
     uint16 port;
-    const Protocol protocol;
-    const Mode mode;
     const char *nfskey;
     int     listen_sk;
 

LCS/Stream/src/SocketStream.cc

@@ -47,10 +47,10 @@ const int MAXPORT = 30000;
 
 SocketStream::SocketStream(const char *hostname, uint16 _port, Protocol protocol, Mode mode, time_t timeout, const char *nfskey )
 :
-  hostname(hostname),
-  port(_port),
   protocol(protocol),
   mode(mode),
+  hostname(hostname),
+  port(_port),
   nfskey(nfskey),
   listen_sk(-1)
 {  

RTCP/IONProc/src/InputThread.cc

@@ -32,6 +32,7 @@
 #include <Interface/AlignedStdAllocator.h>
 #include <Interface/Exceptions.h>
 #include <Stream/NullStream.h>
+#include <Stream/SocketStream.h>
 #include <BeamletBuffer.h>
 #include <InputThread.h>
 #include <RSP.h>
@@ -91,18 +92,26 @@ template <typename SAMPLE_TYPE> void InputThread<SAMPLE_TYPE>::mainLoop()
   bool		previousSeqidIsAccepted	    = false;
   
   bool		dataShouldContainValidStamp = dynamic_cast<NullStream *>(itsArgs.stream) == 0;
+  bool		isUDPstream		    = dynamic_cast<SocketStream *>(itsArgs.stream) != 0 && dynamic_cast<SocketStream *>(itsArgs.stream)->protocol == SocketStream::UDP;
   WallClockTime wallClockTime;
 
   LOG_DEBUG_STR(itsArgs.logPrefix << " input thread " << itsArgs.threadID << " entering loop");
 
   while (!itsShouldStop) {
-    size_t size;
-
     try {
       // interruptible read, to allow stopping this thread even if the station
       // does not send data
 
-      size = itsArgs.stream->tryRead(currentPacketPtr, packetSize);
+      if (isUDPstream) {
+	if (itsArgs.stream->tryRead(currentPacketPtr, packetSize) != packetSize) {
+	  ++ itsArgs.packetCounters->received;
+	  ++ itsArgs.packetCounters->badSize;
+	  continue;
+	}
+      } else {
+	itsArgs.stream->read(currentPacketPtr, packetSize);
+      }
+
     } catch (Stream::EndOfStreamException &) {
       break;
     } catch (SystemCallException &ex) {
@@ -114,11 +123,6 @@ template <typename SAMPLE_TYPE> void InputThread<SAMPLE_TYPE>::mainLoop()
 
     ++ itsArgs.packetCounters->received;
 
-    if (size != packetSize) {
-      ++ itsArgs.packetCounters->badSize;
-      continue;
-    }
-
     if (dataShouldContainValidStamp) {
 #if defined __PPC__
       unsigned seqid, blockid;