# Streaming Writes

{class}`trx::TrxStream` is an append-only writer for cases where the total
streamline count is not known ahead of time. It writes to temporary files and
finalizes to a standard TRX archive or directory when complete.

```{eval-rst}
.. note::

   ``TrxStream`` is **not** thread-safe for concurrent writes. Use a single
   writer thread (or the main thread) to append to the stream, while other
   threads generate streamlines and deliver them via a queue.
```

## Single-threaded streaming

```cpp
#include <trx/trx.h>

trx::TrxStream stream("float16");

for (/* each generated streamline */) {
  std::vector<std::array<float, 3>> points = /* ... */;
  stream.push_streamline(points);
}

stream.finalize<Eigen::half>("tracks.trx", ZIP_CM_STORE);
```

## Multi-threaded producer / single writer

Worker threads generate streamlines and push batches into a queue. A
dedicated writer thread owns the `TrxStream` and consumes from the queue.

```cpp
#include <trx/trx.h>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <thread>

struct Batch {
  std::vector<std::vector<std::array<float, 3>>> streamlines;
};

std::mutex mtx;
std::condition_variable cv;
std::queue<Batch> q;
bool done = false;

// Producer: generates streamlines, pushes batches into the queue.
auto producer = [&]() {
  Batch batch;
  batch.streamlines.reserve(1000);
  for (int i = 0; i < 1000; ++i) {
    batch.streamlines.push_back(/* generate points */);
  }
  {
    std::lock_guard<std::mutex> lock(mtx);
    q.push(std::move(batch));
  }
  cv.notify_one();
};

// Writer: owns TrxStream, appends batches from the queue.
trx::TrxStream stream("float16");
auto writer = [&]() {
  for (;;) {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, [&] { return done || !q.empty(); });
    if (q.empty() && done) return;
    Batch batch = std::move(q.front());
    q.pop();
    lock.unlock();
    for (const auto& pts : batch.streamlines) {
      stream.push_streamline(pts);
    }
  }
};

std::thread writer_thread(writer);
std::thread t1(producer), t2(producer);
t1.join(); t2.join();
{ std::lock_guard<std::mutex> lock(mtx); done = true; }
cv.notify_all();
writer_thread.join();

stream.finalize<Eigen::half>("tracks.trx", ZIP_CM_STORE);
```

## MRtrix-style write kernel

MRtrix3 uses a multi-threaded producer stage and a single-writer kernel to
serialize output. The same pattern works with TRX by encapsulating the
`TrxStream` inside a kernel object:

```cpp
struct TrxWriteKernel {
  explicit TrxWriteKernel(const std::string& path)
      : stream("float16"), out_path(path) {}

  void operator()(const std::vector<std::vector<std::array<float, 3>>>& batch) {
    for (const auto& pts : batch) {
      stream.push_streamline(pts);
    }
  }

  void finalize() {
    stream.finalize<Eigen::half>(out_path, ZIP_CM_STORE);
  }

private:
  trx::TrxStream stream;
  std::string out_path;
};
```

The key rule is: **only the writer thread touches \`\`TrxStream\`\`**, while
worker threads only generate streamlines.

## Process-based sharding

For very large tractograms generated in parallel processes, each process can
write to a shard directory and a parent process merges the shards afterward.

`TrxStream` provides two finalization methods for directory output:

- `finalize_directory()` — removes any existing directory before writing.
  Safe for single-process workflows where you control the full lifecycle.
- `finalize_directory_persistent()` — does **not** remove existing
  directories. Required when a parent process pre-creates the output
  directory.

Recommended multiprocess pattern:

1. **Parent** pre-creates shard directories.
2. Each **child** calls `finalize_directory_persistent()` after appending
   all streamlines.
3. Child writes a sentinel file (e.g., `SHARD_OK`) to signal completion.
4. **Parent** waits for all sentinels, then merges shards.

```cpp
// Parent: pre-create shard directories
for (size_t i = 0; i < num_shards; ++i) {
  std::filesystem::create_directories("shards/shard_" + std::to_string(i));
}

// Child: write shard and signal completion
trx::TrxStream stream("float16");
// ... push_streamline calls ...
stream.finalize_directory_persistent("/path/to/shards/shard_0");
std::ofstream ok("/path/to/shards/shard_0/SHARD_OK");
ok << "ok\n";

// Parent: merge shards after all SHARD_OK files are present.
// See bench/bench_trx_stream.cpp for a reference merge implementation.
```

```{eval-rst}
.. note::
   Use ``finalize_directory()`` for single-process writes where you want a
   clean output state. Use ``finalize_directory_persistent()`` for
   multiprocess workflows to avoid removing directories that may be checked
   for existence by other processes.
```