Reader and Writer - chung-leong/zigar GitHub Wiki

The std.io.AnyReader and std.io.AnyWriter interface structs provide read/write access to data streams. Readers and writers of web streams created in JavaScript can be passed to Zig function accepting these structs as arguments. The converse is not possible. JavaScript cannot directly use Zig readers and writers.

Passing reader into Zig

Zig code running in the main thread cannot access JavaScript streams, since that would result in a deadlock. Only code running in work threads can do so:

const std = @import("std");

const zigar = @import("zigar");

var gpa = std.heap.GeneralPurposeAllocator(.{}){};
var work_queue: zigar.thread.WorkQueue(thread_ns) = .{};

pub fn startup() !void {
    try work_queue.init(.{ .allocator = gpa.allocator() });
}

pub fn shutdown(promise: zigar.function.Promise(void)) void {
    work_queue.deinitAsync(promise);
}

pub fn print(
    reader: std.io.AnyReader,
    promise: zigar.function.PromiseOf(thread_ns.print),
) !void {
    try work_queue.push(thread_ns.print, .{reader}, promise);
}

const thread_ns = struct {
    pub fn print(reader: std.io.AnyReader) !void {
        const stdout = std.io.getStdOut();
        var fifo: std.fifo.LinearFifo(u8, .{ .Static = 4096 }) = .init();
        try fifo.pump(reader, stdout);
    }
};

import { print, shutdown, startup } from './reader-example-1.zig';

startup() 
try {
  const response = await fetch('https://raw.githubusercontent.com/chung-leong/zigar/refs/heads/main/zigar-compiler/test/integration/stream-handling/data/test.txt');
  const reader = response.body.getReader();
  await print(reader);
} finally {
  await shutdown();
}

Four score and seven years ago our fathers brought forth on this continent, a new nation, conceived in Liberty, and dedicated to the proposition that all men are created equal.

Now we are engaged in a great civil war, testing whether that nation, or any nation so conceived and so dedicated, can long endure. We are met on a great battle-field of that war. We have come to dedicate a portion of that field, as a final resting place for those who here gave their lives that that nation might live. It is altogether fitting and proper that we should do this.

But, in a larger sense, we can not dedicate—we can not consecrate—we can not hallow—this ground. The brave men, living and dead, who struggled here, have consecrated it, far above our poor power to add or detract. The world will little note, nor long remember what we say here, but it can never forget what they did here. It is for us the living, rather, to be dedicated here to the unfinished work which they who fought here have thus far so nobly advanced. It is rather for us to be here dedicated to the great task remaining before us—that from these honored dead we take increased devotion to that cause for which they gave the last full measure of devotion—that we here highly resolve that these dead shall not have died in vain—that this nation, under God, shall have a new birth of freedom—and that government of the people, by the people, for the people, shall not perish from the earth.

std.fifo.LinearFifo.pump() is a convenient function for moving data from a reader to a writer. The fact that it's not in the std.io can make it difficult to find.

Reading from a JavaScript stream a few bytes at a time is very inefficient, as the operation involves the JavaScript event loop. The Zigar runtime actively monitors for this condition. If it detects that less than 8 bytes are transferred per call after 100 calls, it will stop the operation. The sample code above would fail with the followining error if the buffer length is shrunk to 1:

Error: Inefficient read access. Each call is only reading 1 byte. Please use std.io.BufferedReader.

Passing writer into Zig

Writers work in a similiar fashion:

const std = @import("std");

const zigar = @import("zigar");

var gpa = std.heap.GeneralPurposeAllocator(.{}){};
var work_queue: zigar.thread.WorkQueue(thread_ns) = .{};

pub fn startup() !void {
    try work_queue.init(.{ .allocator = gpa.allocator() });
}

pub fn shutdown(promise: zigar.function.Promise(void)) void {
    work_queue.deinitAsync(promise);
}

pub fn save(
    writer: std.io.AnyWriter,
    promise: zigar.function.PromiseOf(thread_ns.save),
) !void {
    try work_queue.push(thread_ns.save, .{writer}, promise);
}

const thread_ns = struct {
    pub fn save(writer: std.io.AnyWriter) !void {
        var buffer: [128]u8 = undefined;
        for (&buffer, 0..) |*ptr, i| ptr.* = @intCast(i);
        _ = try writer.write(&buffer);
    }
};

import { save, shutdown, startup } from './writer-example-1.zig';

startup() 
try {
  const stream = new WritableStream({
    write(buf) {
      console.log(buf);
    }
  });
  const writer = stream.getWriter();
  await save(writer);
} finally {
  await shutdown();
}

Uint8Array(128) [
   0,  1,  2,  3,  4,  5,  6,  7,  8,  9, 10, 11,
  12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23,
  24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35,
  36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47,
  48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59,
  60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71,
  72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83,
  84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95,
  96, 97, 98, 99,
  ... 28 more items
]

Using TransformStream

While you cannot make use of a Zig reader in JavaScript, you can create an identity TransformStream, pass its writable end to Zig and use std.fifo.LinearFifo.pump() to pump the data to its readable end:

const std = @import("std");

const zigar = @import("zigar");

var gpa = std.heap.GeneralPurposeAllocator(.{}){};
var work_queue: zigar.thread.WorkQueue(thread_ns) = .{};

pub fn startup() !void {
    try work_queue.init(.{ .allocator = gpa.allocator() });
}

pub fn shutdown(promise: zigar.function.Promise(void)) void {
    work_queue.deinitAsync(promise);
}

pub fn decompress(
    reader: std.io.AnyReader,
    writer: std.io.AnyWriter,
    promise: zigar.function.PromiseOf(thread_ns.decompress),
) !void {
    try work_queue.push(thread_ns.decompress, .{ reader, writer }, promise);
}

const thread_ns = struct {
    pub fn decompress(reader: std.io.AnyReader, writer: std.io.AnyWriter) !void {
        var buffer = std.io.bufferedReader(reader);
        var xz = try std.compress.xz.decompress(gpa.allocator(), buffer.reader());
        defer xz.deinit();
        var fifo: std.fifo.LinearFifo(u8, .{ .Static = 4096 }) = .init();
        try fifo.pump(xz.reader(), writer);
    }
};

import { decompress, shutdown, startup } from './reader-example-2.zig';

startup() 
try {
  const response = await fetch('https://github.com/chung-leong/zigar/raw/refs/heads/main/zigar-compiler/test/integration/stream-handling/data/test.txt.xz');
  const reader1 = response.body.getReader();
  const transform = new TransformStream(undefined, { highWaterMark: 1024 * 4 });
  const writer = transform.writable.getWriter();
  decompress(reader1, writer).then(() => writer.close());
  const reader2 = transform.readable.getReader();
  const decoder = new TextDecoder();
  while (true) {
    const { value, done } = await reader2.read();
    if (value) {
      const text = decoder.decode(value);
      console.log(text);
    }
    if (done) break;
  }
} finally {
  await shutdown();
}