Change streams to be asynchronous

Major restructuring of how streams work. Neither input nor output
streams are now blocking. This avoids stalling the rest of the client or
server when a peer is slow or unresponsive.

Note that this puts an extra burden on users of streams to make sure
they are allowed to do their work once the underlying transports are
ready (e.g. monitoring fds).

1 files changed, 31 insertions, 5 deletions

diff --git a/common/rfb/HextileDecoder.cxx b/common/rfb/HextileDecoder.cxx
index 742dfb28..34392ea8 100644
--- a/common/rfb/HextileDecoder.cxx
+++ b/common/rfb/HextileDecoder.cxx
@@ -44,12 +44,14 @@ HextileDecoder::~HextileDecoder()
 {
 }
 
-void HextileDecoder::readRect(const Rect& r, rdr::InStream* is,
+bool HextileDecoder::readRect(const Rect& r, rdr::InStream* is,
                               const ServerParams& server, rdr::OutStream* os)
 {
   Rect t;
   size_t bytesPerPixel;
 
+  is->setRestorePoint();
+
   bytesPerPixel = server.pf().bpp/8;
 
   for (t.tl.y = r.tl.y; t.tl.y < r.br.y; t.tl.y += 16) {
@@ -61,33 +63,57 @@ void HextileDecoder::readRect(const Rect& r, rdr::InStream* is,
 
       t.br.x = __rfbmin(r.br.x, t.tl.x + 16);
 
+      if (!is->hasDataOrRestore(1))
+        return false;
+
       tileType = is->readU8();
       os->writeU8(tileType);
 
       if (tileType & hextileRaw) {
+        if (!is->hasDataOrRestore(t.area() * bytesPerPixel))
+          return false;
         os->copyBytes(is, t.area() * bytesPerPixel);
         continue;
       }
 
-      if (tileType & hextileBgSpecified)
+
+      if (tileType & hextileBgSpecified) {
+        if (!is->hasDataOrRestore(bytesPerPixel))
+          return false;
         os->copyBytes(is, bytesPerPixel);
+      }
 
-      if (tileType & hextileFgSpecified)
+      if (tileType & hextileFgSpecified) {
+        if (!is->hasDataOrRestore(bytesPerPixel))
+          return false;
         os->copyBytes(is, bytesPerPixel);
+      }
 
       if (tileType & hextileAnySubrects) {
         rdr::U8 nSubrects;
 
+        if (!is->hasDataOrRestore(1))
+          return false;
+
         nSubrects = is->readU8();
         os->writeU8(nSubrects);
 
-        if (tileType & hextileSubrectsColoured)
+        if (tileType & hextileSubrectsColoured) {
+          if (!is->hasDataOrRestore(nSubrects * (bytesPerPixel + 2)))
+            return false;
           os->copyBytes(is, nSubrects * (bytesPerPixel + 2));
-        else
+        } else {
+          if (!is->hasDataOrRestore(nSubrects * 2))
+            return false;
           os->copyBytes(is, nSubrects * 2);
+        }
       }
     }
   }
+
+  is->clearRestorePoint();
+
+  return true;
 }
 
 void HextileDecoder::decodeRect(const Rect& r, const void* buffer,