Add node.js compat streams split2 example

samples/nodejs-compat-streams-split2/README.md

@@ -0,0 +1,35 @@
+# Node.js Compat Example
+
+To run the example on http://localhost:8080
+
+```sh
+$ bazel-bin/src/workerd/server/workerd serve --experimental $(realpath samples/nodejs-compat-streams-split2/config.capnp)
+```
+
+To run using bazel
+
+```sh
+$ bazel run //src/workerd/server:workerd -- serve $(realpath samples/nodejs-compat-streams-split2/config.capnp)
+```
+
+To create a standalone binary that can be run:
+
+```sh
+$ ./workerd compile config.capnp > nodejs-compat
+
+$ ./nodejs-compat
+```
+
+To test:
+
+```sh
+% curl http://localhost:8080
+hello
+from
+the
+wonderful
+world
+of
+node.js
+streams!
+```

samples/nodejs-compat-streams-split2/config.capnp

@@ -0,0 +1,36 @@
+# Imports the base schema for workerd configuration files.
+
+# Refer to the comments in /src/workerd/server/workerd.capnp for more details.
+
+using Workerd = import "/workerd/workerd.capnp";
+
+# A constant of type Workerd.Config defines the top-level configuration for an
+# instance of the workerd runtime. A single config file can contain multiple
+# Workerd.Config definitions and must have at least one.
+const helloWorldExample :Workerd.Config = (
+
+  # Every workerd instance consists of a set of named services. A worker, for instance,
+  # is a type of service. Other types of services can include external servers, the
+  # ability to talk to a network, or accessing a disk directory. Here we create a single
+  # worker service. The configuration details for the worker are defined below.
+  services = [ (name = "main", worker = .helloWorld) ],
+
+  # Every configuration defines the one or more sockets on which the server will listene.
+  # Here, we create a single socket that will listen on localhost port 8080, and will
+  # dispatch to the "main" service that we defined above.
+  sockets = [ ( name = "http", address = "*:8080", http = (), service = "main" ) ]
+);
+
+# The definition of the actual helloWorld worker exposed using the "main" service.
+# In this example the worker is implemented as a single simple script (see worker.js).
+# The compatibilityDate is required. For more details on compatibility dates see:
+#   https://developers.cloudflare.com/workers/platform/compatibility-dates/
+
+const helloWorld :Workerd.Worker = (
+  modules = [
+    (name = "worker", esModule = embed "worker.js"),
+    (name = "split2", nodeJsCompatModule = embed "split2.js")
+  ],
+  compatibilityDate = "2023-03-01",
+  compatibilityFlags = ["nodejs_compat", "experimental"]
+);

samples/nodejs-compat-streams-split2/split2.js

@@ -0,0 +1,143 @@
+// https://www.npmjs.com/package/split2
+
+/*
+Copyright (c) 2014-2021, Matteo Collina <hello@matteocollina.com>
+
+Permission to use, copy, modify, and/or distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR
+IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+
+'use strict'
+
+const { Transform } = require('stream')
+const { StringDecoder } = require('string_decoder')
+const kLast = Symbol('last')
+const kDecoder = Symbol('decoder')
+
+function transform (chunk, enc, cb) {
+  let list
+  if (this.overflow) { // Line buffer is full. Skip to start of next line.
+    const buf = this[kDecoder].write(chunk)
+    list = buf.split(this.matcher)
+
+    if (list.length === 1) return cb() // Line ending not found. Discard entire chunk.
+
+    // Line ending found. Discard trailing fragment of previous line and reset overflow state.
+    list.shift()
+    this.overflow = false
+  } else {
+    this[kLast] += this[kDecoder].write(chunk)
+    list = this[kLast].split(this.matcher)
+  }
+
+  this[kLast] = list.pop()
+
+  for (let i = 0; i < list.length; i++) {
+    try {
+      push(this, this.mapper(list[i]))
+    } catch (error) {
+      return cb(error)
+    }
+  }
+
+  this.overflow = this[kLast].length > this.maxLength
+  if (this.overflow && !this.skipOverflow) {
+    cb(new Error('maximum buffer reached'))
+    return
+  }
+
+  cb()
+}
+
+function flush (cb) {
+  // forward any gibberish left in there
+  this[kLast] += this[kDecoder].end()
+
+  if (this[kLast]) {
+    try {
+      push(this, this.mapper(this[kLast]))
+    } catch (error) {
+      return cb(error)
+    }
+  }
+
+  cb()
+}
+
+function push (self, val) {
+  if (val !== undefined) {
+    self.push(val)
+  }
+}
+
+function noop (incoming) {
+  return incoming
+}
+
+function split (matcher, mapper, options) {
+  // Set defaults for any arguments not supplied.
+  matcher = matcher || /\r?\n/
+  mapper = mapper || noop
+  options = options || {}
+
+  // Test arguments explicitly.
+  switch (arguments.length) {
+    case 1:
+      // If mapper is only argument.
+      if (typeof matcher === 'function') {
+        mapper = matcher
+        matcher = /\r?\n/
+      // If options is only argument.
+      } else if (typeof matcher === 'object' && !(matcher instanceof RegExp) && !matcher[Symbol.split]) {
+        options = matcher
+        matcher = /\r?\n/
+      }
+      break
+
+    case 2:
+      // If mapper and options are arguments.
+      if (typeof matcher === 'function') {
+        options = mapper
+        mapper = matcher
+        matcher = /\r?\n/
+      // If matcher and options are arguments.
+      } else if (typeof mapper === 'object') {
+        options = mapper
+        mapper = noop
+      }
+  }
+
+  options = Object.assign({}, options)
+  options.autoDestroy = true
+  options.transform = transform
+  options.flush = flush
+  options.readableObjectMode = true
+
+  const stream = new Transform(options)
+
+  stream[kLast] = ''
+  stream[kDecoder] = new StringDecoder('utf8')
+  stream.matcher = matcher
+  stream.mapper = mapper
+  stream.maxLength = options.maxLength
+  stream.skipOverflow = options.skipOverflow || false
+  stream.overflow = false
+  stream._destroy = function (err, cb) {
+    // Weird Node v12 bug that we need to work around
+    this._writableState.errorEmitted = false
+    cb(err)
+  }
+
+  return stream
+}
+
+module.exports = split

samples/nodejs-compat-streams-split2/worker.js

@@ -0,0 +1,31 @@
+import {
+  PassThrough,
+  Transform,
+  Readable,
+} from 'node:stream';
+
+import { default as split2 } from 'split2';
+
+const enc = new TextEncoder();
+
+export default {
+  async fetch() {
+    const pt = new PassThrough();
+
+    // split2 will remove the new lines from the single input stream,
+    // pushing each individual line as a separate chunk. We use this
+    // transform to add the new lines back in just to show the effect
+    // in the output.
+    const lb = new Transform({
+      transform(chunk, encoding, callback) {
+        callback(null, enc.encode(chunk + '\n'));
+      }
+    });
+
+    const readable = pt.pipe(split2()).pipe(lb);
+
+    pt.end('hello\nfrom\nthe\nwonderful\nworld\nof\nnode.js\nstreams!');
+
+    return new Response(Readable.toWeb(readable));
+  }
+};