cmUVStream: Add cmUVStreamRead() function

2 files changed, 95 insertions, 0 deletions

diff --git a/Source/cmUVStream.h b/Source/cmUVStream.h
index fff2381..5998256 100644
--- a/Source/cmUVStream.h
+++ b/Source/cmUVStream.h
@@ -2,8 +2,11 @@
    file Copyright.txt or https://cmake.org/licensing for details.  */
 #pragma once
 
+#include <cassert>
 #include <istream>
 
+#include <cm3p/uv.h>
+
 #include "cmUVHandlePtr.h"
 #include "cmUVStreambuf.h"
 
@@ -100,3 +103,38 @@ void cmBasicUVPipeIStream<CharT, Traits>::close()
 }
 
 using cmUVPipeIStream = cmBasicUVPipeIStream<char>;
+
+template <typename ReadCallback, typename FinishCallback>
+void cmUVStreamRead(uv_stream_t* stream, ReadCallback onRead,
+                    FinishCallback onFinish)
+{
+  struct ReadData
+  {
+    std::vector<char> Buffer;
+    ReadCallback OnRead;
+    FinishCallback OnFinish;
+  };
+
+  stream->data = new ReadData{ {}, std::move(onRead), std::move(onFinish) };
+  uv_read_start(
+    stream,
+    [](uv_handle_t* s, std::size_t suggestedSize, uv_buf_t* buffer) {
+      auto* data = static_cast<ReadData*>(s->data);
+      data->Buffer.resize(suggestedSize);
+      buffer->base = data->Buffer.data();
+      buffer->len = suggestedSize;
+    },
+    [](uv_stream_t* s, ssize_t nread, const uv_buf_t* buffer) {
+      auto* data = static_cast<ReadData*>(s->data);
+      if (nread > 0) {
+        (void)buffer;
+        assert(buffer->base == data->Buffer.data());
+        data->Buffer.resize(nread);
+        data->OnRead(std::move(data->Buffer));
+      } else if (nread < 0 /*|| nread == UV_EOF*/) {
+        data->OnFinish();
+        uv_read_stop(s);
+        delete data;
+      }
+    });
+}
diff --git a/Tests/CMakeLib/testUVStreambuf.cxx b/Tests/CMakeLib/testUVStreambuf.cxx
index c97e695..f3977d4 100644
--- a/Tests/CMakeLib/testUVStreambuf.cxx
+++ b/Tests/CMakeLib/testUVStreambuf.cxx
@@ -3,6 +3,8 @@
 #include <string>
 #include <vector>
 
+#include <cmext/algorithm>
+
 #include <cm3p/uv.h>
 #include <stdint.h>
 
@@ -472,6 +474,56 @@ bool testUVPipeIStream()
   return true;
 }
 
+bool testUVStreamRead()
+{
+  int pipe[] = { -1, -1 };
+  if (cmGetPipes(pipe) < 0) {
+    std::cout << "cmGetPipes() returned an error" << std::endl;
+    return false;
+  }
+
+  cm::uv_loop_ptr loop;
+  loop.init();
+  cm::uv_pipe_ptr pipeSink;
+  pipeSink.init(*loop, 0);
+  uv_pipe_open(pipeSink, pipe[1]);
+
+  std::string str = "Hello world!";
+  uv_write_t writeReq;
+  uv_buf_t buf;
+  buf.base = &str.front();
+  buf.len = str.length();
+  uv_write(&writeReq, pipeSink, &buf, 1, nullptr);
+  uv_run(loop, UV_RUN_DEFAULT);
+  pipeSink.reset();
+
+  cm::uv_pipe_ptr pipeSource;
+  pipeSource.init(*loop, 0);
+  uv_pipe_open(pipeSource, pipe[0]);
+
+  std::string output;
+  bool finished = false;
+  cmUVStreamRead(
+    pipeSource,
+    [&output](std::vector<char> data) { cm::append(output, data); },
+    [&output, &finished]() {
+      if (output != "Hello world!") {
+        std::cout << "Output was \"" << output
+                  << "\", should be \"Hello world!\"" << std::endl;
+        return;
+      }
+      finished = true;
+    });
+  uv_run(loop, UV_RUN_DEFAULT);
+
+  if (!finished) {
+    std::cout << "finished was not set" << std::endl;
+    return false;
+  }
+
+  return true;
+}
+
 int testUVStreambuf(int argc, char** const argv)
 {
   if (argc < 2) {
@@ -494,5 +546,10 @@ int testUVStreambuf(int argc, char** const argv)
     return -1;
   }
 
+  if (!testUVStreamRead()) {
+    std::cout << "While executing testUVPipeIStream().\n";
+    return -1;
+  }
+
   return 0;
 }