Index: src/xplcommon/BlockingStream.cc
===================================================================
--- src/xplcommon/BlockingStream.cc	(revision 29)
+++ src/xplcommon/BlockingStream.cc	(revision 29)
@@ -0,0 +1,158 @@
+// 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 "BlockingStream.h"
+
+#include "ReadingBuffer.h"
+#include "WritingBuffer.h"
+
+#include <cstring>
+#include <cstdio>
+
+//------------------------------------------------------------------------------
+
+using xplcommon::BlockingStream;
+
+using std::min;
+
+//------------------------------------------------------------------------------
+
+bool BlockingStream::isInterrupted()
+{
+    if (failed()) return false;
+
+    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 toCopy = min(readingBuffer.getLength() - readingOffset, length);
+        if (toCopy>0) {
+            memcpy(d, readingBuffer.getData() + readingOffset, toCopy);
+            readingOffset += toCopy;
+            length -= toCopy;
+            d += toCopy;
+        }
+
+        if (length==0) break;
+
+        readingBuffer.reset();
+        readingOffset = 0;
+        while (!failed()) {
+            if (isInterrupted()) return false;
+
+            if (readingBuffer.read()) {
+                if (readingBuffer.isEmpty()) return false;
+                break;
+            } else if (readingBuffer.failed()) {
+                setErrorCode(readingBuffer.getErrorCode());
+            } else {
+                Waiter* waiter = stream.getWaiter();
+                waiter->wait();
+                if (waiter->failed()) {
+                    setErrorCode(waiter->getErrorCode());
+                }
+            }
+        }
+        if (failed()) 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 toCopy = min(length, writingBuffer.getAvailable());
+        if (toCopy>0) {
+            memcpy(writingBuffer.getData() + writingBuffer.getLength(),
+                   s, toCopy);
+            writingBuffer.addLength(toCopy);
+            length -= toCopy;
+            s += toCopy;
+        }
+
+        if (length==0) break;
+        if (!flush()) return false;
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+bool BlockingStream::flush()
+{
+    WritingBuffer& writingBuffer = stream.getWritingBuffer();
+    while (!failed()) {
+        if (isInterrupted()) return false;
+
+        if (writingBuffer.write()) {
+            return true;
+        } else if (writingBuffer.failed()) {
+            setErrorCode(writingBuffer.getErrorCode());
+            return false;
+        } else {
+            Waiter* waiter = stream.getWaiter();
+            waiter->wait();
+            if (waiter->failed()) {
+                setErrorCode(waiter->getErrorCode());
+            }
+        }
+    }
+    return false;
+}
+
+//------------------------------------------------------------------------------
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/xplcommon/BlockingStream.h
===================================================================
--- src/xplcommon/BlockingStream.h	(revision 29)
+++ src/xplcommon/BlockingStream.h	(revision 29)
@@ -0,0 +1,153 @@
+// 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_BLOCKINGSTREAM_H
+#define XPLCOMMON_BLOCKINGSTREAM_H
+//------------------------------------------------------------------------------
+
+#include "Failable.h"
+
+#include "BufferedStream.h"
+#include "Waiter.h"
+#include "WaitableEvent.h"
+
+//------------------------------------------------------------------------------
+
+namespace xplcommon {
+
+//------------------------------------------------------------------------------
+
+/**
+ * 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;
+
+public:
+    /**
+     * Construct the blocking stream for the given buffered stream.
+     *
+     * The stream must have a waiter which is used for waiting.
+     */
+    BlockingStream(BufferedStream& stream);
+
+    /**
+     * 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);
+
+    /**
+     * 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();
+};
+
+//------------------------------------------------------------------------------
+// Inline definitions
+//------------------------------------------------------------------------------
+
+inline BlockingStream::BlockingStream(BufferedStream& stream) :
+    stream(stream),
+    event(stream.getWaiter()),
+    interrupted(false),
+    readingOffset(0)
+{
+}
+
+//------------------------------------------------------------------------------
+
+inline void BlockingStream::interrupt()
+{
+    event.fire();
+}
+
+//------------------------------------------------------------------------------
+
+} /* namespace xplcommon */
+
+//------------------------------------------------------------------------------
+#endif // XPLCOMMON_BLOCKINGSTREAM_H
+
+// Local Variables:
+// mode: C++
+// c-basic-offset: 4
+// indent-tabs-mode: nil
+// End:
Index: src/xplcommon/Buffer.h
===================================================================
--- src/xplcommon/Buffer.h	(revision 5)
+++ src/xplcommon/Buffer.h	(revision 29)
@@ -96,4 +96,9 @@
 
     /**
+     * Get whether the buffer is empty or not.
+     */
+    bool isEmpty() const;
+
+    /**
      * Get the available capacity in the buffer.
      */
@@ -172,4 +177,11 @@
 {
     return length;
+}
+
+//------------------------------------------------------------------------------
+
+inline bool Buffer::isEmpty() const
+{
+    return length==0;
 }
 
Index: src/xplcommon/BufferedStream.h
===================================================================
--- src/xplcommon/BufferedStream.h	(revision 21)
+++ src/xplcommon/BufferedStream.h	(revision 29)
@@ -93,7 +93,7 @@
 //------------------------------------------------------------------------------
 
-// #include "win32/BufferedStream.h"
+#include "win32/BufferedStream.h"
 
-// namespace xplcommon { typedef win32::BufferedStream BufferedStream; }
+namespace xplcommon { typedef win32::BufferedStream BufferedStream; }
 
 //------------------------------------------------------------------------------
Index: src/xplcommon/Makefile.am
===================================================================
--- src/xplcommon/Makefile.am	(revision 27)
+++ src/xplcommon/Makefile.am	(revision 29)
@@ -10,6 +10,7 @@
 
 libxplcommon_la_SOURCES= \
+	PseudoRandom.cc		\
 	Buffer.cc		\
-	PseudoRandom.cc
+	BlockingStream.cc
 
 if TARGET_API_POSIX
@@ -25,4 +26,5 @@
 	config.h		\
 	types.h			\
+	PseudoRandom.h		\
 	Failable.h		\
 	Thread.h		\
@@ -38,3 +40,3 @@
 	LocalConnector.h	\
 	LocalClientSocket.h	\
-	PseudoRandom.h
+	BlockingStream.h
Index: src/xplcommon/posix/ReadingBuffer.cc
===================================================================
--- src/xplcommon/posix/ReadingBuffer.cc	(revision 21)
+++ src/xplcommon/posix/ReadingBuffer.cc	(revision 29)
@@ -63,6 +63,6 @@
         stream.events &= ~POLLIN;
         size_t added = addLength(result);
-        assert(added==result);
-        assert(getLength()==result);
+        assert(static_cast<ssize_t>(added)==result);
+        assert(static_cast<ssize_t>(getLength())==result);
     }
     return result>=0;
Index: src/xplcommon/posix/Waiter.cc
===================================================================
--- src/xplcommon/posix/Waiter.cc	(revision 14)
+++ src/xplcommon/posix/Waiter.cc	(revision 29)
@@ -49,5 +49,7 @@
     {
         Waitable* waitable = i->second;
-        if (waitable->ready()) return true;
+        if (waitable->ready()) {
+            return true;
+        }
     }
     return false;
