WIP feat: Implement StreamWriter and StructuredIrStreamWriter for IR data streaming (based on #74).

.gitignore

@@ -4,10 +4,21 @@ build
 cmake-build-*
 dist
 
+# Node.js
+node_modules
+package-lock.json
+
 # Generated lint configs
 .clang-format
 .clang-tidy
 
 # IDEs
 .idea
 .vscode
+
+# Apple
+.DS_Store
+
+# Test
+test/clp-ffi-js
+test/assets

.gitmodules

@@ -4,12 +4,15 @@
 [submodule "src/submodules/fmt"]
 	path = src/submodules/fmt
 	url = https://github.com/fmtlib/fmt.git
+[submodule "src/submodules/msgpack"]
+	path = src/submodules/msgpack
+	url = https://github.com/msgpack/msgpack-c.git
 [submodule "src/submodules/spdlog"]
 	path = src/submodules/spdlog
 	url = https://github.com/gabime/spdlog.git
+[submodule "tools/yscope-dev-utils"]
+    path = tools/yscope-dev-utils
+    url = https://github.com/y-scope/yscope-dev-utils.git
 [submodule "src/submodules/zstd"]
 	path = src/submodules/zstd
 	url = https://github.com/facebook/zstd.git
-[submodule "tools/yscope-dev-utils"]
-	path = tools/yscope-dev-utils
-	url = https://github.com/y-scope/yscope-dev-utils.git

CMakeLists.txt

@@ -83,12 +83,13 @@ set(CLP_FFI_JS_COMMON_COMPILE_OPTIONS
     -fwasm-exceptions
 )
 set(CLP_FFI_JS_COMMON_LINK_OPTIONS
-    -fwasm-exceptions
+#    -fwasm-exceptions
     -sALLOW_MEMORY_GROWTH
     -sEXPORT_ES6
     -sMAXIMUM_MEMORY=4GB
     -sMODULARIZE
     -sWASM_BIGINT
+    -sASYNCIFY=1
 )
 if(CMAKE_BUILD_TYPE MATCHES "Release")
     list(APPEND CLP_FFI_JS_COMMON_COMPILE_OPTIONS
@@ -102,7 +103,9 @@ endif()
 
 set(CLP_FFI_JS_SRC_MAIN
     src/clp_ffi_js/ir/StreamReader.cpp
+    src/clp_ffi_js/ir/StreamWriter.cpp
     src/clp_ffi_js/ir/StructuredIrStreamReader.cpp
+    src/clp_ffi_js/ir/StructuredIrStreamWriter.cpp
     src/clp_ffi_js/ir/StructuredIrUnitHandler.cpp
     src/clp_ffi_js/ir/UnstructuredIrStreamReader.cpp
 )
@@ -119,6 +122,11 @@ set(CLP_FFI_JS_SRC_CLP_CORE
     src/submodules/clp/components/core/src/clp/ReaderInterface.cpp
     src/submodules/clp/components/core/src/clp/streaming_compression/zstd/Decompressor.cpp
     src/submodules/clp/components/core/src/clp/TimestampPattern.cpp
+
+    src/submodules/clp/components/core/src/clp/ffi/ir_stream/Serializer.cpp
+    src/submodules/clp/components/core/src/clp/ffi/ir_stream/encoding_methods.cpp
+    src/submodules/clp/components/core/src/clp/ir/parsing.cpp
+    src/submodules/clp/components/core/src/clp/streaming_compression/zstd/Compressor.cpp
 )
 
 set(CLP_FFI_JS_SRC_FMT src/submodules/fmt/src/format.cc)
@@ -134,6 +142,19 @@ set(CLP_FFI_JS_SRC_ZSTD
     src/submodules/zstd/lib/decompress/zstd_ddict.c
     src/submodules/zstd/lib/decompress/zstd_decompress_block.c
     src/submodules/zstd/lib/decompress/zstd_decompress.c
+
+    src/submodules/zstd/lib/compress/fse_compress.c
+    src/submodules/zstd/lib/compress/hist.c
+    src/submodules/zstd/lib/compress/huf_compress.c
+    src/submodules/zstd/lib/compress/zstd_compress.c
+    src/submodules/zstd/lib/compress/zstd_compress_literals.c
+    src/submodules/zstd/lib/compress/zstd_compress_sequences.c
+    src/submodules/zstd/lib/compress/zstd_compress_superblock.c
+    src/submodules/zstd/lib/compress/zstd_double_fast.c
+    src/submodules/zstd/lib/compress/zstd_fast.c
+    src/submodules/zstd/lib/compress/zstd_lazy.c
+    src/submodules/zstd/lib/compress/zstd_ldm.c
+    src/submodules/zstd/lib/compress/zstd_opt.c
 )
 
 set(CLP_FFI_JS_SUPPORTED_ENVIRONMENTS
@@ -183,6 +204,7 @@ Link options: ${CLP_FFI_JS_LINK_OPTIONS}."
         src/submodules/clp/components/core/src/clp
         src/submodules/clp/components/core/submodules
         src/submodules/fmt/include
+        src/submodules/msgpack/include
         src/submodules/spdlog/include
         src/submodules/zstd/lib
     )
@@ -198,3 +220,22 @@ Link options: ${CLP_FFI_JS_LINK_OPTIONS}."
         ${CLP_FFI_JS_SRC_ZSTD}
     )
 endforeach()
+
+# Copy final JS and WASM outputs to the test directory
+foreach(env ${CLP_FFI_JS_SUPPORTED_ENVIRONMENTS})
+    set(bin_name "ClpFfiJs-${env}")
+    set(js_output "${CMAKE_CURRENT_BINARY_DIR}/${bin_name}.js")
+    set(wasm_output "${CMAKE_CURRENT_BINARY_DIR}/${bin_name}.wasm")
+    set(tsd_output "${CMAKE_CURRENT_BINARY_DIR}/${bin_name}.d.ts")
+    set(test_output_dir "${CMAKE_CURRENT_SOURCE_DIR}/test/clp-ffi-js")
+
+    add_custom_command(
+            TARGET ${bin_name}
+            POST_BUILD
+            COMMAND ${CMAKE_COMMAND} -E make_directory ${test_output_dir}
+            COMMAND ${CMAKE_COMMAND} -E copy_if_different ${js_output} ${test_output_dir}/${bin_name}.js
+            COMMAND ${CMAKE_COMMAND} -E copy_if_different ${wasm_output} ${test_output_dir}/${bin_name}.wasm
+            COMMAND ${CMAKE_COMMAND} -E copy_if_different ${tsd_output} ${test_output_dir}/${bin_name}.d.ts
+            COMMENT "Copying ${bin_name} build output to test/clp-ffi-js directory"
+    )
+endforeach()

package.json

@@ -32,5 +32,8 @@
       "import": "./dist/ClpFfiJs-worker.js",
       "types": "./dist/ClpFfiJs-worker.d.ts"
     }
+  },
+  "devDependencies": {
+    "msgpackr": "^1.11.2"
   }
 }

src/clp_ffi_js/ir/StreamWriter.cpp

@@ -0,0 +1,32 @@
+#include <emscripten/bind.h>
+#include <emscripten/val.h>
+
+#include <clp_ffi_js/ir/StreamWriter.hpp>
+#include <clp_ffi_js/ir/StructuredIrStreamWriter.hpp>
+
+namespace {
+EMSCRIPTEN_BINDINGS(ClpStreamWriter) {
+    // JS types used as inputs
+    emscripten::register_type<clp_ffi_js::ir::WriterOptions>("{compressionLevel: number | undefined}");
+
+    // JS types used as outputs
+
+    emscripten::class_<clp_ffi_js::ir::StreamWriter>("StreamWriter")
+            .constructor(
+                    &clp_ffi_js::ir::StreamWriter::create,
+                    emscripten::return_value_policy::take_ownership()
+            )
+            .function("abort", &clp_ffi_js::ir::StreamWriter::abort)
+            .function("write", &clp_ffi_js::ir::StreamWriter::write)
+            .function("close", &clp_ffi_js::ir::StreamWriter::close)
+            .function("getLastWritePromise",
+                      &clp_ffi_js::ir::StreamWriter::get_last_write_promise)
+            .property("desiredSize", &clp_ffi_js::ir::StreamWriter::get_desired_size);
+}
+}  // namespace
+
+namespace clp_ffi_js::ir {
+auto StreamWriter::create(emscripten::val const& stream, WriterOptions const& writer_options) -> std::unique_ptr<StreamWriter> {
+    return std::make_unique<StructuredIrStreamWriter>(stream, writer_options);
+}
+}  // namespace clp_ffi_js::ir

src/clp_ffi_js/ir/StreamWriter.hpp

@@ -0,0 +1,77 @@
+#ifndef CLP_FFI_JS_IR_STREAMWRITER_HPP
+#define CLP_FFI_JS_IR_STREAMWRITER_HPP
+
+#include <clp/ffi/ir_stream/Serializer.hpp>
+#include <clp/ir/types.hpp>
+#include <emscripten/val.h>
+
+namespace clp_ffi_js::ir {
+// JS types used as inputs
+EMSCRIPTEN_DECLARE_VAL_TYPE(WriterOptions);
+
+class StreamWriter {
+public:
+    using ClpIrSerializer = clp::ffi::ir_stream::Serializer<clp::ir::four_byte_encoded_variable_t>;
+    using BufferView = ClpIrSerializer::BufferView;
+
+    /**
+     * Creates a `StreamWriter` to write to a give string.
+     *
+     * @return The created instance.
+     * @throw ClpFfiJsException if any error occurs.
+     */
+    [[nodiscard]] static auto
+    create(emscripten::val const& stream, WriterOptions const& writer_options)
+            -> std::unique_ptr<StreamWriter>;
+
+    // Delete copy & move constructors and assignment operators
+    StreamWriter(StreamWriter const&) = delete;
+    StreamWriter(StreamWriter&&) = delete;
+    auto operator=(StreamWriter const&) -> StreamWriter& = delete;
+    auto operator=(StreamWriter&&) -> StreamWriter& = delete;
+
+    // Destructor
+    virtual ~StreamWriter() = default;
+
+    /**
+     * FIXME: consider separation.
+     * Writes a passed chunk of data to a WritableStream and its underlying sink.
+     */
+    virtual auto write(emscripten::val chunk) -> void = 0;
+
+    /**
+     * FIXME: look into integrating this with `WritableStreamDefaultWriter.ready`
+     * Flushes the underlying IR buffer and `m_output_stream`.
+     */
+    virtual auto flush() -> void = 0;
+
+    /**
+     * Closes the serializer by writing the buffered results into the output
+     * stream with end-of-stream IR Unit appended in the end.
+     * @return true on success.
+     * @return false on failure.
+     */
+    virtual auto close() -> void = 0;
+
+    /**
+     * Abort any ongoing writing and invalidate the serializer.
+     * @param reason
+     * @return a promise.
+     */
+    virtual auto abort(emscripten::val reason) -> emscripten::val = 0;
+
+    /**
+     * Get the desiredSize property from the writer
+     * @return desiredSize
+     */
+    virtual auto get_desired_size() const -> const int& = 0;
+
+    virtual auto get_last_write_promise() -> emscripten::val = 0;
+
+protected:
+    // TODO: add docs
+    StreamWriter() = default;
+};
+}  // namespace clp_ffi_js::ir
+
+#endif  // CLP_FFI_JS_IR_STREAMWRITER_HPP