Class: VibeZstd::CompressWriter

Inherits:

Object

• Object
• VibeZstd::CompressWriter

Defined in:

lib/vibe_zstd.rb,
ext/vibe_zstd/vibe_zstd.c

Class Method Summary

• .open(io, **options) ⇒ Object

  Block-based resource management Automatically calls finish when block completes.

Instance Method Summary

• #close ⇒ Object

  alias.

• #finish ⇒ Object
• #flush ⇒ Object
• #write(data) ⇒ Object

Class Method Details

.open(io, **options) ⇒ Object

Block-based resource management Automatically calls finish when block completes

# File 'lib/vibe_zstd.rb', line 167

def self.open(io, **options)
  writer = new(io, **options)
  return writer unless block_given?

  begin
    yield writer
  ensure
    writer.finish
  end
end

Instance Method Details

#close ⇒ Object

alias

# File 'ext/vibe_zstd/streaming.c', line 172

static VALUE
vibe_zstd_writer_finish(VALUE self) {
    vibe_zstd_cstream* cstream;
    TypedData_Get_Struct(self, vibe_zstd_cstream, &vibe_zstd_cstream_type, cstream);

    size_t outBufferSize = ZSTD_CStreamOutSize();
    VALUE outBuffer = rb_str_buf_new(outBufferSize);

    ZSTD_inBuffer input = { NULL, 0, 0 };
    size_t remaining;

    // ZSTD_e_end: finalize frame with checksum and epilogue
    // Loop until remaining == 0 (frame complete)
    do {
        ZSTD_outBuffer output = {
            .dst = RSTRING_PTR(outBuffer),
            .size = outBufferSize,
            .pos = 0
        };

        // Return value > 0 means more epilogue data to write
        remaining = ZSTD_compressStream2((ZSTD_CCtx*)cstream->cstream, &output, &input, ZSTD_e_end);
        if (ZSTD_isError(remaining)) {
            rb_raise(rb_eRuntimeError, "Finish failed: %s", ZSTD_getErrorName(remaining));
        }

        if (output.pos > 0) {
            rb_str_set_len(outBuffer, output.pos);
            rb_funcall(cstream->io, rb_intern("write"), 1, outBuffer);
            // No need to resize - buffer capacity remains at outBufferSize
        }
    } while (remaining > 0);

    return self;
}

#finish ⇒ Object

# File 'ext/vibe_zstd/streaming.c', line 172

static VALUE
vibe_zstd_writer_finish(VALUE self) {
    vibe_zstd_cstream* cstream;
    TypedData_Get_Struct(self, vibe_zstd_cstream, &vibe_zstd_cstream_type, cstream);

    size_t outBufferSize = ZSTD_CStreamOutSize();
    VALUE outBuffer = rb_str_buf_new(outBufferSize);

    ZSTD_inBuffer input = { NULL, 0, 0 };
    size_t remaining;

    // ZSTD_e_end: finalize frame with checksum and epilogue
    // Loop until remaining == 0 (frame complete)
    do {
        ZSTD_outBuffer output = {
            .dst = RSTRING_PTR(outBuffer),
            .size = outBufferSize,
            .pos = 0
        };

        // Return value > 0 means more epilogue data to write
        remaining = ZSTD_compressStream2((ZSTD_CCtx*)cstream->cstream, &output, &input, ZSTD_e_end);
        if (ZSTD_isError(remaining)) {
            rb_raise(rb_eRuntimeError, "Finish failed: %s", ZSTD_getErrorName(remaining));
        }

        if (output.pos > 0) {
            rb_str_set_len(outBuffer, output.pos);
            rb_funcall(cstream->io, rb_intern("write"), 1, outBuffer);
            // No need to resize - buffer capacity remains at outBufferSize
        }
    } while (remaining > 0);

    return self;
}

#flush ⇒ Object

# File 'ext/vibe_zstd/streaming.c', line 136

static VALUE
vibe_zstd_writer_flush(VALUE self) {
    vibe_zstd_cstream* cstream;
    TypedData_Get_Struct(self, vibe_zstd_cstream, &vibe_zstd_cstream_type, cstream);

    size_t outBufferSize = ZSTD_CStreamOutSize();
    VALUE outBuffer = rb_str_buf_new(outBufferSize);

    ZSTD_inBuffer input = { NULL, 0, 0 };
    size_t remaining;

    // ZSTD_e_flush: flush internal buffers, making all data readable
    // Loop until remaining == 0 (flush complete)
    do {
        ZSTD_outBuffer output = {
            .dst = RSTRING_PTR(outBuffer),
            .size = outBufferSize,
            .pos = 0
        };

        // Return value > 0 means more flushing needed
        remaining = ZSTD_compressStream2((ZSTD_CCtx*)cstream->cstream, &output, &input, ZSTD_e_flush);
        if (ZSTD_isError(remaining)) {
            rb_raise(rb_eRuntimeError, "Flush failed: %s", ZSTD_getErrorName(remaining));
        }

        if (output.pos > 0) {
            rb_str_set_len(outBuffer, output.pos);
            rb_funcall(cstream->io, rb_intern("write"), 1, outBuffer);
            // No need to resize - buffer capacity remains at outBufferSize
        }
    } while (remaining > 0);

    return self;
}

#write(data) ⇒ Object

# File 'ext/vibe_zstd/streaming.c', line 93

static VALUE
vibe_zstd_writer_write(VALUE self, VALUE data) {
    Check_Type(data, T_STRING);

    vibe_zstd_cstream* cstream;
    TypedData_Get_Struct(self, vibe_zstd_cstream, &vibe_zstd_cstream_type, cstream);

    // Input buffer: pos advances as ZSTD consumes data
    ZSTD_inBuffer input = {
        .src = RSTRING_PTR(data),
        .size = RSTRING_LEN(data),
        .pos = 0
    };

    size_t outBufferSize = ZSTD_CStreamOutSize();
    VALUE outBuffer = rb_str_buf_new(outBufferSize);

    // Process all input data in chunks
    while (input.pos < input.size) {
        ZSTD_outBuffer output = {
            .dst = RSTRING_PTR(outBuffer),
            .size = outBufferSize,
            .pos = 0
        };

        // ZSTD_e_continue: continue compression without flushing
        // Return value is a hint for preferred input size (can be ignored)
        size_t result = ZSTD_compressStream2((ZSTD_CCtx*)cstream->cstream, &output, &input, ZSTD_e_continue);
        if (ZSTD_isError(result)) {
            rb_raise(rb_eRuntimeError, "Compression failed: %s", ZSTD_getErrorName(result));
        }

        // Write any compressed output that was produced
        if (output.pos > 0) {
            rb_str_set_len(outBuffer, output.pos);
            rb_funcall(cstream->io, rb_intern("write"), 1, outBuffer);
            // No need to resize - buffer capacity remains at outBufferSize
        }
    }

    return self;
}