Index: src/hu/varadiistvan/scpl/io/BlockingStream.cc
===================================================================
--- src/hu/varadiistvan/scpl/io/BlockingStream.cc	(revision 10)
+++ src/hu/varadiistvan/scpl/io/BlockingStream.cc	(revision 10)
@@ -0,0 +1,175 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of VSCPL, a simple cross-platform utility library
+
+// 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 "BlockingStream.h"
+
+#include "ReadingBuffer.h"
+#include "WritingBuffer.h"
+#include "Waiter.h"
+
+#include <cstring>
+#include <cstdio>
+
+//------------------------------------------------------------------------------
+
+using hu::varadiistvan::scpl::io::BlockingStream;
+
+using std::min;
+
+//------------------------------------------------------------------------------
+
+inline bool BlockingStream::checkInterrupted()
+{
+    if (!interrupted) {
+        interrupted = event.check();
+        if (!interrupted && event.failed()) {
+            setErrorCode(event.getErrorCode());
+        }
+    }
+
+    return interrupted;
+}
+
+//------------------------------------------------------------------------------
+
+bool BlockingStream::read(void* dest, size_t length)
+{
+    unsigned char* d = reinterpret_cast<unsigned char*>(dest);
+
+    ReadingBuffer& readingBuffer = stream.getReadingBuffer();
+    while (length>0) {
+        size_t copied = readingBuffer.extract(d, length, readingOffset);
+        readingOffset += copied;
+        length -= copied;
+        d += copied;
+
+        if (length!=0) {
+            if (!fill()) return false;
+        }
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+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)
+{
+    const unsigned char* s = reinterpret_cast<const unsigned char*>(src);
+
+    WritingBuffer& writingBuffer = stream.getWritingBuffer();
+    while(length>0) {
+        size_t copied = writingBuffer.append(s, length);
+        length -= copied;
+        s += copied;
+
+        if (length==0) break;
+        if (!flush()) return false;
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+bool BlockingStream::flush()
+{
+    WritingBuffer& writingBuffer = stream.getWritingBuffer();
+    while (*this) {
+        if (checkInterrupted()) return false;
+
+        if (writingBuffer.write()) {
+            return true;
+        } else if (writingBuffer.failed()) {
+            setErrorCode(writingBuffer.getErrorCode());
+        } else {
+            Waiter* waiter = stream.getWaiter();
+            waiter->wait();
+            if (waiter->failed()) {
+                setErrorCode(waiter->getErrorCode());
+            }
+        }
+    }
+    return false;
+}
+
+//------------------------------------------------------------------------------
+
+bool BlockingStream::fill()
+{
+    ReadingBuffer& readingBuffer = stream.getReadingBuffer();
+
+    readingBuffer.reset();
+    readingOffset = 0;
+    while (*this) {
+        if (checkInterrupted()) return false;
+
+        if (readingBuffer.read()) {
+            eof = readingBuffer.isEmpty();
+            return !eof;
+        } else if (readingBuffer.failed()) {
+            setErrorCode(readingBuffer.getErrorCode());
+        } else {
+            Waiter* waiter = stream.getWaiter();
+            waiter->wait();
+            if (waiter->failed()) {
+                setErrorCode(waiter->getErrorCode());
+                return false;
+            }
+        }
+    }
+    return false;
+}
+
+//------------------------------------------------------------------------------
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/hu/varadiistvan/scpl/io/BlockingStream.h
===================================================================
--- src/hu/varadiistvan/scpl/io/BlockingStream.h	(revision 10)
+++ src/hu/varadiistvan/scpl/io/BlockingStream.h	(revision 10)
@@ -0,0 +1,197 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of VSCPL, a simple cross-platform utility library
+
+// 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 HU_VARADIISTVAN_SCPL_IO_BLOCKINGSTREAM_H
+#define HU_VARADIISTVAN_SCPL_IO_BLOCKINGSTREAM_H
+//------------------------------------------------------------------------------
+
+#include "Failable.h"
+
+#include "BufferedStream.h"
+#include "WaitableEvent.h"
+
+#include <cstdlib>
+
+//------------------------------------------------------------------------------
+
+namespace hu { namespace varadiistvan { namespace scpl { namespace io {
+
+//------------------------------------------------------------------------------
+
+/**
+ * Wrapper for a buffered stream, which an event and provides
+ * blocking, but interruptible read() and write() operations.
+ *
+ * The class is designed in a way, that it expects its constructed
+ * instances to be called only from a single thread at a time, except
+ * for the interrupt() function, which may be called from another
+ * thread anytime.
+ */
+class BlockingStream : public Failable
+{
+private:
+    /**
+     * The buffered stream used.
+     */
+    BufferedStream& stream;
+
+    /**
+     * The waitable event.
+     */
+    WaitableEvent event;
+
+    /**
+     * Indicate if we are interrupted.
+     */
+    bool interrupted;
+
+    /**
+     * The offset in the buffer of the reading stream.
+     */
+    size_t readingOffset;
+
+    /**
+     * Indicate if the end-of-file has been reached while reading.
+     */
+    bool eof;
+
+public:
+    /**
+     * Construct the blocking stream for the given buffered stream.
+     *
+     * The stream must have a waiter which is used for waiting.
+     */
+    BlockingStream(BufferedStream& stream);
+
+    /**
+     * Determine if the stream has neither failed nor been
+     * interrupted.
+     */
+    operator bool() const;
+
+    /**
+     * Interrupt the stream. Pending or future read and write calls
+     * will return indicating error. The isInterrupted() function can
+     * be used to check if the stream was interrupted or not.
+     */
+    void interrupt();
+
+    /**
+     * Indicate if the stream is interrupted or not.
+     */
+    bool isInterrupted();
+
+    /**
+     * Read the given number of bytes into the given buffer.
+     *
+     * @return if the reading has succeeded
+     */
+    bool read(void* dest, size_t length);
+
+    /**
+     * Skip the given number of bytes.
+     */
+    bool skip(size_t length);
+
+    /**
+     * Write the given number of bytes from the given buffer. The data
+     * may not be written to the underlying stream fully, so if you
+     * want to make sure that it is sent, call flush() afterwards.
+     *
+     * @return true on success, false on error.
+     */
+    bool write(const void* src, size_t length);
+
+    /**
+     * Flush anything written so far into the buffer, but not to the
+     * underlying stream.
+     *
+     * @return whether the operation has succeeded.
+     */
+    bool flush();
+
+private:
+    /**
+     * Fill the buffer with data from the stream.
+     */
+    bool fill();
+
+    /**
+     * Check for the stream being interrupted.
+     *
+     * @return if the stream is interrupted, false otherwise
+     */
+    bool checkInterrupted();
+};
+
+//------------------------------------------------------------------------------
+// Inline definitions
+//------------------------------------------------------------------------------
+
+inline BlockingStream::BlockingStream(BufferedStream& stream) :
+    stream(stream),
+    event(stream.getWaiter()),
+    interrupted(false),
+    readingOffset(0),
+    eof(false)
+{
+}
+
+//------------------------------------------------------------------------------
+
+inline BlockingStream::operator bool() const
+{
+    return !failed() && !interrupted && !eof;
+}
+
+//------------------------------------------------------------------------------
+
+inline void BlockingStream::interrupt()
+{
+    event.fire();
+}
+
+//------------------------------------------------------------------------------
+
+inline bool BlockingStream::isInterrupted()
+{
+    return interrupted;
+}
+
+//------------------------------------------------------------------------------
+
+} /* namespace hu::varadiistvan::scpl::io */ } /* namespace hu::varadiistvan::scpl */ } /* namespace hu::varadiistvan */ } /* namespace hu */
+
+//------------------------------------------------------------------------------
+#endif // HU_VARADIISTVAN_SCPL_IO_BLOCKINGSTREAM_H
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/hu/varadiistvan/scpl/io/DataStream.cc
===================================================================
--- src/hu/varadiistvan/scpl/io/DataStream.cc	(revision 10)
+++ src/hu/varadiistvan/scpl/io/DataStream.cc	(revision 10)
@@ -0,0 +1,119 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of VSCPL, a simple cross-platform utility library
+
+// 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 hu::varadiistvan::scpl::io::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/hu/varadiistvan/scpl/io/DataStream.h
===================================================================
--- src/hu/varadiistvan/scpl/io/DataStream.h	(revision 10)
+++ src/hu/varadiistvan/scpl/io/DataStream.h	(revision 10)
@@ -0,0 +1,444 @@
+// Copyright (c) 2013 by István Váradi
+
+// This file is part of VSCPL, a simple cross-platform utility library
+
+// 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 HU_VARADIISTVAN_SCPL_IO_DATASTREAM_H
+#define HU_VARADIISTVAN_SCPL_IO_DATASTREAM_H
+//------------------------------------------------------------------------------
+
+#include "BlockingStream.h"
+
+#include <string>
+
+#include <cstring>
+
+#include <inttypes.h>
+
+//------------------------------------------------------------------------------
+
+namespace hu { namespace varadiistvan { namespace scpl { namespace io {
+
+//------------------------------------------------------------------------------
+
+/**
+ * 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 hu::varadiistvan::scpl::io */ } /* namespace hu::varadiistvan::scpl */ } /* namespace hu::varadiistvan */ } /* namespace hu */
+
+//------------------------------------------------------------------------------
+#endif // HU_VARADIISTVAN_SCPL_IO_DATASTREAM_H
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/hu/varadiistvan/scpl/io/Makefile.am
===================================================================
--- src/hu/varadiistvan/scpl/io/Makefile.am	(revision 8)
+++ src/hu/varadiistvan/scpl/io/Makefile.am	(revision 10)
@@ -1,3 +1,5 @@
 SUBDIRS=
+
+INCLUDES = -I$(top_builddir)/src
 
 if TARGET_API_POSIX
@@ -12,5 +14,7 @@
 noinst_LTLIBRARIES=libvscpl_io.la
 
-libvscpl_io_la_SOURCES=
+libvscpl_io_la_SOURCES=\
+        BlockingStream.cc       \
+        DataStream.cc
 
 if TARGET_API_POSIX
@@ -35,3 +39,5 @@
         LocalSocket.h           \
         LocalAcceptor.h         \
-        LocalServerSocket.h
+        LocalServerSocket.h     \
+        BlockingStream.h        \
+        DataStream.h
Index: test/Makefile.am
===================================================================
--- test/Makefile.am	(revision 8)
+++ test/Makefile.am	(revision 10)
@@ -1,5 +1,5 @@
 INCLUDES=-I$(top_srcdir)/src -I$(top_builddir)/src
 
-noinst_PROGRAMS=testthread testmutex testcondvar testevent testlocsock
+noinst_PROGRAMS=testthread testmutex testcondvar testevent testlocsock testblkstream
 
 LDADD=../src/hu/varadiistvan/scpl/libvscpl.la
@@ -20,2 +20,4 @@
 
 testlocsock_SOURCES=testlocsock.cc
+
+testblkstream_SOURCES=testblkstream.cc
Index: test/testblkstream.cc
===================================================================
--- test/testblkstream.cc	(revision 10)
+++ test/testblkstream.cc	(revision 10)
@@ -0,0 +1,341 @@
+// Copyright (c) 2012 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 <hu/varadiistvan/scpl/io/Waiter.h>
+#include <hu/varadiistvan/scpl/io/LocalServerSocket.h>
+#include <hu/varadiistvan/scpl/io/LocalAcceptor.h>
+#include <hu/varadiistvan/scpl/io/LocalSocket.h>
+#include <hu/varadiistvan/scpl/io/LocalClientSocket.h>
+#include <hu/varadiistvan/scpl/io/LocalConnector.h>
+#include <hu/varadiistvan/scpl/io/DataStream.h>
+#include <hu/varadiistvan/scpl/io/WaitableEvent.h>
+
+#include <hu/varadiistvan/scpl/Thread.h>
+#include <hu/varadiistvan/scpl/PseudoRandom.h>
+
+#include <cstdlib>
+#include <cstdio>
+#include <cassert>
+
+#include <signal.h>
+
+#include <sys/time.h>
+
+//------------------------------------------------------------------------------
+
+using hu::varadiistvan::scpl::io::Waiter;
+using hu::varadiistvan::scpl::io::LocalServerSocket;
+using hu::varadiistvan::scpl::io::LocalAcceptor;
+using hu::varadiistvan::scpl::io::LocalSocket;
+using hu::varadiistvan::scpl::io::LocalClientSocket;
+using hu::varadiistvan::scpl::io::LocalConnector;
+using hu::varadiistvan::scpl::io::BufferedStream;
+using hu::varadiistvan::scpl::io::DataStream;
+
+using hu::varadiistvan::scpl::Thread;
+using hu::varadiistvan::scpl::PseudoRandom;
+
+using std::min;
+
+//------------------------------------------------------------------------------
+
+class TestThread : public Thread
+{
+private:
+    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:
+    TestThread();
+
+public:
+    void interrupt();
+
+protected:
+    void communicate(BufferedStream& stream, unsigned seed, bool isClient);
+};
+
+//------------------------------------------------------------------------------
+
+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)
+{
+}
+
+//------------------------------------------------------------------------------
+
+inline void TestThread::interrupt()
+{
+    if (dataStream!=0) dataStream->interrupt();
+}
+
+//------------------------------------------------------------------------------
+
+void TestThread::communicate(BufferedStream& stream, unsigned seed, bool isClient)
+{
+    static const size_t reportInterval = 100000;
+
+    const char* prefix = isClient ? "Client" : "Server";
+
+    dataStream = new DataStream(stream);
+    PseudoRandom random(seed);
+
+    bool first = true;
+
+    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;
+                }
+            }
+        }
+
+        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;
+    }
+
+    if (dataStream->failed()) {
+        printf("%s: data stream failed with error code: %lu\n", prefix,
+               static_cast<unsigned long>(dataStream->getErrorCode()));
+    } else if (dataStream->isInterrupted()) {
+        printf("%s: data stream was interrupted\n", prefix);
+    }
+
+    printf("%s: value=%d, %lu requests\n", prefix, value,
+           static_cast<unsigned long>(numRequests));
+}
+
+
+//------------------------------------------------------------------------------
+//------------------------------------------------------------------------------
+
+class ServerThread : public TestThread
+{
+public:
+    virtual void run();
+};
+
+//------------------------------------------------------------------------------
+
+void ServerThread::run()
+{
+    Waiter waiter;
+
+    unsigned seed = static_cast<unsigned>(time(0)*1.2);
+    printf("ServerThread::run: seed=%u\n", seed);
+
+    LocalServerSocket serverSocket("test", &waiter);
+    LocalAcceptor& acceptor = serverSocket.getAcceptor();
+
+    while(!acceptor.accept()) {
+        if (acceptor.failed()) {
+            printf("ServerThread::run: acceptor failed...: %lu\n",
+                   (unsigned long)acceptor.getErrorCode());
+            return;
+        }
+        printf("ServerThread::run: waiting...\n");
+        waiter.wait();
+    }
+
+    printf("ServerThread::run: waiting done, received connection\n");
+    LocalSocket* socket = acceptor.getSocket();
+
+    acceptor.accept();
+
+    communicate(*socket, seed, false);
+
+    delete socket;
+}
+
+//------------------------------------------------------------------------------
+//------------------------------------------------------------------------------
+
+class ClientThread : public TestThread
+{
+public:
+    virtual void run();
+};
+
+//------------------------------------------------------------------------------
+
+void ClientThread::run()
+{
+    Waiter waiter;
+
+    unsigned seed = static_cast<unsigned>(time(0)*1.8);
+    printf("ClientThread::run: seed=%u\n", seed);
+
+    printf("ClientThread::run: sleeping\n");
+    sleep(500);
+    printf("ClientThread::run: connecting\n");
+
+    LocalClientSocket socket("test", &waiter);
+
+    LocalConnector& connector = socket.getConnector();
+
+    while(!connector.connect()) {
+        if (connector.failed()) {
+            printf("ClientThread::run: connector failed...\n");
+            return;
+        }
+        printf("ClientThread::run: waiting...\n");
+        waiter.wait();
+    }
+
+    printf("ClientThread::run: connected\n");
+
+    communicate(socket, seed, true);
+}
+
+//------------------------------------------------------------------------------
+//------------------------------------------------------------------------------
+
+int main()
+{
+#if TARGET_API_POSIX
+    signal(SIGPIPE, SIG_IGN);
+#endif
+
+    ServerThread serverThread;
+    ClientThread clientThread;
+
+    printf("Starting threads\n");
+    serverThread.start();
+    clientThread.start();
+
+
+    Thread::sleep(60000);
+    //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");
+    clientThread.join();
+    printf("Waiting for the server thread\n");
+    serverThread.join();
+    printf("Both threads returned\n");
+
+    return 0;
+}
+
+//------------------------------------------------------------------------------
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
