Add ReadableStream support to k6

examples/experimental/streams.js

@@ -0,0 +1,34 @@
+import { ReadableStream } from 'k6/experimental/streams'
+import { setTimeout } from 'k6/timers'
+
+function numbersStream() {
+	let currentNumber = 0
+
+	return new ReadableStream({
+		start(controller) {
+			const fn = () => {
+				if (currentNumber < 5) {
+					controller.enqueue(++currentNumber)
+					setTimeout(fn, 1000)
+					return;
+				}
+
+				controller.close()
+			}
+			setTimeout(fn, 1000)
+		},
+	})
+}
+
+export default async function () {
+	const stream = numbersStream()
+	const reader = stream.getReader()
+
+	while (true) {
+		const { done, value } = await reader.read()
+		if (done) break
+		console.log(`received number ${value} from stream`)
+	}
+
+	console.log('we are done')
+}

js/jsmodules.go

@@ -14,6 +14,7 @@ import (
 	"go.k6.io/k6/js/modules/k6/encoding"
 	"go.k6.io/k6/js/modules/k6/execution"
 	"go.k6.io/k6/js/modules/k6/experimental/fs"
+	"go.k6.io/k6/js/modules/k6/experimental/streams"
 	"go.k6.io/k6/js/modules/k6/experimental/tracing"
 	"go.k6.io/k6/js/modules/k6/grpc"
 	"go.k6.io/k6/js/modules/k6/html"
@@ -38,6 +39,7 @@ func getInternalJSModules() map[string]interface{} {
 		"k6/timers":                  timers.New(),
 		"k6/execution":               execution.New(),
 		"k6/experimental/redis":      redis.New(),
+		"k6/experimental/streams":    streams.New(),
 		"k6/experimental/webcrypto":  webcrypto.New(),
 		"k6/experimental/websockets": &expws.RootModule{},
 		"k6/experimental/timers": newWarnExperimentalModule(timers.New(),

js/modules/k6/experimental/streams/errors.go

@@ -0,0 +1,64 @@
+package streams
+
+import "github.com/dop251/goja"
+
+func newError(k errorKind, message string) *streamError {
+	return &streamError{
+		Name:    k.String(),
+		Message: message,
+		kind:    k,
+	}
+}
+
+//go:generate enumer -type=errorKind -output errors_gen.go
+type errorKind uint8
+
+const (
+	// TypeError is thrown when an argument is not of an expected type
+	TypeError errorKind = iota + 1
+
+	// RangeError is thrown when an argument is not within the expected range
+	RangeError
+
+	// RuntimeError is thrown when an error occurs that was caused by the JS runtime
+	// and is not likely caused by the user, but rather the implementation.
+	RuntimeError
+
+	// AssertionError is thrown when an assertion fails
+	AssertionError
+)
+
+type streamError struct {
+	// Name contains the name of the error
+	Name string `json:"name"`
+
+	// Message contains the error message
+	Message string `json:"message"`
+
+	// kind contains the kind of error
+	kind errorKind
+}
+
+// Ensure that the fsError type implements the Go `error` interface
+var _ error = (*streamError)(nil)
+
+func (e *streamError) Error() string {
+	return e.Name + ":" + e.Message
+}
+
+func throw(rt *goja.Runtime, err any) {
+	panic(errToObj(rt, err))
+}
+
+func errToObj(rt *goja.Runtime, err any) goja.Value {
+	// Undefined remains undefined.
+	if goja.IsUndefined(rt.ToValue(err)) {
+		return rt.ToValue(err)
+	}
+
+	if e, ok := err.(*goja.Exception); ok {
+		return e.Value().ToObject(rt)
+	}
+
+	return rt.ToValue(err).ToObject(rt)
+}

js/modules/k6/experimental/streams/goja.go

@@ -0,0 +1,107 @@
+package streams
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/dop251/goja"
+	"go.k6.io/k6/js/common"
+	"go.k6.io/k6/js/modules"
+)
+
+// newResolvedPromise instantiates a new resolved promise.
+func newResolvedPromise(vu modules.VU, with goja.Value) *goja.Promise {
+	promise, resolve, _ := vu.Runtime().NewPromise()
+	resolve(with)
+	return promise
+}
+
+// newRejectedPromise instantiates a new rejected promise.
+func newRejectedPromise(vu modules.VU, with any) *goja.Promise {
+	promise, _, reject := vu.Runtime().NewPromise()
+	reject(with)
+	return promise
+}
+
+// promiseThen facilitates instantiating a new promise and defining callbacks for to be executed
+// on fulfillment as well as rejection, directly from Go.
+func promiseThen(
+	rt *goja.Runtime,
+	promise *goja.Promise,
+	onFulfilled, onRejected func(goja.Value),
+) (*goja.Promise, error) {
+	val, err := rt.RunString(
+		`(function(promise, onFulfilled, onRejected) { return promise.then(onFulfilled, onRejected) })`)
+	if err != nil {
+		return nil, newError(RuntimeError, "unable to initialize promiseThen internal helper function")
+	}
+
+	cal, ok := goja.AssertFunction(val)
+	if !ok {
+		return nil, newError(RuntimeError, "the internal promiseThen helper is not a function")
+	}
+
+	if onRejected == nil {
+		val, err = cal(goja.Undefined(), rt.ToValue(promise), rt.ToValue(onFulfilled))
+	} else {
+		val, err = cal(goja.Undefined(), rt.ToValue(promise), rt.ToValue(onFulfilled), rt.ToValue(onRejected))
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	newPromise, ok := val.Export().(*goja.Promise)
+	if !ok {
+		return nil, newError(RuntimeError, "unable to cast the internal promiseThen helper's return value to a promise")
+	}
+
+	return newPromise, nil
+}
+
+// isNumber returns true if the given goja value holds a number
+func isNumber(value goja.Value) bool {
+	_, isFloat := value.Export().(float64)
+	_, isInt := value.Export().(int64)
+
+	return isFloat || isInt
+}
+
+// isNonNegativeNumber implements the [IsNonNegativeNumber] algorithm.
+//
+// [IsNonNegativeNumber]: https://streams.spec.whatwg.org/#is-non-negative-number
+func isNonNegativeNumber(value goja.Value) bool {
+	if common.IsNullish(value) {
+		return false
+	}
+
+	if !isNumber(value) {
+		return false
+	}
+
+	if value.ToFloat() < 0 || value.ToInteger() < 0 {
+		return false
+	}
+
+	return true
+}
+
+// setReadOnlyPropertyOf sets a read-only property on the given [goja.Object].
+func setReadOnlyPropertyOf(obj *goja.Object, name string, value goja.Value) error {
+	err := obj.DefineDataProperty(name,
+		value,
+		goja.FLAG_FALSE,
+		goja.FLAG_FALSE,
+		goja.FLAG_TRUE,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to define %s read-only property on TextEncoder object; reason: %w", name, err)
+	}
+
+	return nil
+}
+
+// isObject determines whether the given [goja.Value] is a [goja.Object] or not.
+func isObject(val goja.Value) bool {
+	return val != nil && val.ExportType() != nil && val.ExportType().Kind() == reflect.Map
+}