diff --git a/README.md b/README.md index cf4c49c..d79bd96 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ import regular pub fn main() -> Nil { // Decompressing Zstd data. - let compressed = <<>> + let compressed = <<0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x0d, 0x69, 0x00, 0x00, 0x57, 0x69, 0x62, 0x62, 0x6c, 0x65, 0x20, 0x57, 0x6f, 0x62, 0x62, 0x6c, 0x65>> let decompressed = regular.decompress(compressed) // Compressing to Zstd. diff --git a/gleam.toml b/gleam.toml index a36739e..5b8227b 100644 --- a/gleam.toml +++ b/gleam.toml @@ -3,3 +3,11 @@ version = "1.0.0" description = "Gleam bindings to Erlang's zstd module 📚" licences = ["Apache-2.0"] repository = { type = "tangled", user = "folospior.dev", repo = "regular" } + +[dependencies] +gleam_otp = ">= 1.3.0 and < 2.0.0" +gleam_erlang = ">= 1.3.0 and < 2.0.0" +gleam_stdlib = ">= 1.0.5 and < 2.0.0" + +[dev_dependencies] +gleeunit = ">= 1.11.0 and < 2.0.0" diff --git a/manifest.toml b/manifest.toml index 6e7ebce..b38d691 100644 --- a/manifest.toml +++ b/manifest.toml @@ -7,6 +7,14 @@ # You should check this file into your source control repository. packages = [ + { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, + { name = "gleam_otp", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "DE4CA6850842F0266EE95317A25DD6A0A0F20CDFAB7C0ADC2E63251D7C3C72EC" }, + { name = "gleam_stdlib", version = "1.0.5", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "CEE5B6C076A85B45F60C585F4316C63EC8B7127C119D5738C3958A9C4D50404E" }, + { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, ] [requirements] +gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } +gleam_otp = { version = ">= 1.3.0 and < 2.0.0" } +gleam_stdlib = { version = ">= 1.0.5 and < 2.0.0" } +gleeunit = { version = ">= 1.11.0 and < 2.0.0" } diff --git a/src/regular.gleam b/src/regular.gleam index 8df57b9..83c81cc 100644 --- a/src/regular.gleam +++ b/src/regular.gleam @@ -1,7 +1,360 @@ +import gleam/bit_array +import gleam/erlang/process.{type Subject} +import gleam/list +import gleam/otp/actor +import gleam/otp/supervision + /// Compress some data into Zstandard format. -@external(erlang, "zstd", "compress") -pub fn compress(data: BitArray) -> BitArray +/// +/// Returns an error if the data isn't byte-aligned. +@external(erlang, "regular_ffi", "compress") +pub fn compress(data: BitArray) -> Result(BitArray, Nil) /// Decompress some Zstandard data. -@external(erlang, "zstd", "decompress") -pub fn decompress(data: BitArray) -> BitArray +/// +/// Returns an error if the data isn't valid Zstd data or if it isn't byte-aligned. +@external(erlang, "regular_ffi", "decompress") +pub fn decompress(data: BitArray) -> Result(BitArray, Nil) + +type CompressionContext + +@external(erlang, "regular_ffi", "new_compression_context") +fn new_compression_context() -> CompressionContext + +type Compressor { + Compressor(context: CompressionContext) +} + +pub opaque type CompressorMessage { + CompressorStream( + chunk: BitArray, + respond_to: Subject(Result(CompressionResult, Nil)), + ) + CompressorFinishFrame( + chunk: BitArray, + respond_to: Subject(Result(BitArray, Nil)), + ) + CompressorShutdown +} + +/// Starts a new compressor, used to create streamed compressed data. +/// +/// It is encouraged to use [`supervised_compressor`](#supervised_compressor) over this function. +pub fn start_compressor() -> Result( + actor.Started(Subject(CompressorMessage)), + actor.StartError, +) { + actor.new_with_initialiser(5000, fn(self) { + let context = new_compression_context() + + let selector = + process.new_selector() + |> process.select(self) + + actor.initialised(Compressor(context)) + |> actor.selecting(selector) + |> actor.returning(self) + |> Ok + }) + |> actor.on_message(on_compressor_message) + |> actor.start +} + +/// Starts a new supervised compressor, used to create streamed compressed data. +pub fn supervised_compressor() -> supervision.ChildSpecification( + Subject(CompressorMessage), +) { + supervision.worker(start_compressor) +} + +pub type CompressionResult { + /// The provided chunk has been fully compressed. + /// + /// Note: The output MAY be empty and returned upon calling `finish_compressor_frame`. + CompressionFinished(output: BitArray) + /// The provided chunk has been partially compressed. + /// + /// Run `stream_compressor` on the remainder to try to compress the full chunk. + CompressionUnfinished(output: BitArray, remainder: BitArray) +} + +/// Stream data to the compressor. The compressor will only compress ~64kB of data +/// at a time. +/// +/// If the compressor fails to compress all of the data in one go, `Ok(CompressionUnfinished(..))` will be returned. +/// Call `stream_to_compressor` on the remainder to attempt compressing the rest of the chunk. +/// +/// Do so before continuing to the next chunk, otherwise you will end up with malformed data. +/// ``` +/// chunk1 -> chunk1's remainders -> chunk2 +/// ``` +/// +/// Returns `Error(Nil)` if the data isn't byte-aligned. +pub fn stream_to_compressor( + compressor: Subject(CompressorMessage), + waiting timeout: Int, + chunk chunk: BitArray, +) -> Result(CompressionResult, Nil) { + process.call(compressor, waiting: timeout, sending: CompressorStream( + chunk:, + respond_to: _, + )) +} + +/// Stream data to the compressor, continuing to compress until the end of the chunk. +/// +/// Timeout applies to each subsequent call to `stream_to_compressor` - this function just calls +/// that one until it finishes the chunk. +/// +/// Enforce size limits on untrusted data! +pub fn stream_all_to_compressor( + compressor: Subject(CompressorMessage), + waiting timeout: Int, + chunk chunk: BitArray, +) -> Result(BitArray, Nil) { + stream_all_to_compressor_loop(compressor, timeout, chunk, []) +} + +fn stream_all_to_compressor_loop( + compressor: Subject(CompressorMessage), + timeout: Int, + chunk: BitArray, + acc: List(BitArray), +) -> Result(BitArray, Nil) { + let result = stream_to_compressor(compressor, waiting: timeout, chunk:) + + case result { + Ok(CompressionFinished(output)) -> + [output, ..acc] + |> list.reverse + |> bit_array.concat + |> Ok + Ok(CompressionUnfinished(output, remainder)) -> + stream_all_to_compressor_loop(compressor, timeout, remainder, [ + output, + ..acc + ]) + Error(Nil) -> Error(Nil) + } +} + +pub fn shutdown_compressor(compressor: Subject(CompressorMessage)) -> Nil { + process.send(compressor, CompressorShutdown) +} + +/// Flushes the compressor's output buffer, making it ready to compress new data. +/// +/// Returns the final compressed chunk, if successful. +/// +/// Returns `Error(Nil)` if the data isn't byte-aligned. +pub fn finish_compressor_frame( + compressor: Subject(CompressorMessage), + waiting timeout: Int, + final_chunk chunk: BitArray, +) -> Result(BitArray, Nil) { + process.call(compressor, waiting: timeout, sending: CompressorFinishFrame( + chunk:, + respond_to: _, + )) +} + +fn on_compressor_message(compressor: Compressor, message: CompressorMessage) { + case message { + CompressorStream(chunk:, respond_to:) -> { + let result = compress_chunk(compressor.context, chunk) + process.send(respond_to, result) + actor.continue(compressor) + } + CompressorFinishFrame(chunk:, respond_to:) -> { + let result = finish_compress_frame(compressor.context, chunk) + process.send(respond_to, result) + actor.continue(compressor) + } + CompressorShutdown -> actor.stop() + } +} + +@external(erlang, "regular_ffi", "compress_chunk") +fn compress_chunk( + context: CompressionContext, + chunk: BitArray, +) -> Result(CompressionResult, Nil) + +@external(erlang, "regular_ffi", "finish_compress_frame") +fn finish_compress_frame( + context: CompressionContext, + chunk: BitArray, +) -> Result(BitArray, Nil) + +type DecompressionContext + +@external(erlang, "regular_ffi", "new_decompression_context") +fn new_decompression_context() -> DecompressionContext + +type Decompressor { + Decompressor(context: DecompressionContext) +} + +pub opaque type DecompressorMessage { + DecompressorStream( + chunk: BitArray, + respond_to: Subject(Result(DecompressionResult, Nil)), + ) + DecompressorFinishFrame( + chunk: BitArray, + respond_to: Subject(Result(BitArray, Nil)), + ) + DecompressorShutdown +} + +/// Starts a new decompressor, used to decompress data flowing in through a stream. +/// +/// It is encouraged to use [`supervised_decompressor`](#supervised_decompressor) over this function. +pub fn start_decompressor() -> Result( + actor.Started(Subject(DecompressorMessage)), + actor.StartError, +) { + actor.new_with_initialiser(5000, fn(self) { + let context = new_decompression_context() + + let selector = + process.new_selector() + |> process.select(self) + + actor.initialised(Decompressor(context)) + |> actor.selecting(selector) + |> actor.returning(self) + |> Ok + }) + |> actor.on_message(on_decompressor_message) + |> actor.start +} + +/// Starts a new supervised decompressor, used to decompress data flowing in through a stream. +pub fn supervised_decompressor() -> supervision.ChildSpecification( + Subject(DecompressorMessage), +) { + supervision.worker(start_decompressor) +} + +pub type DecompressionResult { + /// The provided chunk has been fully decompressed. + /// + /// Note: The output MAY be empty and returned upon calling `finish_decompressor_frame`. + DecompressionFinished(output: BitArray) + /// The provided chunk has been partially decompressed. + /// + /// Run `stream_decompressor` on the remainder to try to decompress the full chunk. + DecompressionUnfinished(output: BitArray, remainder: BitArray) +} + +/// Stream data to the decompressor. The compressor will only inflate ~64kB of data +/// at a time. +/// +/// If the decompressor fails to inflate all of the data in one go, `Ok(DecompressionUnfinished(..))` will be returned. +/// Call `stream_to_decompressor` on the remainder to attempt compressing the rest of the chunk. +/// +/// Do so before continuing to the next chunk, otherwise you will end up with malformed inflated data. +/// ``` +/// chunk1 -> chunk1's remainders -> chunk2 +/// ``` +/// +/// Returns `Error(Nil)` if the data isn't byte-aligned or isn't valid zstd data. +pub fn stream_to_decompressor( + compressor: Subject(DecompressorMessage), + waiting timeout: Int, + chunk chunk: BitArray, +) -> Result(DecompressionResult, Nil) { + process.call(compressor, waiting: timeout, sending: DecompressorStream( + chunk:, + respond_to: _, + )) +} + +/// Stream data to the decompressor, continuing to inflate until the end of the chunk. +/// +/// Timeout applies to each subsequent call to `stream_to_decompressor` - this function just calls +/// that one until it finishes the chunk. +/// +/// Enforce size limits on untrusted data! +pub fn stream_all_to_decompressor( + compressor: Subject(DecompressorMessage), + waiting timeout: Int, + chunk chunk: BitArray, +) -> Result(BitArray, Nil) { + stream_all_to_decompressor_loop(compressor, timeout, chunk, []) +} + +fn stream_all_to_decompressor_loop( + compressor: Subject(DecompressorMessage), + timeout: Int, + chunk: BitArray, + acc: List(BitArray), +) -> Result(BitArray, Nil) { + let result = stream_to_decompressor(compressor, waiting: timeout, chunk:) + + case result { + Ok(DecompressionFinished(output)) -> + [output, ..acc] + |> list.reverse + |> bit_array.concat + |> Ok + Ok(DecompressionUnfinished(output, remainder)) -> + stream_all_to_decompressor_loop(compressor, timeout, remainder, [ + output, + ..acc + ]) + Error(Nil) -> Error(Nil) + } +} + +pub fn shutdown_decompressor(compressor: Subject(DecompressorMessage)) -> Nil { + process.send(compressor, DecompressorShutdown) +} + +/// Flushes the decompressor's output buffer, making it ready to decompress new data. +/// +/// Returns the final inflated chunk, if successful. +/// +/// Returns `Error(Nil)` if the data isn't byte-aligned or isn't valid zstd data. +pub fn finish_decompressor_frame( + compressor: Subject(DecompressorMessage), + waiting timeout: Int, + final_chunk chunk: BitArray, +) -> Result(BitArray, Nil) { + process.call(compressor, waiting: timeout, sending: DecompressorFinishFrame( + chunk:, + respond_to: _, + )) +} + +fn on_decompressor_message( + compressor: Decompressor, + message: DecompressorMessage, +) { + case message { + DecompressorStream(chunk:, respond_to:) -> { + let result = decompress_chunk(compressor.context, chunk) + process.send(respond_to, result) + actor.continue(compressor) + } + DecompressorFinishFrame(chunk:, respond_to:) -> { + let result = finish_decompress_frame(compressor.context, chunk) + process.send(respond_to, result) + actor.continue(compressor) + } + DecompressorShutdown -> actor.stop() + } +} + +@external(erlang, "regular_ffi", "decompress_chunk") +fn decompress_chunk( + context: DecompressionContext, + chunk: BitArray, +) -> Result(DecompressionResult, Nil) + +@external(erlang, "regular_ffi", "finish_decompress_frame") +fn finish_decompress_frame( + context: DecompressionContext, + chunk: BitArray, +) -> Result(BitArray, Nil) diff --git a/src/regular_ffi.erl b/src/regular_ffi.erl new file mode 100644 index 0000000..df41383 --- /dev/null +++ b/src/regular_ffi.erl @@ -0,0 +1,85 @@ +-module(regular_ffi). +-export([ + compress/1, + decompress/1, + new_compression_context/0, + compress_chunk/2, + finish_compress_frame/2, + new_decompression_context/0, + decompress_chunk/2, + finish_decompress_frame/2 +]). + +-spec compress(Data :: bitstring()) -> {ok, binary()} | {error, nil}. +compress(Data) -> + try zstd:compress(Data) of + Compressed -> {ok, iolist_to_binary(Compressed)} + catch + _:_ -> {error, nil} + end. + +-spec decompress(Data :: bitstring()) -> {ok, binary()} | {error, nil}. +decompress(Data) -> + try zstd:decompress(Data) of + Decompressed -> {ok, iolist_to_binary(Decompressed)} + catch + _:_ -> {error, nil} + end. + +-spec new_compression_context() -> zstd:context(). +new_compression_context() -> + {ok, Context} = zstd:context(compress), + Context. + +-spec compress_chunk(Context :: zstd:context(), Chunk :: bitstring()) -> + {ok, {compression_finished, binary()}} | + {ok, {compression_unfinished, binary(), binary()}} | + {error, nil}. +compress_chunk(Context, Chunk) -> + try zstd:stream(Context, Chunk) of + {continue, Remainder, Output} -> + {ok, {compression_unfinished, Output, iolist_to_binary(Remainder)}}; + {continue, Output} -> {ok, {compression_finished, Output}} + catch + _:_ -> {error, nil} + end. + + +-spec finish_compress_frame(Context :: zstd:context(), Chunk :: bitstring()) -> + {ok, binary()} | + {error, nil}. +finish_compress_frame(Context, Chunk) -> + try zstd:finish(Context, Chunk) of + {done, Output} -> {ok, iolist_to_binary(Output)} + catch + _:_ -> {error, nil} + end. + +-spec new_decompression_context() -> zstd:context(). +new_decompression_context() -> + {ok, Context} = zstd:context(decompress), + Context. + + +-spec decompress_chunk(Context :: zstd:context(), Chunk :: bitstring()) -> + {ok, {decompression_finished, binary()}} | + {ok, {decompression_unfinished, binary(), binary()}} | + {error, nil}. +decompress_chunk(Context, Chunk) -> + try zstd:stream(Context, Chunk) of + {continue, Remainder, Output} -> + {ok, {decompression_unfinished, Output, iolist_to_binary(Remainder)}}; + {continue, Output} -> {ok, {decompression_finished, Output}} + catch + _:_ -> {error, nil} + end. + +-spec finish_decompress_frame(Context :: zstd:context(), Chunk :: bitstring()) -> + {ok, binary()} | + {error, nil}. +finish_decompress_frame(Context, Chunk) -> + try zstd:finish(Context, Chunk) of + {done, Output} -> {ok, iolist_to_binary(Output)} + catch + _:_ -> {error, nil} + end. diff --git a/test/regular_test.gleam b/test/regular_test.gleam index fba3c88..9192a2e 100644 --- a/test/regular_test.gleam +++ b/test/regular_test.gleam @@ -1,13 +1,67 @@ import gleeunit +import regular pub fn main() -> Nil { gleeunit.main() } -// gleeunit test functions end in `_test` -pub fn hello_world_test() { - let name = "Joe" - let greeting = "Hello, " <> name <> "!" +pub fn compress_test() { + let decompressed = <<"Hello, world!!!">> + assert regular.compress(decompressed) + == Ok(<< + 0x28, + 0xb5, + 0x2f, + 0xfd, + 0x20, + 0x0f, + 0x79, + 0x00, + 0x00, + 0x48, + 0x65, + 0x6c, + 0x6c, + 0x6f, + 0x2c, + 0x20, + 0x77, + 0x6f, + 0x72, + 0x6c, + 0x64, + 0x21, + 0x21, + 0x21, + >>) +} - assert greeting == "Hello, Joe!" +pub fn decompress_test() { + let compressed = << + 0x28, + 0xb5, + 0x2f, + 0xfd, + 0x20, + 0x0f, + 0x79, + 0x00, + 0x00, + 0x48, + 0x65, + 0x6c, + 0x6c, + 0x6f, + 0x2c, + 0x20, + 0x77, + 0x6f, + 0x72, + 0x6c, + 0x64, + 0x21, + 0x21, + 0x21, + >> + assert regular.decompress(compressed) == Ok(<<"Hello, world!!!">>) }