feat: completed AutoReadStream.

NEED TEST: BufferStream, AutoReadStream.

1 files changed, 32 insertions, 5 deletions

diff --git a/src/common/io/AutoReadStream.cpp b/src/common/io/AutoReadStream.cpp
index 1acf8930..edb77a47 100644
--- a/src/common/io/AutoReadStream.cpp
+++ b/src/common/io/AutoReadStream.cpp
@@ -1,14 +1,22 @@
 #include "cru/common/io/AutoReadStream.h"
-#include <vector>
 #include "cru/common/io/Stream.h"
 
 namespace cru::io {
 
 AutoReadStream::AutoReadStream(Stream* stream, bool auto_delete,
                                const AutoReadStreamOptions& options) {
+  auto buffer_stream_options = options.GetBufferStreamOptions();
+  size_per_read_ = buffer_stream_options.GetBlockSizeOrDefault();
+  buffer_stream_ = std::make_unique<BufferStream>(buffer_stream_options);
   background_thread_ = std::thread(&AutoReadStream::BackgroundThreadRun, this);
 }
 
+AutoReadStream::~AutoReadStream() {
+  if (auto_delete_) {
+    delete stream_;
+  }
+}
+
 bool AutoReadStream::CanSeek() { return false; }
 
 Index AutoReadStream::Seek(Index offset, SeekOrigin origin) {
@@ -16,6 +24,13 @@ Index AutoReadStream::Seek(Index offset, SeekOrigin origin) {
       u"AutoReadStream does not support seek.");
 }
 
+bool AutoReadStream::CanRead() { return true; }
+
+Index AutoReadStream::Read(std::byte* buffer, Index offset, Index size) {
+  std::unique_lock lock(buffer_stream_mutex_);
+  return buffer_stream_->Read(buffer, offset, size);
+}
+
 bool AutoReadStream::CanWrite() { return stream_->CanWrite(); }
 
 Index AutoReadStream::Write(const std::byte* buffer, Index offset, Index size) {
@@ -24,12 +39,24 @@ Index AutoReadStream::Write(const std::byte* buffer, Index offset, Index size) {
 
 void AutoReadStream::Flush() { stream_->Flush(); }
 
+void AutoReadStream::Close() { stream_->Close(); }
+
 void AutoReadStream::BackgroundThreadRun() {
-  std::unique_lock<std::mutex> lock(buffer_mutex_);
-  std::vector<std::byte>* buffer = nullptr;
-  if (!buffer_list_.empty()) {
+  std::vector<std::byte> buffer(size_per_read_);
+  while (true) {
+    try {
+      auto read = stream_->Read(buffer.data(), buffer.size());
+      if (read == 0) {
+        buffer_stream_->SetEof();
+        break;
+      } else {
+        buffer_stream_->Write(buffer.data(), read);
+      }
+    } catch (const StreamAlreadyClosedException& exception) {
+      buffer_stream_->SetEof();
+      break;
+    }
   }
-  stream_->Read();
 }
 
 }  // namespace cru::io