Streaming

service StreamService {
  rpc ListItems   (ListItemsRequest)    returns (stream Item);           // server streaming
  rpc UploadItems (stream UploadChunk)  returns (UploadSummary);        // client streaming
  rpc Sync        (stream SyncRequest)  returns (stream SyncResponse);  // bidi streaming
}

function listItems(call) {
  // call.request  — the single incoming request message
  // call.write()  — send one item to the client
  // call.end()    — signal end of stream (required)
  // call.cancelled — boolean; check before expensive ops
  // call.on('cancelled', cb) — fires if client disconnects

  call.on('cancelled', () => {
    console.log('client cancelled');
    // clean up database cursor, etc.
  });

  for (const item of db.getAll()) {
    if (call.cancelled) return;
    call.write({ item });
  }
  call.end();
}

// Async generator pattern
async function listItemsAsync(call) {
  for await (const item of db.streamAll()) {
    if (call.cancelled) return;
    call.write({ item });
  }
  call.end();
}

const stream = client.listItems({ filter: 'active' });

stream.on('data', (response) => {
  console.log('received:', response.item);
});

stream.on('end', () => {
  console.log('stream ended');
});

stream.on('error', (err) => {
  if (err.code === grpc.status.CANCELLED) return;   // we cancelled it
  console.error('stream error', err.code, err.message);
});

stream.on('status', (status) => {
  console.log('final status:', status.code);
  console.log('trailing metadata:', status.metadata.getMap());
});

// Cancel the stream early
setTimeout(() => stream.cancel(), 2000);

async function consumeStream() {
  const stream = client.listItems({ filter: 'active' });
  try {
    for await (const response of stream) {
      console.log(response.item);
    }
  } catch (err) {
    if (err.code !== grpc.status.CANCELLED) throw err;
  }
}

const call = client.uploadItems((err, response) => {
  // called once when the server responds
  if (err) return console.error(err);
  console.log('uploaded', response.count, 'items');
});

// Optionally attach metadata listeners
call.on('metadata', (meta) => console.log('headers:', meta.getMap()));

// Write chunks
for (const chunk of chunks) {
  call.write({ data: chunk.data, filename: chunk.name });
}

// Signal end of client stream — server will now respond
call.end();

// Cancel mid-upload
// call.cancel()

function uploadItems(call, callback) {
  const results = [];

  call.on('data', (chunk) => {
    // process each client chunk
    results.push(processChunk(chunk));
  });

  call.on('end', () => {
    // client done sending — send the single response
    callback(null, { count: results.length, ids: results });
  });

  call.on('error', (err) => {
    console.error('client stream error:', err);
  });
}

function streamToPromise(call, chunks) {
  return new Promise((resolve, reject) => {
    call.on('error', reject);
    for (const chunk of chunks) call.write(chunk);
    call.end();
    // response comes via callback passed at call creation
  });
}

function sync(call) {
  // call is a duplex stream: readable (client messages) + writable (server messages)

  call.on('data', (request) => {
    // process incoming message
    const response = processRequest(request);
    call.write(response);   // respond to this message
  });

  call.on('end', () => {
    // client closed its write side
    call.end();             // close server write side
  });

  call.on('error', (err) => {
    console.error('bidi stream error', err);
  });

  call.on('cancelled', () => {
    console.log('client cancelled');
  });
}

const call = client.sync();

call.on('data', (response) => {
  console.log('server says:', response);
});

call.on('end', () => {
  console.log('server closed stream');
});

call.on('error', (err) => {
  if (err.code !== grpc.status.CANCELLED) console.error(err);
});

// Send messages
call.write({ action: 'push', payload: '...' });
call.write({ action: 'push', payload: '...' });

// Signal we're done sending
call.end();

// Node.js Duplex streams are async iterable (for await...of reads)
async function runSync(messages) {
  const call = client.sync();

  // Send in background
  (async () => {
    for (const msg of messages) {
      call.write(msg);
      await sleep(100);
    }
    call.end();
  })();

  // Receive
  for await (const response of call) {
    console.log('received', response);
  }
}

// call.write() returns false when the internal buffer is full
// Resume when 'drain' fires
function writeWithBackpressure(call, items, onDone) {
  let i = 0;

  function writeNext() {
    while (i < items.length) {
      const ok = call.write(items[i++]);
      if (!ok) {
        call.once('drain', writeNext);    // wait for drain
        return;
      }
    }
    call.end();
    onDone();
  }

  writeNext();
}

// Disable automatic flow control on readable (server/bidi streams)
// Use call.resume() to pull data manually
const stream = client.listItems({ filter: 'active' });
stream.pause();    // pause immediately

setTimeout(() => {
  stream.resume();    // start consuming after 1s delay
}, 1000);

// Server: send initial metadata before first write
function listItems(call) {
  const header = new grpc.Metadata();
  header.set('x-total-estimate', '1000');
  call.sendMetadata(header);    // fires before any call.write()

  for (const item of db.getAll()) {
    call.write({ item });
  }

  const trailer = new grpc.Metadata();
  trailer.set('x-items-sent', String(db.count()));
  call.end(trailer);            // trailing metadata in end()
}

// Client: read header and trailer
const stream = client.listItems({});
stream.on('metadata', (header) => console.log('header:', header.getMap()));
stream.on('status', (status) => console.log('trailer:', status.metadata.getMap()));

const deadline = new Date(Date.now() + 30_000);    // 30s budget
const stream = client.listItems({ filter: 'active' }, new grpc.Metadata(), { deadline });

stream.on('error', (err) => {
  if (err.code === grpc.status.DEADLINE_EXCEEDED) {
    console.error('stream timed out');
  }
});

// Pattern: treat errors as stream termination rather than throw
stream.on('error', (err) => {
  if (err.code === grpc.status.OK) return;    // grpc-js emits 'error' even on OK sometimes
  handleError(err);
});

const fs = require('fs');
const call = client.uploadItems((err, res) => console.log(res));

const file = fs.createReadStream('./data.bin', { highWaterMark: 64 * 1024 });
file.on('data', (chunk) => {
  call.write({ data: chunk });
});
file.on('end', () => call.end());
file.on('error', (err) => {
  call.cancel();
  console.error(err);
});