Index: src/xplcommon/BlockingStream.cc
===================================================================
--- src/xplcommon/BlockingStream.cc	(revision 30)
+++ src/xplcommon/BlockingStream.cc	(revision 31)
@@ -46,8 +46,6 @@
 //------------------------------------------------------------------------------
 
-bool BlockingStream::isInterrupted()
+inline bool BlockingStream::checkInterrupted()
 {
-    if (failed()) return false;
-
     if (!interrupted) {
         interrupted = event.check();
@@ -83,4 +81,22 @@
 //------------------------------------------------------------------------------
 
+bool BlockingStream::skip(size_t length)
+{
+    ReadingBuffer& readingBuffer = stream.getReadingBuffer();
+    while (length>0) {
+        size_t toSkip = min(length, readingBuffer.getLength() - readingOffset);
+        readingOffset += toSkip;
+        length -= toSkip;
+
+        if (length!=0) {
+            if (!fill()) return false;
+        }
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
 bool BlockingStream::write(const void* src, size_t length)
 {
@@ -105,6 +121,6 @@
 {
     WritingBuffer& writingBuffer = stream.getWritingBuffer();
-    while (!failed()) {
-        if (isInterrupted()) return false;
+    while (*this) {
+        if (checkInterrupted()) return false;
 
         if (writingBuffer.write()) {
@@ -112,5 +128,4 @@
         } else if (writingBuffer.failed()) {
             setErrorCode(writingBuffer.getErrorCode());
-            return false;
         } else {
             Waiter* waiter = stream.getWaiter();
@@ -132,12 +147,12 @@
     readingBuffer.reset();
     readingOffset = 0;
-    while (true) {
-        if (isInterrupted()) return false;
+    while (*this) {
+        if (checkInterrupted()) return false;
 
         if (readingBuffer.read()) {
-            return !readingBuffer.isEmpty();
+            eof = readingBuffer.isEmpty();
+            return !eof;
         } else if (readingBuffer.failed()) {
             setErrorCode(readingBuffer.getErrorCode());
-            return false;
         } else {
             Waiter* waiter = stream.getWaiter();
@@ -149,4 +164,5 @@
         }
     }
+    return false;
 }
 
Index: src/xplcommon/BlockingStream.h
===================================================================
--- src/xplcommon/BlockingStream.h	(revision 30)
+++ src/xplcommon/BlockingStream.h	(revision 31)
@@ -35,4 +35,5 @@
 
 #include "BufferedStream.h"
+
 #include "Waiter.h"
 #include "WaitableEvent.h"
@@ -76,4 +77,9 @@
     size_t readingOffset;
 
+    /**
+     * Indicate if the end-of-file has been reached while reading.
+     */
+    bool eof;
+
 public:
     /**
@@ -83,4 +89,10 @@
      */
     BlockingStream(BufferedStream& stream);
+
+    /**
+     * Determine if the stream has neither failed nor been
+     * interrupted.
+     */
+    operator bool() const;
 
     /**
@@ -102,4 +114,9 @@
      */
     bool read(void* dest, size_t length);
+
+    /**
+     * Skip the given number of bytes.
+     */
+    bool skip(size_t length);
 
     /**
@@ -125,4 +142,11 @@
      */
     bool fill();
+
+    /**
+     * Check for the stream being interrupted.
+     *
+     * @return if the stream is interrupted, false otherwise
+     */
+    bool checkInterrupted();
 };
 
@@ -135,6 +159,14 @@
     event(stream.getWaiter()),
     interrupted(false),
-    readingOffset(0)
+    readingOffset(0),
+    eof(false)
 {
+}
+
+//------------------------------------------------------------------------------
+
+inline BlockingStream::operator bool() const
+{
+    return !failed() && !interrupted && !eof;
 }
 
@@ -144,4 +176,11 @@
 {
     event.fire();
+}
+
+//------------------------------------------------------------------------------
+
+inline bool BlockingStream::isInterrupted()
+{
+    return interrupted;
 }
 
Index: src/xplcommon/DataStream.cc
===================================================================
--- src/xplcommon/DataStream.cc	(revision 31)
+++ src/xplcommon/DataStream.cc	(revision 31)
@@ -0,0 +1,120 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of libxplcommon, a common utility library for
+// development related to X-Plane
+
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// The views and conclusions contained in the software and documentation are those
+// of the authors and should not be interpreted as representing official policies,
+// either expressed or implied, of the FreeBSD Project.
+
+//------------------------------------------------------------------------------
+
+#include "DataStream.h"
+
+//------------------------------------------------------------------------------
+
+using xplcommon::DataStream;
+
+using std::string;
+using std::min;
+
+//------------------------------------------------------------------------------
+
+size_t DataStream::readLength()
+{
+    size_t length = 0;
+
+    uint8_t d = 0;
+    do {
+        d = readU8();
+        length <<= 7;
+        length |= d&0x7f;
+    } while( (d&0x80)==0x80 );
+
+    return length;
+}
+
+//------------------------------------------------------------------------------
+
+string DataStream::readString()
+{
+    size_t length = readLength();
+    if (!*this) return "";
+
+    char* buffer = new char[length];
+    if (read(buffer, length)) {
+        string str(buffer, length);
+        delete buffer;
+        return str;
+    } else {
+        delete buffer;
+        return "";
+    }
+}
+
+//------------------------------------------------------------------------------
+
+size_t DataStream::readString(char* dest, size_t capacity)
+{
+    size_t length = readLength();
+    size_t destLength = 0;
+    if (*this && length>0) {
+        size_t destLength = min(length, capacity - 1);
+        if (read(dest, destLength)) {
+            if (destLength<length) {
+                if (!skip(length - destLength)) {
+                    length = destLength = 0;
+                }
+            }
+        } else {
+            length = destLength = 0;
+        }
+    } else {
+        length = 0;
+    }
+
+    dest[destLength] = '\0';
+
+    return length;
+}
+
+//------------------------------------------------------------------------------
+
+bool DataStream::writeLength(size_t length)
+{
+    do {
+        uint8_t x = length&0x7f;
+        length >>= 7;
+        if (length>0) x |= 0x80;
+        if (!writeU8(x)) return false;
+    } while(length>0);
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/xplcommon/DataStream.h
===================================================================
--- src/xplcommon/DataStream.h	(revision 31)
+++ src/xplcommon/DataStream.h	(revision 31)
@@ -0,0 +1,445 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of libxplcommon, a common utility library for
+// development related to X-Plane
+
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are met:
+
+// 1. Redistributions of source code must retain the above copyright notice, this
+//    list of conditions and the following disclaimer.
+// 2. Redistributions in binary form must reproduce the above copyright notice,
+//    this list of conditions and the following disclaimer in the documentation
+//    and/or other materials provided with the distribution.
+
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+// ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// The views and conclusions contained in the software and documentation are those
+// of the authors and should not be interpreted as representing official policies,
+// either expressed or implied, of the FreeBSD Project.
+
+#ifndef XPLCOMMON_DATASTREAM_H
+#define XPLCOMMON_DATASTREAM_H
+//------------------------------------------------------------------------------
+
+#include "BlockingStream.h"
+
+#include <string>
+
+#include <cstring>
+
+#include <inttypes.h>
+
+//------------------------------------------------------------------------------
+
+namespace xplcommon {
+
+//------------------------------------------------------------------------------
+
+/**
+ * A blocking stream with extra functions to read and write values of
+ * basic data types. The values are read and written in host
+ * byte-order, so this stream does not work properly between machines
+ * of different endianness.
+ *
+ * The reading functions return the value read and you can check of
+ * error or interruption only by calling the appropriate
+ * functions. The purpose of this behaviour is that the data could be
+ * read item-by-item without having to check for the stream's
+ * health all the time. Such checks are needed only, if some decision
+ * is made, based on the data. In case of an error, the read functions
+ * return some well-define default value, typically 0.
+ *
+ * The writing functions return a boolean value indicating success or
+ * failure, but writing can be handled the same way as reading from
+ * the point of view of error checking.
+ */
+class DataStream : public BlockingStream
+{
+public:
+    /**
+     * Construct the stream.
+     */
+    DataStream(BufferedStream& stream);
+
+    /**
+     * Template for simple, type-specific reading.
+     */
+    template <typename T> T readT(T defaultValue = 0);
+
+    /**
+     * Read a signed, 8-bit value
+     */
+    int8_t readS8();
+
+    /**
+     * Read an unsigned, 8-bit value
+     */
+    uint8_t readU8();
+
+    /**
+     * Read a signed, 16-bit value
+     */
+    int16_t readS16();
+
+    /**
+     * Read an unsigned, 16-bit value
+     */
+    uint16_t readU16();
+
+    /**
+     * Read a signed, 32-bit value
+     */
+    int32_t readS32();
+
+    /**
+     * Read an unsigned, 32-bit value
+     */
+    uint32_t readU32();
+
+    /**
+     * Read a signed, 64-bit value
+     */
+    int64_t readS64();
+
+    /**
+     * Read an unsigned, 64-bit value
+     */
+    uint64_t readU64();
+
+    /**
+     * Read a float value.
+     */
+    float readFloat();
+
+    /**
+     * Read a double value.
+     */
+    double readDouble();
+
+    /**
+     * Read a long double value.
+     */
+    long double readLongDouble();
+
+    /**
+     * Read a length value. This is a of at most 32 bits, represented
+     * by a variable number of bytes for space efficiency.
+     *
+     * @see writeLength()
+     */
+    size_t readLength();
+
+    /**
+     * Read a string.
+     */
+    std::string readString();
+
+    /**
+     * Read a string into the given buffer. If the string is too large
+     * for the buffer, its end will be discarded. A trailing 0 is
+     * added in cases!
+     *
+     * @return the actual length of the string.
+     */
+    size_t readString(char* dest, size_t capacity);
+
+    /**
+     * Write a signed, 8-bit value
+     */
+    bool writeS8(int8_t x);
+
+    /**
+     * Write an unsigned, 8-bit value
+     */
+    bool writeU8(uint8_t x);
+
+    /**
+     * Write a signed, 16-bit value
+     */
+    bool writeS16(int16_t x);
+
+    /**
+     * Write an unsigned, 16-bit value
+     */
+    bool writeU16(uint16_t x);
+
+    /**
+     * Write a signed, 32-bit value
+     */
+    bool writeS32(int32_t x);
+
+    /**
+     * Write an unsigned, 32-bit value
+     */
+    bool writeU32(uint32_t x);
+
+    /**
+     * Write a signed, 64-bit value
+     */
+    bool writeS64(int64_t x);
+
+    /**
+     * Write an unsigned, 64-bit value
+     */
+    bool writeU64(uint64_t x);
+
+    /**
+     * Write a float value.
+     */
+    bool writeFloat(float x);
+
+    /**
+     * Write a double value.
+     */
+    bool writeDouble(double x);
+
+    /**
+     * Write a long double value.
+     */
+    bool writeLongDouble(long double x);
+
+    /**
+     * Write a length value. This a 32-bit value written as a sequence
+     * of bytes of variable length. A bytes's lower 7 bits contain
+     * parts of the value. The MSB indicates if another byte is left.
+     */
+    bool writeLength(size_t length);
+
+    /**
+     * Write a string. First the string's length is written as a
+     * length (see writeLength()), then the characters of the string
+     * without the trailing 0.
+     */
+    bool writeString(const char* s, size_t length);
+
+    /**
+     * Write a string. First the string's length is written as a
+     * length (see writeLength()), then the characters of the string
+     * without the trailing 0.
+     */
+    bool writeString(const std::string& s);
+
+    /**
+     * Write a string. First the string's length is written as a
+     * length (see writeLength()), then the characters of the string
+     * without the trailing 0.
+     */
+    bool writeString(const char* s);
+};
+
+//------------------------------------------------------------------------------
+// Inline definitions
+//------------------------------------------------------------------------------
+
+inline DataStream::DataStream(BufferedStream& stream) :
+    BlockingStream(stream)
+{
+}
+
+//------------------------------------------------------------------------------
+
+template <typename T> T DataStream::readT(T defaultValue)
+{
+    T value = defaultValue;
+    read(&value, sizeof(value));
+    return value;
+}
+
+//------------------------------------------------------------------------------
+
+inline int8_t DataStream::readS8()
+{
+    return readT<int8_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline uint8_t DataStream::readU8()
+{
+    return readT<uint8_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline int16_t DataStream::readS16()
+{
+    return readT<int16_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline uint16_t DataStream::readU16()
+{
+    return readT<uint16_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline int32_t DataStream::readS32()
+{
+    return readT<int32_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline uint32_t DataStream::readU32()
+{
+    return readT<uint32_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline int64_t DataStream::readS64()
+{
+    return readT<int64_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline uint64_t DataStream::readU64()
+{
+    return readT<uint64_t>(0);
+}
+
+//------------------------------------------------------------------------------
+
+inline float DataStream::readFloat()
+{
+    return readT<float>(0.0);
+}
+
+//------------------------------------------------------------------------------
+
+inline double DataStream::readDouble()
+{
+    return readT<double>(0.0);
+}
+
+//------------------------------------------------------------------------------
+
+inline long double DataStream::readLongDouble()
+{
+    return readT<long double>(0.0);
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeS8(int8_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeU8(uint8_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeS16(int16_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeU16(uint16_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeS32(int32_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeU32(uint32_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeS64(int64_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeU64(uint64_t x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeFloat(float x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeDouble(double x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeLongDouble(long double x)
+{
+    return write(&x, sizeof(x));
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeString(const char* s, size_t length)
+{
+    if (!writeLength(length)) return false;
+    else return write(s, length);
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeString(const std::string& s)
+{
+    return writeString(s.c_str(), s.length());
+}
+
+//------------------------------------------------------------------------------
+
+inline bool DataStream::writeString(const char* s)
+{
+    return writeString(s, strlen(s));
+}
+
+//------------------------------------------------------------------------------
+
+} /* namespace xplcommon */
+
+//------------------------------------------------------------------------------
+#endif // XPLCOMMON_DATASTREAM_H
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/xplcommon/Makefile.am
===================================================================
--- src/xplcommon/Makefile.am	(revision 30)
+++ src/xplcommon/Makefile.am	(revision 31)
@@ -11,5 +11,6 @@
 libxplcommon_la_SOURCES= \
 	PseudoRandom.cc		\
-	BlockingStream.cc
+	BlockingStream.cc	\
+	DataStream.cc
 
 if TARGET_API_POSIX
@@ -39,3 +40,4 @@
 	LocalConnector.h	\
 	LocalClientSocket.h	\
-	BlockingStream.h
+	BlockingStream.h	\
+	DataStream.h
Index: test/testblkstream.cc
===================================================================
--- test/testblkstream.cc	(revision 30)
+++ test/testblkstream.cc	(revision 31)
@@ -37,5 +37,5 @@
 #include <xplcommon/LocalClientSocket.h>
 #include <xplcommon/LocalConnector.h>
-#include <xplcommon/BlockingStream.h>
+#include <xplcommon/DataStream.h>
 #include <xplcommon/WaitableEvent.h>
 #include <xplcommon/PseudoRandom.h>
@@ -59,5 +59,5 @@
 using xplcommon::LocalConnector;
 using xplcommon::BufferedStream;
-using xplcommon::BlockingStream;
+using xplcommon::DataStream;
 using xplcommon::PseudoRandom;
 
@@ -69,5 +69,16 @@
 {
 private:
-    BlockingStream* dataStream;
+    static const uint8_t OP_ADD = 1;
+
+    static const uint8_t OP_SUB = 2;
+
+    static const uint8_t OP_MUL = 3;
+
+    static const uint8_t OP_DIV = 4;
+
+private:
+    static int32_t perform(int32_t value, uint8_t operation, int32_t x);
+
+    DataStream* dataStream;
 
 protected:
@@ -83,4 +94,22 @@
 //------------------------------------------------------------------------------
 
+inline int32_t TestThread::perform(int32_t value, uint8_t operation, int32_t x)
+{
+    switch (operation) {
+      case OP_ADD:
+        return value + x;
+      case OP_SUB:
+        return value - x;
+      case OP_MUL:
+        return value * x;
+      case OP_DIV:
+        return value / x;
+      default:
+        return value;
+    }
+}
+
+//------------------------------------------------------------------------------
+
 inline TestThread::TestThread() :
     dataStream(0)
@@ -99,68 +128,80 @@
 void TestThread::communicate(BufferedStream& stream, unsigned seed, bool isClient)
 {
-    static const size_t bufferSize = 1024;
-
-    static const size_t minToWrite = 16;
-    static const size_t maxToWrite = 6124;
-
-    static const size_t reportingInterval = 100000000;
+    static const size_t reportInterval = 100000;
 
     const char* prefix = isClient ? "Client" : "Server";
 
-    dataStream = new BlockingStream(stream);
+    dataStream = new DataStream(stream);
     PseudoRandom random(seed);
 
-
-    size_t numBytesRead = 0;
-    size_t nextReportRead = reportingInterval;
-    size_t numBytesWritten = 0;
-    size_t nextReportWritten = reportingInterval;
-
     bool first = true;
 
-    unsigned char* buffer = new unsigned char[bufferSize];
-
-    while(true) {
-        if (isClient || !first) {
-            unsigned dataSize = random.nextUnsigned(maxToWrite, minToWrite);
-            *reinterpret_cast<uint32_t*>(buffer) = dataSize;
-            while(dataSize>0) {
-                unsigned toWrite = min(static_cast<size_t>(dataSize),
-                                       bufferSize);
-                if (!dataStream->write(buffer, toWrite)) break;
-                dataSize -= toWrite;
-                numBytesWritten += toWrite;
-                if (numBytesWritten>=nextReportWritten) {
-                    printf("%s: written %lu bytes\n", prefix,
-                           static_cast<unsigned long>(numBytesWritten));
-                    nextReportWritten += reportingInterval;
+    size_t numRequests = 0;
+    size_t nextReportRequests = reportInterval;
+
+    int32_t value = 0;
+    if (isClient) {
+        value = random.next();
+        dataStream->writeS32(value);
+    } else {
+        value = dataStream->readS32();
+    }
+
+    int32_t newValue = value;
+
+    while(*dataStream) {
+        if (!first || isClient) {
+            unsigned numOperations = random.nextUnsigned(20, 1);
+            dataStream->writeU16(numOperations);
+            //printf("%s: numOperations=%u\n", prefix, numOperations);
+            for(unsigned i = 0; i<numOperations; ++i) {
+                double r = random.nextDouble();
+                uint8_t operation =
+                    (r<0.4) ? OP_ADD : ((r<0.8) ? OP_SUB :
+                                        ((r<0.98) ? OP_MUL : OP_DIV));
+                int32_t x = random.next();
+
+                dataStream->writeU8(operation);
+                dataStream->writeS32(x);
+
+                newValue = perform(newValue, operation, x);
+            }
+            if (!dataStream->flush()) break;
+
+            if (isClient) {
+                value = newValue;
+                ++numRequests;
+                if (numRequests>=nextReportRequests) {
+                    printf("%s: after %lu requests, value=%d\n",
+                           prefix, static_cast<unsigned long>(numRequests),
+                           value);
+                    nextReportRequests += reportInterval;
                 }
             }
-            if (!dataStream->flush()) break;
-        }
-
-        uint32_t dataSize = 0;
-        if (!dataStream->read(&dataSize, sizeof(dataSize))) break;
-        assert(dataSize>=minToWrite);
-        assert(dataSize<=maxToWrite);
-        dataSize -= sizeof(dataSize);
-        numBytesRead += 4;
-        while(dataSize>0) {
-            unsigned toRead = min(static_cast<size_t>(dataSize),
-                                  bufferSize);
-            if (!dataStream->read(buffer, toRead)) break;
-            dataSize -= toRead;
-            numBytesRead += toRead;
-            if (numBytesRead>=nextReportRead) {
-                printf("%s: read %lu bytes\n", prefix,
-                       static_cast<unsigned long>(numBytesRead));
-                nextReportRead += reportingInterval;
+        }
+
+        unsigned numOperations = dataStream->readU16();
+        for (unsigned i = 0; *dataStream && i<numOperations; ++i) {
+            uint8_t operation = dataStream->readU8();
+            int32_t x = dataStream->readS32();
+            newValue = perform(newValue, operation, x);
+        }
+
+        if (*dataStream) {
+            if (!isClient) {
+                value = newValue;
+                ++numRequests;
+                if (numRequests>=nextReportRequests) {
+                    printf("%s: after %lu requests, value=%d\n",
+                           prefix, static_cast<unsigned long>(numRequests),
+                           value);
+                    nextReportRequests += reportInterval;
+                }
             }
         }
 
+
         first = false;
     }
-
-    delete [] buffer;
 
     if (dataStream->failed()) {
@@ -171,7 +212,6 @@
     }
 
-    printf("%s: written %lu bytes and read %lu bytes\n", prefix,
-           static_cast<unsigned long>(numBytesWritten),
-           static_cast<unsigned long>(numBytesRead));
+    printf("%s: value=%d, %lu requests\n", prefix, value,
+           static_cast<unsigned long>(numRequests));
 }
 
@@ -275,9 +315,9 @@
 
     Thread::sleep(60000);
-    //Thread::sleep(5000);
-    printf("Signalling the server thread\n");
-    serverThread.interrupt();
-    //clientThread.interrupt();
-    printf("Signalled the server thread\n");
+    //Thread::sleep(1000);
+    printf("Signalling the client thread\n");
+    //serverThread.interrupt();
+    clientThread.interrupt();
+    printf("Signalled the client thread\n");
 
     printf("Waiting for the client thread\n");
