# Implementing Streaming Compression and Decompression with Kanzi: A Complete C++ Guide

> Implement streaming compression and decompression in C++ with Kanzi's CompressedOutputStream and CompressedInputStream. Process data incrementally, leveraging multi-core parallelism for efficiency. A complete guide for developers.

- Repository: [flanglet/kanzi-cpp](https://github.com/flanglet/kanzi-cpp)
- Tags: how-to-guide
- Published: 2026-03-02

---

**Kanzi provides a fully-streaming API through `CompressedOutputStream` and `CompressedInputStream` classes that process data incrementally in configurable blocks, automatically distributing work across CPU cores via task-based parallelism.**

The flanglet/kanzi-cpp repository delivers a high-performance compression engine designed for scenarios where data arrives incrementally or total size is unknown beforehand. Unlike batch-oriented compressors, Kanzi’s I/O layer in `src/io/` enables real-time, block-by-block processing while maintaining random-access capabilities and optional integrity checking.

## Core Streaming Architecture

Kanzi’s streaming implementation centers on two primary I/O classes that manage the full lifecycle of compression and decompression pipelines.

### CompressedOutputStream (Compression Pipeline)

Located in [`src/io/CompressedOutputStream.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/io/CompressedOutputStream.hpp) and [`src/io/CompressedOutputStream.cpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/io/CompressedOutputStream.cpp), this class manages block buffering, transform application, entropy encoding, and bit-stream output. Key methods include `writeHeader()` for emitting stream metadata, `processBuffer()` for accumulating input data, and `submitBlock()` for dispatching work units. The class uses internal `_buffers` to accumulate raw data until reaching the configured block size or receiving an explicit `flush()` call.

### CompressedInputStream (Decompression Pipeline)

Defined in [`src/io/CompressedInputStream.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/io/CompressedInputStream.hpp) and [`src/io/CompressedInputStream.cpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/io/CompressedInputStream.cpp), this component handles header parsing via `readHeader()`, on-demand block decoding through `submitBlock()`, and seekable random access via `seek(bitPos)`. The stream maintains an `_available` byte counter to track ready-to-read data and resets the pipeline when seeking to block boundaries.

### Task-Based Concurrency

When `CONCURRENCY_ENABLED` is defined (the default for modern compilers), Kanzi utilizes a `ThreadPool` from [`src/concurrent.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/concurrent.hpp) to parallelize block processing. The headers define template classes `EncodingTask` (in [`CompressedOutputStream.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/CompressedOutputStream.hpp)) and `DecodingTask` (in [`CompressedInputStream.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/CompressedInputStream.hpp)) that encapsulate the transform-plus-entropy work for individual blocks. These tasks execute XXHash32/64 checksum computation (implemented in [`src/util/XXHash.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/util/XXHash.hpp)) alongside codec operations.

## How Streaming Compression Works

The compression workflow follows a strict pipeline from context configuration to bit-stream finalization.

1. **Context Configuration** – Create a `Context` object (defined in [`src/Context.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/Context.hpp)) specifying the **entropy codec** (e.g., "ANS0"), **transform pipeline** (e.g., "BWT+LZ"), block size, and job count. This context is reusable across multiple streams.

2. **Stream Construction** – Instantiate `CompressedOutputStream` with an underlying `OutputStream` (file, socket, or memory buffer). The constructor immediately calls `writeHeader()` to emit a bit-stream header recording these parameters.

3. **Block Buffering** – Incoming data accumulates in internal `_buffers`. When the buffer reaches the configured block size or upon `flush()`, `processBuffer()` copies the data to an `EncodingTask`.

4. **Parallel Processing** – `submitBlock()` hands tasks to the thread pool. Each `EncodingTask` applies transforms via `TransformFactory` (from [`src/transform/TransformFactory.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/transform/TransformFactory.hpp)), encodes entropy via `EntropyEncoderFactory` (from [`src/entropy/EntropyEncoderFactory.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/entropy/EntropyEncoderFactory.hpp)), and writes results into a shared `DefaultOutputBitStream` (`_obs`), maintaining byte alignment on block boundaries.

5. **Finalization** – Calling `close()` flushes any pending blocks, finalizes the underlying stream, and optionally writes a footer containing the total uncompressed size.

```cpp
#include <fstream>
#include "kanzi/CompressedOutputStream.hpp"
#include "kanzi/Context.hpp"

int main() {
    std::ofstream fout("data.knz", std::ios::binary);
    
    kanzi::Context ctx;
    ctx.putString("entropy", "ANS0");
    ctx.putString("transform", "BWT+LZ");
    ctx.putInt("blockSize", 4*1024*1024);  // 4 MiB blocks
    ctx.putInt("jobs", 4);                 // 4 threads

    kanzi::CompressedOutputStream cos(fout, ctx);
    
    std::string data = "Streaming data chunk...";
    cos.write(data.c_str(), data.size());
    cos.close();  // Flushes automatically
}

```

## How Streaming Decompression Works

Decompression reverses the pipeline using lazy evaluation and supports random access at block granularity.

1. **Header Parsing** – The `CompressedInputStream` constructor calls `readHeader()` to populate a `Context` with the original compression parameters, unless operating in headerless mode.

2. **On-Demand Decoding** – When `read()` requests data, the stream checks `_available` bytes. If insufficient, `submitBlock()` creates a `DecodingTask` for the next compressed block.

3. **Block Processing** – The task reads per-block metadata (size, checksum, transform identifiers), decodes the entropy layer, applies inverse transforms via `TransformFactory`, and verifies checksums.

4. **Data Delivery** – Decoded bytes populate internal `_buffers`. The `read()` method copies from these buffers to the caller’s buffer, updating `_available`.

5. **Random Access** – The `seek(bitPos)` method resets the pipeline, cancels outstanding futures, and re-initializes the block queue, enabling seeks only to block boundaries for large-file archival.

```cpp
#include <fstream>
#include "kanzi/CompressedInputStream.hpp"

int main() {
    std::ifstream fin("data.knz", std::ios::binary);
    kanzi::CompressedInputStream cis(fin);
    
    std::vector<char> buffer(8192);
    std::size_t total = 0;
    
    while (true) {
        std::size_t got = cis.read(buffer.data(), buffer.size());
        total += got;
        if (got < buffer.size()) break;  // EOF
    }
    // buffer[0..total) holds original data
}

```

## C API for Cross-Language Integration

The repository provides a thin C wrapper in [`src/api/Compressor.cpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/api/Compressor.cpp) and [`src/api/Decompressor.cpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/api/Decompressor.cpp) exposing the streaming model through opaque handles. This API enables integration with Python ctypes, Java JNI, and other foreign function interfaces.

- `initCompressor()` creates a `CompressedOutputStream` bound to a `FILE*`.
- `compress()` writes raw buffers incrementally, returning compressed byte counts.
- `disposeCompressor()` finalizes and releases resources.

The decompressor provides mirrored functions: `initDecompressor()`, `decompress()`, and `disposeDecompressor()`.

```c
#include "kanzi/kanzi.h"
#include <stdio.h>

int main() {
    FILE *src = fopen("bigfile.bin", "rb");
    FILE *dst = fopen("bigfile.knz", "wb");
    
    struct cData cfg = {0};
    strcpy(cfg.entropy, "ANS0");
    strcpy(cfg.transform, "BWT+LZ");
    cfg.blockSize = 4*1024*1024;
    cfg.jobs = 4;

    struct cContext *ctx = NULL;
    if (initCompressor(&cfg, dst, &ctx) != 0) return -1;

    unsigned char buf[65536];
    size_t read, written;
    
    while ((read = fread(buf, 1, sizeof(buf), src)) > 0) {
        if (compress(ctx, buf, read, &written) != 0) break;
    }
    
    disposeCompressor(&ctx, &written);
    fclose(src); 
    fclose(dst);
}

```

## Summary

- **Block-based streaming** – Kanzi processes data through `CompressedOutputStream` and `CompressedInputStream` without requiring total input size upfront.
- **Parallel execution** – `EncodingTask` and `DecodingTask` templates distribute work across a `ThreadPool` when `CONCURRENCY_ENABLED` is active.
- **Configurable pipelines** – The `Context` class and factory methods in [`src/transform/TransformFactory.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/transform/TransformFactory.hpp) and [`src/entropy/EntropyEncoderFactory.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/entropy/EntropyEncoderFactory.hpp) allow runtime selection of transforms and entropy codecs.
- **Seekable archives** – Block-aligned output enables `seek(bitPos)` for random access decompression, suitable for video archives and large datasets.
- **C interoperability** – The C API in `src/api/` provides opaque handles for embedding in other languages while preserving the full streaming workflow.

## Frequently Asked Questions

### What block size should I use for streaming compression with Kanzi?

Select a block size between 1 MiB and 8 MiB via `ctx.putInt("blockSize", 4*1024*1024)`. Larger blocks improve compression ratio but increase latency before data is written to the output stream, while smaller blocks enable finer-grained parallelism and faster flushing.

### Can I seek to arbitrary byte positions in a Kanzi-compressed stream?

No, seeking is supported only on block boundaries. The `seek(bitPos)` method in `CompressedInputStream` resets the decoding pipeline to specific block starts, making Kanzi suitable for applications requiring random access to large files but not byte-level seeking within compressed data.

### How does Kanzi verify data integrity during streaming?

Each block optionally includes a checksum computed via XXHash32 or XXHash64 (implemented in [`src/util/XXHash.hpp`](https://github.com/flanglet/kanzi-cpp/blob/main/src/util/XXHash.hpp)). When enabled in the `Context`, `EncodingTask` computes checksums during compression and `DecodingTask` verifies them during decompression, ensuring block-level integrity without buffering the entire stream.

### Is the Kanzi streaming API thread-safe for concurrent stream processing?

Individual `CompressedOutputStream` and `CompressedInputStream` instances are not thread-safe for concurrent calls, but the internal `ThreadPool` safely handles parallel block processing across multiple tasks. For concurrent compression of multiple independent streams, create separate stream instances each with their own `Context` configuration.