Standalone 13/N: Integrate components into ZarrStream_s (#305)

Depends on #304.

src/streaming/acquire.zarr.cpp

@@ -145,6 +145,10 @@ extern "C"
 
     void ZarrStream_destroy(struct ZarrStream_s* stream)
     {
+        if (!finalize_stream(stream)) {
+            return;
+        }
+
         delete stream;
     }
 

src/streaming/array.writer.cpp

@@ -49,7 +49,7 @@ zarr::downsample(const ArrayWriterConfig& config,
                                     shard_size_chunks };
         }
     }
-    downsampled_config.dimensions = std::make_unique<ArrayDimensions>(
+    downsampled_config.dimensions = std::make_shared<ArrayDimensions>(
       std::move(downsampled_dims), config.dtype);
 
     downsampled_config.level_of_detail = config.level_of_detail + 1;
@@ -76,17 +76,17 @@ zarr::downsample(const ArrayWriterConfig& config,
 }
 
 /// Writer
-zarr::ArrayWriter::ArrayWriter(ArrayWriterConfig&& config,
+zarr::ArrayWriter::ArrayWriter(const ArrayWriterConfig& config,
                                std::shared_ptr<ThreadPool> thread_pool)
   : ArrayWriter(std::move(config), thread_pool, nullptr)
 {
 }
 
 zarr::ArrayWriter::ArrayWriter(
-  ArrayWriterConfig&& config,
+  const ArrayWriterConfig& config,
   std::shared_ptr<ThreadPool> thread_pool,
   std::shared_ptr<S3ConnectionPool> s3_connection_pool)
-  : config_{ std::move(config) }
+  : config_{ config }
   , thread_pool_{ thread_pool }
   , s3_connection_pool_{ s3_connection_pool }
   , bytes_to_flush_{ 0 }
@@ -97,7 +97,7 @@ zarr::ArrayWriter::ArrayWriter(
 }
 
 size_t
-zarr::ArrayWriter::write_frame(std::span<std::byte> data)
+zarr::ArrayWriter::write_frame(std::span<const std::byte> data)
 {
     const auto nbytes_data = data.size();
     const auto nbytes_frame =
@@ -239,12 +239,12 @@ zarr::ArrayWriter::make_buffers_() noexcept
 
     for (auto& buf : chunk_buffers_) {
         buf.resize(nbytes);
-        std::fill_n(buf.begin(), nbytes, std::byte(0));
+        std::fill(buf.begin(), buf.end(), std::byte(0));
     }
 }
 
 size_t
-zarr::ArrayWriter::write_frame_to_chunks_(std::span<std::byte> data)
+zarr::ArrayWriter::write_frame_to_chunks_(std::span<const std::byte> data)
 {
     // break the frame into tiles and write them to the chunk buffers
     const auto bytes_per_px = bytes_of_type(config_.dtype);
@@ -423,6 +423,11 @@ zarr::ArrayWriter::flush_()
 void
 zarr::ArrayWriter::close_sinks_()
 {
+    for (auto i = 0; i < data_sinks_.size(); ++i) {
+        EXPECT(finalize_sink(std::move(data_sinks_[i])),
+               "Failed to finalize sink ",
+               i);
+    }
     data_sinks_.clear();
 }
 
@@ -438,14 +443,24 @@ zarr::ArrayWriter::rollover_()
 bool
 zarr::finalize_array(std::unique_ptr<ArrayWriter>&& writer)
 {
+    if (writer == nullptr) {
+        LOG_INFO("Array writer is null. Nothing to finalize.");
+        return true;
+    }
+
     writer->is_finalizing_ = true;
     try {
-        writer->flush_();
+        writer->flush_(); // data sinks finalized here
     } catch (const std::exception& exc) {
         LOG_ERROR("Failed to finalize array writer: ", exc.what());
         return false;
     }
 
+    if (!finalize_sink(std::move(writer->metadata_sink_))) {
+        LOG_ERROR("Failed to finalize metadata sink");
+        return false;
+    }
+
     writer.reset();
     return true;
 }

src/streaming/array.writer.hh

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "zarr.stream.hh"
+#include "zarr.dimension.hh"
 #include "thread.pool.hh"
 #include "s3.connection.hh"
 #include "blosc.compression.params.hh"
@@ -12,10 +12,9 @@
 namespace fs = std::filesystem;
 
 namespace zarr {
-
 struct ArrayWriterConfig
 {
-    std::unique_ptr<ArrayDimensions> dimensions;
+    std::shared_ptr<ArrayDimensions> dimensions;
     ZarrDataType dtype;
     int level_of_detail;
     std::optional<std::string> bucket_name;
@@ -40,10 +39,10 @@ downsample(const ArrayWriterConfig& config,
 class ArrayWriter
 {
   public:
-    ArrayWriter(ArrayWriterConfig&& config,
+    ArrayWriter(const ArrayWriterConfig& config,
                 std::shared_ptr<ThreadPool> thread_pool);
 
-    ArrayWriter(ArrayWriterConfig&& config,
+    ArrayWriter(const ArrayWriterConfig& config,
                 std::shared_ptr<ThreadPool> thread_pool,
                 std::shared_ptr<S3ConnectionPool> s3_connection_pool);
 
@@ -54,7 +53,7 @@ class ArrayWriter
      * @param data The frame data.
      * @return The number of bytes written.
      */
-    [[nodiscard]] size_t write_frame(std::span<std::byte> data);
+    [[nodiscard]] size_t write_frame(std::span<const std::byte> data);
 
   protected:
     ArrayWriterConfig config_;
@@ -89,7 +88,7 @@ class ArrayWriter
     bool should_flush_() const;
     virtual bool should_rollover_() const = 0;
 
-    size_t write_frame_to_chunks_(std::span<std::byte> data);
+    size_t write_frame_to_chunks_(std::span<const std::byte> data);
     void compress_buffers_();
 
     void flush_();

src/streaming/file.sink.cpp

@@ -10,7 +10,7 @@ zarr::FileSink::FileSink(std::string_view filename)
 }
 
 bool
-zarr::FileSink::write(size_t offset, std::span<std::byte> data)
+zarr::FileSink::write(size_t offset, std::span<const std::byte> data)
 {
     const auto bytes_of_buf = data.size();
     if (data.data() == nullptr || bytes_of_buf == 0) {

src/streaming/file.sink.hh

@@ -11,7 +11,7 @@ class FileSink : public Sink
   public:
     explicit FileSink(std::string_view filename);
 
-    bool write(size_t offset, std::span<std::byte> data) override;
+    bool write(size_t offset, std::span<const std::byte> data) override;
 
   protected:
     bool flush_() override;

src/streaming/s3.sink.cpp

@@ -50,7 +50,7 @@ zarr::S3Sink::flush_()
 }
 
 bool
-zarr::S3Sink::write(size_t offset, std::span<std::byte> data)
+zarr::S3Sink::write(size_t offset, std::span<const std::byte> data)
 {
     if (data.data() == nullptr || data.empty()) {
         return true;
@@ -66,7 +66,7 @@ zarr::S3Sink::write(size_t offset, std::span<std::byte> data)
     nbytes_buffered_ = offset - nbytes_flushed_;
 
     size_t bytes_of_data = data.size();
-    std::byte* data_ptr = data.data();
+    const std::byte* data_ptr = data.data();
     while (bytes_of_data > 0) {
         const auto bytes_to_write =
           std::min(bytes_of_data, part_buffer_.size() - nbytes_buffered_);

src/streaming/s3.sink.hh

@@ -17,7 +17,7 @@ class S3Sink : public Sink
            std::string_view object_key,
            std::shared_ptr<S3ConnectionPool> connection_pool);
 
-    bool write(size_t offset, std::span<std::byte> data) override;
+    bool write(size_t offset, std::span<const std::byte> data) override;
 
   protected:
     bool flush_() override;

src/streaming/sink.cpp

@@ -1,8 +1,14 @@
 #include "sink.hh"
+#include "macros.hh"
 
 bool
 zarr::finalize_sink(std::unique_ptr<zarr::Sink>&& sink)
 {
+    if (sink == nullptr) {
+        LOG_INFO("Sink is null. Nothing to finalize.");
+        return true;
+    }
+
     if (!sink->flush_()) {
         return false;
     }

src/streaming/sink.hh

@@ -18,7 +18,7 @@ class Sink
      * @return True if the write was successful, false otherwise.
      */
     [[nodiscard]] virtual bool write(size_t offset,
-                                     std::span<std::byte> buf) = 0;
+                                     std::span<const std::byte> buf) = 0;
 
   protected:
     [[nodiscard]] virtual bool flush_() = 0;