bluelib/compress/cstream.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)