891 lines
22 KiB
C
891 lines
22 KiB
C
#include <blue/compress/compressor.h>
|
|
#include <blue/compress/cstream.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
|
|
/*** PRIVATE DATA *************************************************************/
|
|
|
|
enum cstream_flags {
|
|
CSTREAM_CURSOR_MOVED = 0x01u,
|
|
};
|
|
|
|
struct b_cstream_p {
|
|
enum cstream_flags s_flags;
|
|
b_stream *s_endpoint;
|
|
b_compressor *s_compressor;
|
|
/* s_in is the input buffer, and s_out is the output buffer.
|
|
*
|
|
* the input buffer holds data that will be provided to the
|
|
* (de)compression function. in compression mode, this data is provided
|
|
* by the code using the cstream (via b_cstream_write). in decompression
|
|
* mode, this data is read from s_endpoint.
|
|
*
|
|
* the output buffer holds data produced by the (de)compression
|
|
* function. in compression mode, this data will be written to
|
|
* s_endpoint. in decompression mode, this data will be returned to the
|
|
* code using the cstream (via b_cstream_read)
|
|
*
|
|
* heavy usage of cstream's compressed sections facility can result
|
|
* in the input buffer holding uncompressed data while the stream is in
|
|
* decompression mode. this is handled by the uncompressed read code path.
|
|
*/
|
|
b_ringbuffer *s_in, *s_out;
|
|
b_compressor_mode s_mode;
|
|
|
|
unsigned int s_compression_depth;
|
|
/* tracks the number of bytes read from or written to the endpoint.
|
|
* this counter is not reset at the beginning/end of each section.
|
|
*
|
|
* during compressed sections, this counter is incremented by the number
|
|
* of compressed bytes written/consumed.
|
|
*
|
|
* during uncompressed sections, this counter is incremented by the
|
|
* number of uncompressed bytes written/returned.
|
|
*
|
|
* this does not include bytes read/written while the cursor is moved.
|
|
*/
|
|
size_t s_tx_bytes;
|
|
/* tracks the number of compressed bytes that have passed through this
|
|
* stream in the current section.
|
|
*
|
|
* in compression mode, this tracks the number of post-compression bytes
|
|
* that have been written to the endpoint within the current section,
|
|
* including any bytes written during end_compression_section()
|
|
*
|
|
* in decompression mode, this tracks the number of compressed bytes
|
|
* that were decompressed while reading the current section. it does not
|
|
* include any uncompressed bytes that may have been read from the
|
|
* endpoint while reading a compressed section due to cstream's
|
|
* read-ahead caching behaviour.
|
|
*/
|
|
size_t s_tx_bytes_compressed;
|
|
/* tracks the number of uncompressed bytes that have passed through this
|
|
* stream in the current section.
|
|
*
|
|
* in compression mode, this tracks the number of bytes given to
|
|
* b_cstream_write
|
|
*
|
|
* in decompression mode, this tracks the number of bytes returned by
|
|
* b_cstream_read
|
|
*/
|
|
size_t s_tx_bytes_uncompressed;
|
|
|
|
/* when the endpoint cursor is moved, the previous cursor position is
|
|
* saved here so it can be restored later */
|
|
size_t s_cursor;
|
|
};
|
|
|
|
/*** PRIVATE FUNCTIONS ********************************************************/
|
|
|
|
static enum b_status read_cursor(
|
|
struct b_cstream_p *stream, void *buf, size_t count, size_t *out_nr_read)
|
|
{
|
|
return b_stream_read_bytes(stream->s_endpoint, buf, count, out_nr_read);
|
|
}
|
|
|
|
static enum b_status read_uncompressed(
|
|
struct b_cstream_p *stream, void *buf, size_t count, size_t *out_nr_read)
|
|
{
|
|
size_t remaining = count;
|
|
unsigned char *dest = buf;
|
|
size_t nr_read_from_buf = 0;
|
|
size_t nr_read_from_endpoint = 0;
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
/* liberal usage of begin_compressed_section and end_compressed_section
|
|
* can result in uncompressed data getting stuck in the input buffer.
|
|
* return any data remaining in the input buffer before reading more
|
|
* from the endpoint */
|
|
|
|
while (remaining > 0) {
|
|
const void *data;
|
|
size_t available;
|
|
status = b_ringbuffer_open_read_buffer(
|
|
stream->s_in, &data, &available);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t to_copy = remaining;
|
|
if (to_copy > available) {
|
|
to_copy = available;
|
|
}
|
|
|
|
memcpy(dest, data, to_copy);
|
|
|
|
b_ringbuffer_close_read_buffer(stream->s_in, &data, to_copy);
|
|
|
|
stream->s_tx_bytes_uncompressed += to_copy;
|
|
stream->s_tx_bytes += to_copy;
|
|
dest += to_copy;
|
|
remaining -= to_copy;
|
|
nr_read_from_buf += to_copy;
|
|
}
|
|
|
|
if (remaining == 0) {
|
|
*out_nr_read = nr_read_from_buf;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
status = b_stream_read_bytes(
|
|
stream->s_endpoint, dest, remaining, &nr_read_from_endpoint);
|
|
stream->s_tx_bytes_uncompressed += nr_read_from_endpoint;
|
|
stream->s_tx_bytes += nr_read_from_endpoint;
|
|
|
|
*out_nr_read = nr_read_from_endpoint + nr_read_from_buf;
|
|
|
|
return status;
|
|
}
|
|
|
|
/* read compressed data from the endpoint and store it in the input buffer.
|
|
* note that uncompressed data that is trailing the compressed blob may
|
|
* also be read by this function, but this will be handled by read_uncompressed.
|
|
*/
|
|
static enum b_status refill_input_buffer(struct b_cstream_p *stream)
|
|
{
|
|
enum b_status status = B_SUCCESS;
|
|
size_t nr_read = 0;
|
|
|
|
while (1) {
|
|
void *data;
|
|
size_t capacity;
|
|
status = b_ringbuffer_open_write_buffer(
|
|
stream->s_in, &data, &capacity);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t r = 0;
|
|
status = b_stream_read_bytes(
|
|
stream->s_endpoint, data, capacity, &r);
|
|
|
|
b_ringbuffer_close_write_buffer(stream->s_in, &data, r);
|
|
nr_read += r;
|
|
|
|
if (r < capacity) {
|
|
break;
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (status == B_ERR_NO_SPACE && nr_read > 0) {
|
|
status = B_SUCCESS;
|
|
}
|
|
|
|
return status;
|
|
}
|
|
|
|
/* push compressed data out of the input buffer, through the (de)compressor,
|
|
* and store the resulting uncompressed data in the output buffer */
|
|
static enum b_status refill_output_buffer(struct b_cstream_p *stream)
|
|
{
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
if (b_compressor_eof(stream->s_compressor)) {
|
|
return B_ERR_NO_DATA;
|
|
}
|
|
|
|
if (!b_ringbuffer_available_data_remaining(stream->s_in)) {
|
|
status = refill_input_buffer(stream);
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
return status;
|
|
}
|
|
|
|
size_t bytes_before = b_ringbuffer_available_data_remaining(stream->s_in);
|
|
status = b_compressor_step(stream->s_compressor);
|
|
size_t bytes_after = b_ringbuffer_available_data_remaining(stream->s_in);
|
|
|
|
stream->s_tx_bytes_compressed += (bytes_before - bytes_after);
|
|
stream->s_tx_bytes += (bytes_before - bytes_after);
|
|
|
|
return status;
|
|
}
|
|
|
|
static enum b_status cstream_read(
|
|
struct b_cstream_p *stream, void *buf, size_t count, size_t *out_nr_read)
|
|
{
|
|
if (stream->s_mode != B_COMPRESSOR_MODE_DECOMPRESS) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return read_cursor(stream, buf, count, out_nr_read);
|
|
}
|
|
|
|
if (stream->s_compression_depth == 0) {
|
|
return read_uncompressed(stream, buf, count, out_nr_read);
|
|
}
|
|
|
|
unsigned char *dest = buf;
|
|
size_t nr_read = 0;
|
|
size_t remaining = count;
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
while (remaining > 0) {
|
|
if (!b_ringbuffer_available_data_remaining(stream->s_out)) {
|
|
status = refill_output_buffer(stream);
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
const void *data;
|
|
size_t available;
|
|
status = b_ringbuffer_open_read_buffer(
|
|
stream->s_out, &data, &available);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t to_copy = remaining;
|
|
if (to_copy > available) {
|
|
to_copy = available;
|
|
}
|
|
|
|
memcpy(dest, data, to_copy);
|
|
|
|
b_ringbuffer_close_read_buffer(stream->s_out, &data, to_copy);
|
|
|
|
stream->s_tx_bytes_uncompressed += to_copy;
|
|
dest += to_copy;
|
|
nr_read += to_copy;
|
|
remaining -= to_copy;
|
|
}
|
|
|
|
if (status == B_ERR_NO_DATA) {
|
|
status = B_SUCCESS;
|
|
}
|
|
|
|
*out_nr_read = nr_read;
|
|
return status;
|
|
}
|
|
|
|
static enum b_status write_cursor(
|
|
struct b_cstream_p *stream, const void *buf, size_t count, size_t *nr_written)
|
|
{
|
|
return b_stream_write_bytes(stream->s_endpoint, buf, count, nr_written);
|
|
}
|
|
|
|
static enum b_status write_uncompressed(
|
|
struct b_cstream_p *stream, const void *buf, size_t count, size_t *nr_written)
|
|
{
|
|
size_t w = 0;
|
|
enum b_status status
|
|
= b_stream_write_bytes(stream->s_endpoint, buf, count, &w);
|
|
|
|
stream->s_tx_bytes_uncompressed += w;
|
|
stream->s_tx_bytes += w;
|
|
|
|
*nr_written = w;
|
|
return status;
|
|
}
|
|
|
|
/* push uncompressed data out of the input buffer, through the compressor,
|
|
* and store the resulting compressed data in the output buffer */
|
|
static enum b_status flush_input_buffer(struct b_cstream_p *stream)
|
|
{
|
|
if (!b_ringbuffer_available_data_remaining(stream->s_in)) {
|
|
return B_ERR_NO_DATA;
|
|
}
|
|
|
|
return b_compressor_step(stream->s_compressor);
|
|
}
|
|
|
|
/* push compressed data from the output buffer into the endpoint */
|
|
static enum b_status flush_output_buffer(struct b_cstream_p *stream)
|
|
{
|
|
enum b_status status = B_SUCCESS;
|
|
size_t nr_written = 0;
|
|
|
|
while (1) {
|
|
const void *data;
|
|
size_t capacity;
|
|
status = b_ringbuffer_open_read_buffer(
|
|
stream->s_out, &data, &capacity);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t w = 0;
|
|
status = b_stream_write_bytes(
|
|
stream->s_endpoint, data, capacity, &w);
|
|
|
|
b_ringbuffer_close_read_buffer(stream->s_out, &data, w);
|
|
nr_written += w;
|
|
stream->s_tx_bytes_compressed += w;
|
|
stream->s_tx_bytes += w;
|
|
|
|
if (w < capacity) {
|
|
break;
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (status == B_ERR_NO_DATA && nr_written > 0) {
|
|
status = B_SUCCESS;
|
|
}
|
|
|
|
return status;
|
|
}
|
|
|
|
static enum b_status cstream_write(
|
|
struct b_cstream_p *stream, const void *buf, size_t count,
|
|
size_t *out_nr_written)
|
|
{
|
|
if (stream->s_mode != B_COMPRESSOR_MODE_COMPRESS) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return write_cursor(stream, buf, count, out_nr_written);
|
|
}
|
|
|
|
if (stream->s_compression_depth == 0) {
|
|
return write_uncompressed(stream, buf, count, out_nr_written);
|
|
}
|
|
|
|
const unsigned char *src = buf;
|
|
size_t nr_written = 0;
|
|
size_t remaining = count;
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
while (remaining > 0) {
|
|
if (!b_ringbuffer_write_capacity_remaining(stream->s_out)) {
|
|
status = flush_output_buffer(stream);
|
|
}
|
|
|
|
if (!b_ringbuffer_write_capacity_remaining(stream->s_in)) {
|
|
status = flush_input_buffer(stream);
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
void *data;
|
|
size_t available;
|
|
status = b_ringbuffer_open_write_buffer(
|
|
stream->s_in, &data, &available);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t to_copy = remaining;
|
|
if (to_copy > available) {
|
|
to_copy = available;
|
|
}
|
|
|
|
memcpy(data, src, to_copy);
|
|
|
|
b_ringbuffer_close_write_buffer(stream->s_in, &data, to_copy);
|
|
|
|
stream->s_tx_bytes_uncompressed += to_copy;
|
|
src += to_copy;
|
|
nr_written += to_copy;
|
|
remaining -= to_copy;
|
|
}
|
|
|
|
if (status == B_ERR_NO_DATA) {
|
|
status = B_SUCCESS;
|
|
}
|
|
|
|
*out_nr_written = nr_written;
|
|
return status;
|
|
}
|
|
|
|
static enum b_status skip_uncompressed(
|
|
struct b_cstream_p *stream, size_t count, size_t *out_nr_skipped)
|
|
{
|
|
size_t remaining = count;
|
|
size_t nr_read_from_buf = 0;
|
|
size_t nr_read_from_endpoint = 0;
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
/* liberal usage of begin_compressed_section and end_compressed_section
|
|
* can result in uncompressed data getting stuck in the input buffer.
|
|
* return any data remaining in the input buffer before reading more
|
|
* from the endpoint */
|
|
|
|
while (remaining > 0) {
|
|
const void *data;
|
|
size_t available;
|
|
status = b_ringbuffer_open_read_buffer(
|
|
stream->s_in, &data, &available);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t to_copy = remaining;
|
|
if (to_copy > available) {
|
|
to_copy = available;
|
|
}
|
|
|
|
b_ringbuffer_close_read_buffer(stream->s_in, &data, to_copy);
|
|
|
|
stream->s_tx_bytes_uncompressed += to_copy;
|
|
stream->s_tx_bytes += to_copy;
|
|
remaining -= to_copy;
|
|
nr_read_from_buf += to_copy;
|
|
}
|
|
|
|
if (remaining == 0) {
|
|
if (out_nr_skipped) {
|
|
*out_nr_skipped = nr_read_from_buf;
|
|
}
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
size_t cursor = b_stream_cursor(stream->s_endpoint);
|
|
|
|
status = b_stream_seek(stream->s_endpoint, remaining, B_STREAM_SEEK_CURRENT);
|
|
nr_read_from_endpoint = b_stream_cursor(stream->s_endpoint) - cursor;
|
|
stream->s_tx_bytes_uncompressed += nr_read_from_endpoint;
|
|
stream->s_tx_bytes += nr_read_from_endpoint;
|
|
|
|
if (out_nr_skipped) {
|
|
*out_nr_skipped = nr_read_from_endpoint + nr_read_from_buf;
|
|
}
|
|
|
|
return status;
|
|
}
|
|
|
|
static enum b_status cstream_skip(
|
|
struct b_cstream_p *stream, size_t count, size_t *out_nr_skipped)
|
|
{
|
|
if (stream->s_mode != B_COMPRESSOR_MODE_DECOMPRESS) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_compression_depth == 0) {
|
|
return skip_uncompressed(stream, count, out_nr_skipped);
|
|
}
|
|
|
|
if (b_compressor_eof(stream->s_compressor)
|
|
&& !b_ringbuffer_available_data_remaining(stream->s_out)) {
|
|
if (out_nr_skipped) {
|
|
*out_nr_skipped = 0;
|
|
}
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
size_t nr_read = 0;
|
|
size_t remaining = count;
|
|
enum b_status status = B_SUCCESS;
|
|
|
|
while (remaining > 0) {
|
|
if (!b_ringbuffer_available_data_remaining(stream->s_out)) {
|
|
status = refill_output_buffer(stream);
|
|
}
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
const void *data;
|
|
size_t available;
|
|
status = b_ringbuffer_open_read_buffer(
|
|
stream->s_out, &data, &available);
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
size_t to_copy = remaining;
|
|
if (to_copy > available) {
|
|
to_copy = available;
|
|
}
|
|
|
|
b_ringbuffer_close_read_buffer(stream->s_out, &data, to_copy);
|
|
|
|
stream->s_tx_bytes_uncompressed += to_copy;
|
|
nr_read += to_copy;
|
|
remaining -= to_copy;
|
|
}
|
|
|
|
if (status == B_ERR_NO_DATA) {
|
|
status = B_SUCCESS;
|
|
}
|
|
|
|
if (out_nr_skipped) {
|
|
*out_nr_skipped = nr_read;
|
|
}
|
|
|
|
return status;
|
|
}
|
|
|
|
static enum b_status cstream_reset(struct b_cstream_p *stream)
|
|
{
|
|
if (stream->s_mode != B_COMPRESSOR_MODE_DECOMPRESS) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
stream->s_flags = 0;
|
|
|
|
b_stream_seek(stream->s_endpoint, 0, B_STREAM_SEEK_START);
|
|
b_ringbuffer_clear(stream->s_in);
|
|
b_ringbuffer_clear(stream->s_out);
|
|
b_compressor_reset(stream->s_compressor);
|
|
|
|
stream->s_compression_depth = 0;
|
|
stream->s_tx_bytes = 0;
|
|
stream->s_tx_bytes_uncompressed = 0;
|
|
stream->s_tx_bytes_compressed = 0;
|
|
stream->s_cursor = 0;
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_begin_compressed_section(
|
|
struct b_cstream_p *stream, size_t *tx_uncompressed_bytes)
|
|
{
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (tx_uncompressed_bytes) {
|
|
*tx_uncompressed_bytes = stream->s_tx_bytes_uncompressed;
|
|
}
|
|
|
|
if (stream->s_compression_depth > 0) {
|
|
stream->s_compression_depth++;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
stream->s_compression_depth = 1;
|
|
stream->s_tx_bytes_uncompressed = 0;
|
|
stream->s_tx_bytes_compressed = 0;
|
|
b_compressor_reset(stream->s_compressor);
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_end_compressed_section(
|
|
struct b_cstream_p *stream, size_t *tx_compressed_bytes,
|
|
size_t *tx_uncompressed_bytes)
|
|
{
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
tx_compressed_bytes
|
|
&& (*tx_compressed_bytes = stream->s_tx_bytes_compressed);
|
|
|
|
tx_uncompressed_bytes
|
|
&& (*tx_uncompressed_bytes = stream->s_tx_bytes_uncompressed);
|
|
|
|
if (stream->s_compression_depth > 1) {
|
|
stream->s_compression_depth--;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
stream->s_compression_depth = 0;
|
|
|
|
if (stream->s_mode == B_COMPRESSOR_MODE_DECOMPRESS) {
|
|
stream->s_tx_bytes_compressed = 0;
|
|
stream->s_tx_bytes_uncompressed = 0;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
enum b_status status = B_SUCCESS;
|
|
while (1) {
|
|
status = b_compressor_end(stream->s_compressor);
|
|
if (!B_OK(status) && status != B_ERR_NO_SPACE) {
|
|
break;
|
|
}
|
|
|
|
status = flush_output_buffer(stream);
|
|
|
|
if (!B_OK(status)) {
|
|
break;
|
|
}
|
|
|
|
if (b_compressor_eof(stream->s_compressor)) {
|
|
status = B_SUCCESS;
|
|
break;
|
|
}
|
|
}
|
|
|
|
/* refresh these output variables to account for any data
|
|
* written by b_compressor_end */
|
|
tx_compressed_bytes
|
|
&& (*tx_compressed_bytes = stream->s_tx_bytes_compressed);
|
|
|
|
tx_uncompressed_bytes
|
|
&& (*tx_uncompressed_bytes = stream->s_tx_bytes_uncompressed);
|
|
|
|
if (!B_OK(status)) {
|
|
return status;
|
|
}
|
|
|
|
stream->s_tx_bytes_compressed = 0;
|
|
stream->s_tx_bytes_uncompressed = 0;
|
|
|
|
return flush_output_buffer(stream);
|
|
}
|
|
|
|
static bool cstream_in_compressed_section(const struct b_cstream_p *stream)
|
|
{
|
|
return stream->s_compression_depth > 0;
|
|
}
|
|
|
|
static enum b_status cstream_tx_bytes(const struct b_cstream_p *stream, size_t *out)
|
|
{
|
|
*out = stream->s_tx_bytes;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_tx_bytes_compressed(
|
|
const struct b_cstream_p *stream, size_t *out)
|
|
{
|
|
*out = stream->s_tx_bytes_compressed;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_tx_bytes_uncompressed(
|
|
const struct b_cstream_p *stream, size_t *out)
|
|
{
|
|
*out = stream->s_tx_bytes_uncompressed;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_set_cursor_position(
|
|
struct b_cstream_p *stream, size_t pos)
|
|
{
|
|
if (stream->s_compression_depth > 0) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
if (stream->s_flags & CSTREAM_CURSOR_MOVED) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
stream->s_cursor = b_stream_cursor(stream->s_endpoint);
|
|
|
|
enum b_status status
|
|
= b_stream_seek(stream->s_endpoint, pos, B_STREAM_SEEK_START);
|
|
if (!B_OK(status)) {
|
|
stream->s_cursor = 0;
|
|
return status;
|
|
}
|
|
|
|
stream->s_flags |= CSTREAM_CURSOR_MOVED;
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
static enum b_status cstream_restore_cursor_position(struct b_cstream_p *stream)
|
|
{
|
|
if (!(stream->s_flags & CSTREAM_CURSOR_MOVED)) {
|
|
return B_ERR_BAD_STATE;
|
|
}
|
|
|
|
enum b_status status = b_stream_seek(
|
|
stream->s_endpoint, stream->s_cursor, B_STREAM_SEEK_START);
|
|
stream->s_cursor = 0;
|
|
|
|
if (!B_OK(status)) {
|
|
return status;
|
|
}
|
|
|
|
stream->s_flags &= ~CSTREAM_CURSOR_MOVED;
|
|
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
/*** PUBLIC FUNCTIONS *********************************************************/
|
|
|
|
enum b_status b_cstream_open(
|
|
b_stream *endpoint, b_type compressor_type, b_compressor_mode mode,
|
|
b_cstream **out)
|
|
{
|
|
size_t inbuf_size = 0, outbuf_size = 0;
|
|
enum b_status status = b_compressor_get_buffer_size(
|
|
compressor_type, mode, &inbuf_size, &outbuf_size);
|
|
if (!B_OK(status)) {
|
|
return status;
|
|
}
|
|
|
|
b_cstream *stream = b_object_create(B_TYPE_CSTREAM);
|
|
if (!stream) {
|
|
return B_ERR_NO_MEMORY;
|
|
}
|
|
|
|
struct b_cstream_p *p = b_object_get_private(stream, B_TYPE_CSTREAM);
|
|
b_stream_cfg *cfg = b_object_get_protected(stream, B_TYPE_STREAM);
|
|
|
|
p->s_mode = mode;
|
|
p->s_endpoint = endpoint;
|
|
|
|
cfg->s_mode = (mode == B_COMPRESSOR_MODE_COMPRESS) ? B_STREAM_WRITE
|
|
: B_STREAM_READ;
|
|
|
|
p->s_in = b_ringbuffer_create(inbuf_size + 1);
|
|
if (!B_OK(status)) {
|
|
free(stream);
|
|
return status;
|
|
}
|
|
|
|
p->s_out = b_ringbuffer_create(outbuf_size + 1);
|
|
if (!B_OK(status)) {
|
|
b_cstream_unref(stream);
|
|
return status;
|
|
}
|
|
|
|
p->s_compressor = b_object_create(compressor_type);
|
|
if (!p->s_compressor) {
|
|
b_cstream_unref(stream);
|
|
|
|
return B_ERR_INVALID_ARGUMENT;
|
|
}
|
|
|
|
b_compressor_set_buffer(p->s_compressor, p->s_in, p->s_out);
|
|
b_compressor_set_mode(p->s_compressor, mode);
|
|
|
|
*out = stream;
|
|
return B_SUCCESS;
|
|
}
|
|
|
|
enum b_status b_cstream_read(
|
|
b_cstream *stream, void *buf, size_t count, size_t *out_nr_read)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_read, stream, buf, count, out_nr_read);
|
|
}
|
|
|
|
enum b_status b_cstream_write(
|
|
b_cstream *stream, const void *buf, size_t count, size_t *out_nr_written)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_write, stream, buf, count, out_nr_written);
|
|
}
|
|
|
|
enum b_status b_cstream_skip(b_cstream *stream, size_t count, size_t *out_nr_skipped)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_skip, stream, count, out_nr_skipped);
|
|
}
|
|
|
|
enum b_status b_cstream_reset(b_cstream *stream)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC_0(B_TYPE_CSTREAM, cstream_reset, stream);
|
|
}
|
|
|
|
enum b_status b_cstream_begin_compressed_section(
|
|
b_cstream *stream, size_t *tx_uncompressed_bytes)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_begin_compressed_section, stream,
|
|
tx_uncompressed_bytes);
|
|
}
|
|
|
|
enum b_status b_cstream_end_compressed_section(
|
|
b_cstream *stream, size_t *tx_compressed_bytes, size_t *tx_uncompressed_bytes)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_end_compressed_section, stream,
|
|
tx_compressed_bytes, tx_uncompressed_bytes);
|
|
}
|
|
|
|
bool b_cstream_in_compressed_section(const b_cstream *stream)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC_0(
|
|
B_TYPE_CSTREAM, cstream_in_compressed_section, stream);
|
|
}
|
|
|
|
enum b_status b_cstream_tx_bytes(const b_cstream *stream, size_t *out)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(B_TYPE_CSTREAM, cstream_tx_bytes, stream, out);
|
|
}
|
|
|
|
enum b_status b_cstream_tx_bytes_compressed(const b_cstream *stream, size_t *out)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_tx_bytes_compressed, stream, out);
|
|
}
|
|
|
|
enum b_status b_cstream_tx_bytes_uncompressed(const b_cstream *stream, size_t *out)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_tx_bytes_uncompressed, stream, out);
|
|
}
|
|
|
|
enum b_status b_cstream_set_cursor_position(b_cstream *stream, size_t pos)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC(
|
|
B_TYPE_CSTREAM, cstream_set_cursor_position, stream, pos);
|
|
}
|
|
|
|
enum b_status b_cstream_restore_cursor_position(b_cstream *stream)
|
|
{
|
|
B_CLASS_DISPATCH_STATIC_0(
|
|
B_TYPE_CSTREAM, cstream_restore_cursor_position, stream);
|
|
}
|
|
|
|
/*** VIRTUAL FUNCTIONS ********************************************************/
|
|
|
|
static void cstream_init(b_object *obj, void *priv)
|
|
{
|
|
}
|
|
|
|
static void cstream_fini(b_object *obj, void *priv)
|
|
{
|
|
struct b_cstream_p *stream = priv;
|
|
|
|
if (stream->s_compressor) {
|
|
b_compressor_unref(stream->s_compressor);
|
|
}
|
|
|
|
if (stream->s_in) {
|
|
b_ringbuffer_unref(stream->s_in);
|
|
}
|
|
|
|
if (stream->s_out) {
|
|
b_ringbuffer_unref(stream->s_out);
|
|
}
|
|
}
|
|
|
|
/*** CLASS DEFINITION *********************************************************/
|
|
|
|
B_TYPE_CLASS_DEFINITION_BEGIN(b_cstream)
|
|
B_TYPE_CLASS_INTERFACE_BEGIN(b_object, B_TYPE_OBJECT)
|
|
B_INTERFACE_ENTRY(to_string) = NULL;
|
|
B_TYPE_CLASS_INTERFACE_END(b_object, B_TYPE_OBJECT)
|
|
|
|
B_TYPE_CLASS_INTERFACE_BEGIN(b_stream, B_TYPE_STREAM)
|
|
B_INTERFACE_ENTRY(s_close) = NULL;
|
|
B_INTERFACE_ENTRY(s_seek) = NULL;
|
|
B_INTERFACE_ENTRY(s_tell) = NULL;
|
|
B_INTERFACE_ENTRY(s_getc) = NULL;
|
|
B_INTERFACE_ENTRY(s_read) = b_cstream_read;
|
|
B_INTERFACE_ENTRY(s_write) = b_cstream_write;
|
|
B_INTERFACE_ENTRY(s_reserve) = NULL;
|
|
B_TYPE_CLASS_INTERFACE_END(b_stream, B_TYPE_STREAM)
|
|
B_TYPE_CLASS_DEFINITION_END(b_cstream)
|
|
|
|
B_TYPE_DEFINITION_BEGIN(b_cstream)
|
|
B_TYPE_ID(0xe1e899b5, 0x6a3c, 0x4f9c, 0xafd0, 0xaab3f156615c);
|
|
B_TYPE_EXTENDS(B_TYPE_STREAM);
|
|
B_TYPE_CLASS(b_cstream_class);
|
|
B_TYPE_INSTANCE_PRIVATE(struct b_cstream_p);
|
|
B_TYPE_INSTANCE_INIT(cstream_init);
|
|
B_TYPE_INSTANCE_FINI(cstream_fini);
|
|
B_TYPE_DEFINITION_END(b_cstream)
|