Class: Zstd::StreamingCompress

Inherits:
Object
  • Object
show all
Defined in:
ext/zstdruby/streaming_compress.c

Constant Summary collapse

CONTINUE = 
INT2FIX(ZSTD_e_continue)
FLUSH = 
INT2FIX(ZSTD_e_flush)
END = 
INT2FIX(ZSTD_e_end)

Instance Method Summary collapse

Constructor Details

#initialize(*args) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'ext/zstdruby/streaming_compress.c', line 80

static VALUE
rb_streaming_compress_initialize(int argc, VALUE *argv, VALUE obj)
{
  VALUE kwargs;
  rb_scan_args(argc, argv, "00:", &kwargs);

  struct streaming_compress_t* sc;
  TypedData_Get_Struct(obj, struct streaming_compress_t, &streaming_compress_type, sc);
  size_t const buffOutSize = ZSTD_CStreamOutSize();

  ZSTD_CCtx* ctx = ZSTD_createCCtx();
  if (ctx == NULL) {
    rb_raise(rb_eRuntimeError, "%s", "ZSTD_createCCtx error");
  }
  VALUE dict = set_compress_params(ctx, kwargs);

  sc->ctx = ctx;
  RB_OBJ_WRITE(obj, &sc->dict, dict);
  RB_OBJ_WRITE(obj, &sc->buf, rb_str_new(NULL, buffOutSize));
  sc->buf_size = buffOutSize;
  RB_OBJ_WRITE(obj, &sc->pending, rb_str_new(0, 0));

  return obj;
}

Instance Method Details

#<< ⇒ Object

Same as IO.

#compress(src) ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'ext/zstdruby/streaming_compress.c', line 129

static VALUE
rb_streaming_compress_compress(VALUE obj, VALUE src)
{
  StringValue(src);
  const char* input_data = RSTRING_PTR(src);
  size_t input_size = RSTRING_LEN(src);
  ZSTD_inBuffer input = { input_data, input_size, 0 };

  struct streaming_compress_t* sc;
  TypedData_Get_Struct(obj, struct streaming_compress_t, &streaming_compress_type, sc);

  VALUE result = rb_str_new(0, 0);
  while (input.pos < input.size) {
    const char* output_data = RSTRING_PTR(sc->buf);
    ZSTD_outBuffer output = { (void*)output_data, sc->buf_size, 0 };
    size_t const ret = zstd_stream_compress(sc->ctx, &output, &input, ZSTD_e_continue, false);
    if (ZSTD_isError(ret)) {
      rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(ret));
    }
    rb_str_cat(result, output.dst, output.pos);
  }
  return result;
}

#finish ⇒ Object 



218
219
220
221
222
223
224
225
226
227
228
 
# File 'ext/zstdruby/streaming_compress.c', line 218

static VALUE
rb_streaming_compress_finish(VALUE obj)
{
  struct streaming_compress_t* sc;
  TypedData_Get_Struct(obj, struct streaming_compress_t, &streaming_compress_type, sc);
  VALUE drained = no_compress(sc, ZSTD_e_end);
  VALUE out = rb_str_dup(sc->pending);
  rb_str_cat(out, RSTRING_PTR(drained), RSTRING_LEN(drained));
  rb_str_resize(sc->pending, 0);
  return out;
}

#flush ⇒ Object 



206
207
208
209
210
211
212
213
214
215
216
 
# File 'ext/zstdruby/streaming_compress.c', line 206

static VALUE
rb_streaming_compress_flush(VALUE obj)
{
  struct streaming_compress_t* sc;
  TypedData_Get_Struct(obj, struct streaming_compress_t, &streaming_compress_type, sc);
  VALUE drained = no_compress(sc, ZSTD_e_flush);
  VALUE out = rb_str_dup(sc->pending);
  rb_str_cat(out, RSTRING_PTR(drained), RSTRING_LEN(drained));
  rb_str_resize(sc->pending, 0);
  return out;
}

#print ⇒ Object

Same as IO.

#printf ⇒ Object

Same as IO.

#puts ⇒ Object

Same as IO.

#write(*args) ⇒ Object 



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'ext/zstdruby/streaming_compress.c', line 153

static VALUE
rb_streaming_compress_write(int argc, VALUE *argv, VALUE obj)
{
  size_t total = 0;
  struct streaming_compress_t* sc;
  TypedData_Get_Struct(obj, struct streaming_compress_t, &streaming_compress_type, sc);

  while (argc-- > 0) {
    VALUE str = *argv++;
    StringValue(str);
    const char* input_data = RSTRING_PTR(str);
    size_t input_size = RSTRING_LEN(str);
    ZSTD_inBuffer input = { input_data, input_size, 0 };

    while (input.pos < input.size) {
      const char* output_data = RSTRING_PTR(sc->buf);
      ZSTD_outBuffer output = { (void*)output_data, sc->buf_size, 0 };
      size_t const ret = zstd_stream_compress(sc->ctx, &output, &input, ZSTD_e_continue, false);
      if (ZSTD_isError(ret)) {
        rb_raise(rb_eRuntimeError, "compress error error code: %s", ZSTD_getErrorName(ret));
      }
      /* Directly append to the pending buffer */
      if (output.pos > 0) {
        rb_str_cat(sc->pending, output.dst, output.pos);
      }
    }
    total += RSTRING_LEN(str);
  }

  return SIZET2NUM(total);
}