const std = @import("std"); const Api = @import("../../api/schema.zig").Api; const bun = @import("../../global.zig"); const RequestContext = @import("../../http.zig").RequestContext; const MimeType = @import("../../http.zig").MimeType; const ZigURL = @import("../../url.zig").URL; const HTTPClient = @import("http"); const NetworkThread = HTTPClient.NetworkThread; const AsyncIO = NetworkThread.AsyncIO; const JSC = @import("javascript_core"); const js = JSC.C; const Method = @import("../../http/method.zig").Method; const FetchHeaders = JSC.FetchHeaders; const ObjectPool = @import("../../pool.zig").ObjectPool; const SystemError = JSC.SystemError; const Output = @import("../../global.zig").Output; const MutableString = @import("../../global.zig").MutableString; const strings = @import("../../global.zig").strings; const string = @import("../../global.zig").string; const default_allocator = @import("../../global.zig").default_allocator; const FeatureFlags = @import("../../global.zig").FeatureFlags; const ArrayBuffer = @import("../base.zig").ArrayBuffer; const Properties = @import("../base.zig").Properties; const NewClass = @import("../base.zig").NewClass; const d = @import("../base.zig").d; const castObj = @import("../base.zig").castObj; const getAllocator = @import("../base.zig").getAllocator; const JSPrivateDataPtr = @import("../base.zig").JSPrivateDataPtr; const GetJSPrivateData = @import("../base.zig").GetJSPrivateData; const Environment = @import("../../env.zig"); const ZigString = JSC.ZigString; const IdentityContext = @import("../../identity_context.zig").IdentityContext; const JSInternalPromise = JSC.JSInternalPromise; const JSPromise = JSC.JSPromise; const JSValue = JSC.JSValue; const JSError = JSC.JSError; const JSGlobalObject = JSC.JSGlobalObject; const VirtualMachine = @import("../javascript.zig").VirtualMachine; const Task = JSC.Task; const JSPrinter = @import("../../js_printer.zig"); const picohttp = @import("picohttp"); const StringJoiner = @import("../../string_joiner.zig"); const uws = @import("uws"); const Blob = JSC.WebCore.Blob; const Response = JSC.WebCore.Response; const Request = JSC.WebCore.Request; pub const ReadableStream = struct { value: JSValue, ptr: Source, pub fn done(this: *const ReadableStream) void { this.value.unprotect(); } pub const Tag = enum(i32) { Invalid = -1, JavaScript = 0, Blob = 1, File = 2, HTTPRequest = 3, HTTPSRequest = 4, HTTPResponse = 5, HTTPSResponse = 6, }; pub const Source = union(Tag) { Invalid: void, JavaScript: void, Blob: *ByteBlobLoader, File: *FileBlobLoader, // HTTPRequest: *HTTPRequest, HTTPRequest: void, // HTTPSRequest: *HTTPSRequest, HTTPSRequest: void, // HTTPRequest: *HTTPRequest, HTTPResponse: void, // HTTPSRequest: *HTTPSRequest, HTTPSResponse: void, }; extern fn ReadableStreamTag__tagged(globalObject: *JSGlobalObject, possibleReadableStream: JSValue, ptr: *JSValue) Tag; extern fn ReadableStream__isDisturbed(possibleReadableStream: JSValue, globalObject: *JSGlobalObject) bool; extern fn ReadableStream__isLocked(possibleReadableStream: JSValue, globalObject: *JSGlobalObject) bool; extern fn ReadableStream__empty(*JSGlobalObject) JSC.JSValue; extern fn ReadableStream__fromBlob( *JSGlobalObject, store: *anyopaque, offset: usize, length: usize, ) JSC.JSValue; pub fn isDisturbed(this: *const ReadableStream, globalObject: *JSGlobalObject) bool { JSC.markBinding(); return ReadableStream__isDisturbed(this.value, globalObject); } pub fn isLocked(this: *const ReadableStream, globalObject: *JSGlobalObject) bool { JSC.markBinding(); return ReadableStream__isLocked(this.value, globalObject); } pub fn fromJS(value: JSValue, globalThis: *JSGlobalObject) ?ReadableStream { JSC.markBinding(); var ptr = JSValue.zero; return switch (ReadableStreamTag__tagged(globalThis, value, &ptr)) { .JavaScript => ReadableStream{ .value = value, .ptr = .{ .JavaScript = {}, }, }, .Blob => ReadableStream{ .value = value, .ptr = .{ .Blob = ptr.asPtr(ByteBlobLoader), }, }, .File => ReadableStream{ .value = value, .ptr = .{ .File = ptr.asPtr(FileBlobLoader), }, }, // .HTTPRequest => ReadableStream{ // .value = value, // .ptr = .{ // .HTTPRequest = ptr.asPtr(HTTPRequest), // }, // }, // .HTTPSRequest => ReadableStream{ // .value = value, // .ptr = .{ // .HTTPSRequest = ptr.asPtr(HTTPSRequest), // }, // }, else => null, }; } extern fn ZigGlobalObject__createNativeReadableStream(*JSGlobalObject, nativePtr: JSValue, nativeType: JSValue) JSValue; pub fn fromNative(globalThis: *JSGlobalObject, id: Tag, ptr: *anyopaque) JSC.JSValue { return ZigGlobalObject__createNativeReadableStream(globalThis, JSValue.fromPtr(ptr), JSValue.jsNumber(@enumToInt(id))); } pub fn fromBlob(globalThis: *JSGlobalObject, blob: *const Blob, recommended_chunk_size: Blob.SizeType) JSC.JSValue { if (comptime JSC.is_bindgen) unreachable; var store = blob.store orelse { return ReadableStream.empty(globalThis); }; switch (store.data) { .bytes => { var reader = bun.default_allocator.create(ByteBlobLoader.Source) catch unreachable; reader.* = .{ .context = undefined, }; reader.context.setup(blob, recommended_chunk_size); return reader.toJS(globalThis); }, .file => { var reader = bun.default_allocator.create(FileBlobLoader.Source) catch unreachable; reader.* = .{ .context = undefined, }; reader.context.setup(store, recommended_chunk_size); return reader.toJS(globalThis); }, } } pub fn empty(globalThis: *JSGlobalObject) JSC.JSValue { if (comptime JSC.is_bindgen) unreachable; return ReadableStream__empty(globalThis); } const Base = @import("../../ast/base.zig"); pub const StreamTag = enum(usize) { invalid = 0, _, pub fn init(filedes: JSC.Node.FileDescriptor) StreamTag { var bytes = [8]u8{ 1, 0, 0, 0, 0, 0, 0, 0 }; const filedes_ = @bitCast([8]u8, @as(usize, @truncate(u56, @intCast(usize, filedes)))); bytes[1..8].* = filedes_[0..7].*; return @intToEnum(StreamTag, @bitCast(u64, bytes)); } pub fn fd(this: StreamTag) JSC.Node.FileDescriptor { var bytes = @bitCast([8]u8, @enumToInt(this)); if (bytes[0] != 1) { return std.math.maxInt(JSC.Node.FileDescriptor); } var out: u64 = 0; @bitCast([8]u8, out)[0..7].* = bytes[1..8].*; return @intCast(JSC.Node.FileDescriptor, out); } }; }; pub const StreamStart = union(Tag) { empty: void, err: JSC.Node.Syscall.Error, chunk_size: Blob.SizeType, ArrayBufferSink: struct { chunk_size: Blob.SizeType, as_uint8array: bool, stream: bool, }, ready: void, pub const Tag = enum { empty, err, chunk_size, ArrayBufferSink, ready, }; pub fn toJS(this: StreamStart, globalThis: *JSGlobalObject) JSC.JSValue { switch (this) { .empty, .ready => { return JSC.JSValue.jsUndefined(); }, .chunk_size => |chunk| { return JSC.JSValue.jsNumber(@intCast(Blob.SizeType, chunk)); }, .err => |err| { globalThis.vm().throwError(globalThis, err.toJSC(globalThis)); return JSC.JSValue.jsUndefined(); }, else => { return JSC.JSValue.jsUndefined(); }, } } pub fn fromJS(globalThis: *JSGlobalObject, value: JSValue) StreamStart { if (value.isEmptyOrUndefinedOrNull() or !value.isObject()) { return .{ .empty = {} }; } if (value.get(globalThis, "chunkSize")) |chunkSize| { return .{ .chunk_size = @intCast(Blob.SizeType, @truncate(i52, chunkSize.toInt64())) }; } return .{ .empty = {} }; } pub fn fromJSWithTag( globalThis: *JSGlobalObject, value: JSValue, comptime tag: Tag, ) StreamStart { if (value.isEmptyOrUndefinedOrNull() or !value.isObject()) { return .{ .empty = {} }; } switch (comptime tag) { .ArrayBufferSink => { var as_uint8array = false; var stream = false; var chunk_size: JSC.WebCore.Blob.SizeType = 0; var empty = true; if (value.get(globalThis, "asUint8Array")) |as_array| { as_uint8array = as_array.toBoolean(); empty = false; } if (value.get(globalThis, "stream")) |as_array| { stream = as_array.toBoolean(); empty = false; } if (value.get(globalThis, "highWaterMark")) |chunkSize| { empty = false; chunk_size = @intCast(JSC.WebCore.Blob.SizeType, @maximum(0, @truncate(i51, chunkSize.toInt64()))); } if (!empty) { return .{ .ArrayBufferSink = .{ .chunk_size = chunk_size, .as_uint8array = as_uint8array, .stream = stream, }, }; } }, else => @compileError("Unuspported tag"), } return .{ .empty = {} }; } }; pub const StreamResult = union(Tag) { owned: bun.ByteList, owned_and_done: bun.ByteList, temporary_and_done: bun.ByteList, temporary: bun.ByteList, into_array: IntoArray, into_array_and_done: IntoArray, pending: *Pending, err: JSC.Node.Syscall.Error, done: void, pub const Tag = enum { owned, owned_and_done, temporary_and_done, temporary, into_array, into_array_and_done, pending, err, done, }; pub fn slice(this: *const StreamResult) []const u8 { return switch (this.*) { .owned => |owned| owned.slice(), .owned_and_done => |owned_and_done| owned_and_done.slice(), .temporary_and_done => |temporary_and_done| temporary_and_done.slice(), .temporary => |temporary| temporary.slice(), else => "", }; } pub const Writable = union(StreamResult.Tag) { pending: *Writable.Pending, err: JSC.Node.Syscall.Error, done: void, owned: Blob.SizeType, owned_and_done: Blob.SizeType, temporary_and_done: Blob.SizeType, temporary: Blob.SizeType, into_array: Blob.SizeType, into_array_and_done: Blob.SizeType, pub const Pending = struct { frame: anyframe, result: Writable, consumed: Blob.SizeType = 0, used: bool = false, }; pub fn toPromised(globalThis: *JSGlobalObject, promise: *JSPromise, pending: *Writable.Pending) void { var frame = bun.default_allocator.create(@Frame(Writable.toPromisedWrap)) catch unreachable; frame.* = async Writable.toPromisedWrap(globalThis, promise, pending); pending.frame = frame; } pub fn isDone(this: *const Writable) bool { return switch (this.*) { .owned_and_done, .temporary_and_done, .into_array_and_done, .done, .err => true, else => false, }; } fn toPromisedWrap(globalThis: *JSGlobalObject, promise: *JSPromise, pending: *Writable.Pending) void { suspend {} pending.used = true; const result: Writable = pending.result; switch (result) { .err => |err| { promise.reject(globalThis, err.toJSC(globalThis)); }, .done => { promise.resolve(globalThis, JSValue.jsBoolean(false)); }, else => { promise.resolve(globalThis, result.toJS(globalThis)); }, } } pub fn toJS(this: Writable, globalThis: *JSGlobalObject) JSValue { return switch (this) { .err => |err| JSC.JSPromise.rejectedPromise(globalThis, JSValue.c(err.toJS(globalThis.ref()))).asValue(globalThis), .owned => |len| JSC.JSValue.jsNumber(len), .owned_and_done => |len| JSC.JSValue.jsNumber(len), .temporary_and_done => |len| JSC.JSValue.jsNumber(len), .temporary => |len| JSC.JSValue.jsNumber(len), .into_array => |len| JSC.JSValue.jsNumber(len), .into_array_and_done => |len| JSC.JSValue.jsNumber(len), // false == controller.close() // undefined == noop, but we probably won't send it .done => JSC.JSValue.jsBoolean(true), .pending => |pending| brk: { var promise = JSC.JSPromise.create(globalThis); Writable.toPromised(globalThis, promise, pending); break :brk promise.asValue(globalThis); }, }; } }; pub const IntoArray = struct { value: JSValue = JSValue.zero, len: Blob.SizeType = std.math.maxInt(Blob.SizeType), }; pub const Pending = struct { frame: anyframe, result: StreamResult, used: bool = false, }; pub fn isDone(this: *const StreamResult) bool { return switch (this.*) { .owned_and_done, .temporary_and_done, .into_array_and_done, .done, .err => true, else => false, }; } fn toPromisedWrap(globalThis: *JSGlobalObject, promise: *JSPromise, pending: *Pending) void { suspend {} pending.used = true; const result: StreamResult = pending.result; switch (result) { .err => |err| { promise.reject(globalThis, err.toJSC(globalThis)); }, .done => { promise.resolve(globalThis, JSValue.jsBoolean(false)); }, else => { promise.resolve(globalThis, result.toJS(globalThis)); }, } } pub fn toPromised(globalThis: *JSGlobalObject, promise: *JSPromise, pending: *Pending) void { var frame = bun.default_allocator.create(@Frame(toPromisedWrap)) catch unreachable; frame.* = async toPromisedWrap(globalThis, promise, pending); pending.frame = frame; } pub fn toJS(this: *const StreamResult, globalThis: *JSGlobalObject) JSValue { switch (this.*) { .owned => |list| { return JSC.ArrayBuffer.fromBytes(list.slice(), .Uint8Array).toJS(globalThis.ref(), null); }, .owned_and_done => |list| { return JSC.ArrayBuffer.fromBytes(list.slice(), .Uint8Array).toJS(globalThis.ref(), null); }, .temporary => |temp| { var array = JSC.JSValue.createUninitializedUint8Array(globalThis, temp.len); var slice_ = array.asArrayBuffer(globalThis).?.slice(); @memcpy(slice_.ptr, temp.ptr, temp.len); return array; }, .temporary_and_done => |temp| { var array = JSC.JSValue.createUninitializedUint8Array(globalThis, temp.len); var slice_ = array.asArrayBuffer(globalThis).?.slice(); @memcpy(slice_.ptr, temp.ptr, temp.len); return array; }, .into_array => |array| { return JSC.JSValue.jsNumberFromInt64(array.len); }, .into_array_and_done => |array| { return JSC.JSValue.jsNumberFromInt64(array.len); }, .pending => |pending| { var promise = JSC.JSPromise.create(globalThis); toPromised(globalThis, promise, pending); return promise.asValue(globalThis); }, .err => |err| { return JSC.JSPromise.rejectedPromise(globalThis, JSValue.c(err.toJS(globalThis.ref()))).asValue(globalThis); }, // false == controller.close() // undefined == noop, but we probably won't send it .done => { return JSC.JSValue.jsBoolean(false); }, } } }; pub const Signal = struct { ptr: *anyopaque = @intToPtr(*anyopaque, 0xaaaaaaaa), vtable: VTable = VTable.Dead, pub fn isDead(this: Signal) bool { return this.ptr == @intToPtr(*anyopaque, 0xaaaaaaaa); } pub fn initWithType(comptime Type: type, handler: *Type) Sink { return .{ .ptr = handler, .vtable = VTable.wrap(Type), }; } pub fn init(handler: anytype) Signal { return initWithType(std.meta.Child(@TypeOf(handler)), handler); } pub fn close(this: Signal, err: ?JSC.Node.Syscall.Error) void { if (this.isDead()) return; this.vtable.close(this.ptr, err); } pub fn ready(this: Signal, amount: ?Blob.SizeType, offset: ?Blob.SizeType) void { if (this.isDead()) return; this.vtable.ready(this.ptr, amount, offset); } pub fn start(this: Signal) void { if (this.isDead()) return; this.vtable.start(this.ptr); } pub const VTable = struct { pub const OnCloseFn = fn (this: *anyopaque, err: ?JSC.Node.Syscall.Error) void; pub const OnReadyFn = fn (this: *anyopaque, amount: ?Blob.SizeType, offset: ?Blob.SizeType) void; pub const OnStartFn = fn (this: *anyopaque) void; close: OnCloseFn, ready: OnReadyFn, start: OnStartFn, const DeadFns = struct { pub fn close(_: *anyopaque, _: ?JSC.Node.Syscall.Error) void { unreachable; } pub fn ready(_: *anyopaque, _: ?Blob.SizeType, _: ?Blob.SizeType) void { unreachable; } pub fn start(_: *anyopaque) void { unreachable; } }; pub const Dead = VTable{ .close = DeadFns.close, .ready = DeadFns.ready, .start = DeadFns.start }; pub fn wrap( comptime Wrapped: type, ) VTable { const Functions = struct { fn onClose(this: *anyopaque, err: ?JSC.Node.Syscall.Error) void { Wrapped.close(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), err); } fn onReady(this: *anyopaque, amount: ?Blob.SizeType, offset: ?Blob.SizeType) void { Wrapped.ready(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), amount, offset); } fn onStart(this: *anyopaque) void { Wrapped.start(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this))); } }; return VTable{ .close = Functions.onClose, .ready = Functions.onReady, .start = Functions.onStart, }; } }; }; pub const Sink = struct { ptr: *anyopaque, vtable: VTable, status: Status = Status.closed, used: bool = false, pub const Status = enum { ready, closed, }; pub const Data = union(enum) { utf16: StreamResult, latin1: StreamResult, bytes: StreamResult, }; pub fn initWithType(comptime Type: type, handler: *Type) Sink { return .{ .ptr = handler, .vtable = VTable.wrap(Type), .status = .ready, .used = false, }; } pub fn init(handler: anytype) Sink { return initWithType(std.meta.Child(@TypeOf(handler)), handler); } pub const UTF8Fallback = struct { const stack_size = 1024; pub fn writeLatin1(comptime Ctx: type, ctx: *Ctx, input: StreamResult, comptime writeFn: anytype) StreamResult.Writable { var str = input.slice(); if (strings.isAllASCII(str)) { return writeFn( ctx, input, ); } if (stack_size >= str.len) { var buf: [stack_size]u8 = undefined; @memcpy(&buf, str.ptr, str.len); strings.replaceLatin1WithUTF8(buf[0..str.len]); if (input.isDone()) { const result = writeFn(ctx, .{ .temporary_and_done = bun.ByteList.init(buf[0..str.len]) }); return result; } else { const result = writeFn(ctx, .{ .temporary = bun.ByteList.init(buf[0..str.len]) }); return result; } } { var slice = bun.default_allocator.alloc(u8, str.len) catch return .{ .err = JSC.Node.Syscall.Error.oom }; @memcpy(slice.ptr, str.ptr, str.len); strings.replaceLatin1WithUTF8(slice[0..str.len]); if (input.isDone()) { return writeFn(ctx, .{ .owned_and_done = bun.ByteList.init(slice) }); } else { return writeFn(ctx, .{ .owned = bun.ByteList.init(slice) }); } } } pub fn writeUTF16(comptime Ctx: type, ctx: *Ctx, input: StreamResult, comptime writeFn: anytype) StreamResult.Writable { var str: []const u16 = std.mem.bytesAsSlice(u16, input.slice()); if (stack_size >= str.len * 2) { var buf: [stack_size]u8 = undefined; const copied = strings.copyUTF16IntoUTF8(&buf, []const u16, str); std.debug.assert(copied.written <= stack_size); std.debug.assert(copied.read <= stack_size); if (input.isDone()) { const result = writeFn(ctx, .{ .temporary_and_done = bun.ByteList.init(buf[0..copied.written]) }); return result; } else { const result = writeFn(ctx, .{ .temporary = bun.ByteList.init(buf[0..copied.written]) }); return result; } } { var allocated = strings.toUTF8Alloc(bun.default_allocator, str) catch return .{ .err = JSC.Node.Syscall.Error.oom }; if (input.isDone()) { return writeFn(ctx, .{ .owned_and_done = bun.ByteList.init(allocated) }); } else { return writeFn(ctx, .{ .owned = bun.ByteList.init(allocated) }); } } } }; pub const VTable = struct { pub const WriteUTF16Fn = fn (this: *anyopaque, data: StreamResult) StreamResult.Writable; pub const WriteUTF8Fn = fn (this: *anyopaque, data: StreamResult) StreamResult.Writable; pub const WriteLatin1Fn = fn (this: *anyopaque, data: StreamResult) StreamResult.Writable; pub const EndFn = fn (this: *anyopaque, err: ?JSC.Node.Syscall.Error) JSC.Node.Maybe(void); pub const ConnectFn = fn (this: *anyopaque, signal: Signal) JSC.Node.Maybe(void); connect: ConnectFn, write: WriteUTF8Fn, writeLatin1: WriteLatin1Fn, writeUTF16: WriteUTF16Fn, end: EndFn, pub fn wrap( comptime Wrapped: type, ) VTable { const Functions = struct { pub fn onWrite(this: *anyopaque, data: StreamResult) StreamResult.Writable { return Wrapped.write(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), data); } pub fn onConnect(this: *anyopaque, signal: Signal) JSC.Node.Maybe(void) { return Wrapped.connect(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), signal); } pub fn onWriteLatin1(this: *anyopaque, data: StreamResult) StreamResult.Writable { return Wrapped.writeLatin1(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), data); } pub fn onWriteUTF16(this: *anyopaque, data: StreamResult) StreamResult.Writable { return Wrapped.writeUTF16(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), data); } pub fn onEnd(this: *anyopaque, err: ?JSC.Node.Syscall.Error) JSC.Node.Maybe(void) { return Wrapped.end(@ptrCast(*Wrapped, @alignCast(std.meta.alignment(Wrapped), this)), err); } }; return VTable{ .write = Functions.onWrite, .writeLatin1 = Functions.onWriteLatin1, .writeUTF16 = Functions.onWriteUTF16, .end = Functions.onEnd, .connect = Functions.onConnect, }; } }; pub fn end(this: *Sink, err: ?JSC.Node.Syscall.Error) JSC.Node.Maybe(void) { if (this.status == .closed) { return .{ .result = {} }; } this.status = .closed; return this.vtable.end(this.ptr, err); } pub fn writeLatin1(this: *Sink, data: StreamResult) StreamResult.Writable { if (this.status == .closed) { return .{ .done = {} }; } const res = this.vtable.writeLatin1(this.ptr, data); this.status = if ((res.isDone()) or this.status == .closed) Status.closed else Status.ready; this.used = true; return res; } pub fn writeBytes(this: *Sink, data: StreamResult) StreamResult.Writable { if (this.status == .closed) { return .{ .done = {} }; } const res = this.vtable.write(this.ptr, data); this.status = if ((res.isDone()) or this.status == .closed) Status.closed else Status.ready; this.used = true; return res; } pub fn writeUTF16(this: *Sink, data: StreamResult) StreamResult.Writable { if (this.status == .closed) { return .{ .done = {} }; } const res = this.vtable.writeUTF16(this.ptr, data); this.status = if ((res.isDone()) or this.status == .closed) Status.closed else Status.ready; this.used = true; return res; } pub fn write(this: *Sink, data: Data) StreamResult.Writable { switch (data) { .utf16 => |str| { return this.writeUTF16(str); }, .latin1 => |str| { return this.writeLatin1(str); }, .bytes => |bytes| { return this.writeBytes(bytes); }, } } }; pub const ArrayBufferSink = struct { bytes: bun.ByteList, allocator: std.mem.Allocator, done: bool = false, signal: Signal = .{}, next: ?Sink = null, streaming: bool = false, as_uint8array: bool = false, pub fn connect(this: *ArrayBufferSink, signal: Signal) void { std.debug.assert(this.reader == null); this.signal = signal; } pub fn start(this: *ArrayBufferSink, stream_start: StreamStart) JSC.Node.Maybe(void) { var list = this.bytes.listManaged(this.allocator); list.clearAndFree(); switch (stream_start) { .ArrayBufferSink => |config| { if (config.chunk_size > 0) { list.ensureTotalCapacityPrecise(config.chunk_size) catch return .{ .err = JSC.Node.Syscall.Error.oom }; this.bytes.update(list); } this.as_uint8array = config.as_uint8array; this.streaming = config.stream; }, else => {}, } this.done = false; this.signal.start(); return .{ .result = {} }; } pub fn drain(_: *ArrayBufferSink) JSC.Node.Maybe(void) { return .{ .result = {} }; } pub fn drainFromJS(this: *ArrayBufferSink, globalThis: *JSGlobalObject) JSC.Node.Maybe(JSValue) { if (this.streaming) { const value: JSValue = switch (this.as_uint8array) { true => JSC.ArrayBuffer.create(globalThis, this.bytes.slice(), .Uint8Array), false => JSC.ArrayBuffer.create(globalThis, this.bytes.slice(), .ArrayBuffer), }; this.bytes.len = 0; return .{ .result = value }; } return .{ .result = JSValue.jsUndefined() }; } pub fn finalize(this: *ArrayBufferSink) void { if (this.bytes.len > 0) { this.bytes.listManaged(this.allocator).deinit(); this.bytes = bun.ByteList.init(""); this.done = true; } } pub fn init(allocator: std.mem.Allocator, next: ?Sink) !*ArrayBufferSink { var this = try allocator.create(ArrayBufferSink); this.* = ArrayBufferSink{ .bytes = bun.ByteList.init(&.{}), .allocator = allocator, .next = next, }; return this; } pub fn construct( this: *ArrayBufferSink, allocator: std.mem.Allocator, ) void { this.* = ArrayBufferSink{ .bytes = bun.ByteList.init(&.{}), .allocator = allocator, .next = null, }; } pub fn write(this: *@This(), data: StreamResult) StreamResult.Writable { if (this.next) |*next| { return next.writeBytes(data); } const len = this.bytes.write(this.allocator, data.slice()) catch { return .{ .err = JSC.Node.Syscall.Error.oom }; }; this.signal.ready(null, null); return .{ .owned = len }; } pub const writeBytes = write; pub fn writeLatin1(this: *@This(), data: StreamResult) StreamResult.Writable { if (this.next) |*next| { return next.writeLatin1(data); } const len = this.bytes.writeLatin1(this.allocator, data.slice()) catch { return .{ .err = JSC.Node.Syscall.Error.oom }; }; this.signal.ready(null, null); return .{ .owned = len }; } pub fn writeUTF16(this: *@This(), data: StreamResult) StreamResult.Writable { if (this.next) |*next| { return next.writeUTF16(data); } const len = this.bytes.writeUTF16(this.allocator, @ptrCast([*]const u16, @alignCast(@alignOf(u16), data.slice().ptr))[0..std.mem.bytesAsSlice(u16, data.slice()).len]) catch { return .{ .err = JSC.Node.Syscall.Error.oom }; }; this.signal.ready(null, null); return .{ .owned = len }; } pub fn end(this: *ArrayBufferSink, err: ?JSC.Node.Syscall.Error) JSC.Node.Maybe(void) { if (this.next) |*next| { return next.end(err); } this.signal.close(err); return .{ .result = {} }; } pub fn toJS(this: *ArrayBufferSink, globalThis: *JSGlobalObject, as_uint8array: bool) JSValue { if (this.streaming) { const value: JSValue = switch (as_uint8array) { true => JSC.ArrayBuffer.create(globalThis, this.bytes.slice(), .Uint8Array), false => JSC.ArrayBuffer.create(globalThis, this.bytes.slice(), .ArrayBuffer), }; this.bytes.len = 0; return value; } var list = this.bytes.listManaged(this.allocator); this.bytes = bun.ByteList.init(""); return ArrayBuffer.fromBytes( list.toOwnedSlice(), if (as_uint8array) .Uint8Array else .ArrayBuffer, ).toJS(globalThis, null); } pub fn endFromJS(this: *ArrayBufferSink, _: *JSGlobalObject) JSC.Node.Maybe(ArrayBuffer) { if (this.done) { return .{ .result = ArrayBuffer.fromBytes(&[_]u8{}, .ArrayBuffer) }; } std.debug.assert(this.next == null); var list = this.bytes.listManaged(this.allocator); this.bytes = bun.ByteList.init(""); this.done = true; this.signal.close(null); return .{ .result = ArrayBuffer.fromBytes( list.toOwnedSlice(), if (this.as_uint8array) .Uint8Array else .ArrayBuffer, ) }; } pub fn sink(this: *ArrayBufferSink) Sink { return Sink.init(this); } pub const JSSink = NewJSSink(@This(), "ArrayBufferSink"); }; pub fn NewJSSink(comptime SinkType: type, comptime name_: []const u8) type { return struct { sink: SinkType, const ThisSink = @This(); pub const shim = JSC.Shimmer("", std.mem.span(name_), @This()); pub const name = std.fmt.comptimePrint("{s}", .{std.mem.span(name_)}); pub fn createObject(globalThis: *JSGlobalObject, object: *anyopaque) callconv(.C) JSValue { JSC.markBinding(); return shim.cppFn("createObject", .{ globalThis, object }); } pub fn fromJS(globalThis: *JSGlobalObject, value: JSValue) ?*anyopaque { JSC.markBinding(); return shim.cppFn("fromJS", .{ globalThis, value }); } pub fn construct(globalThis: *JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var allocator = globalThis.bunVM().allocator; var this = allocator.create(ThisSink) catch { globalThis.vm().throwError(globalThis, JSC.Node.Syscall.Error.oom.toJSC( globalThis, )); return JSC.JSValue.jsUndefined(); }; this.sink.construct(allocator); return createObject(globalThis, this); } pub fn finalize(ptr: *anyopaque) callconv(.C) void { var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), ptr)); this.sink.finalize(); } pub fn write(globalThis: *JSGlobalObject, callframe: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), fromJS(globalThis, callframe.this()) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); })); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } const args = callframe.arguments(); if (args.len == 0 or args[0].isEmptyOrUndefinedOrNull() or args[0].isNumber()) { const err = JSC.toTypeError( if (args.len == 0) JSC.Node.ErrorCode.ERR_MISSING_ARGS else JSC.Node.ErrorCode.ERR_INVALID_ARG_TYPE, "write() expects a string, ArrayBufferView, or ArrayBuffer", .{}, globalThis, ); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } const arg = args[0]; if (arg.asArrayBuffer(globalThis)) |buffer| { const slice = buffer.slice(); if (slice.len == 0) { return JSC.JSValue.jsNumber(0); } return this.sink.writeBytes(.{ .temporary = bun.ByteList.init(slice) }).toJS(globalThis); } const str = arg.getZigString(globalThis); if (str.len == 0) { return JSC.JSValue.jsNumber(0); } if (str.is16Bit()) { return this.sink.writeUTF16(.{ .temporary = bun.ByteList.init(std.mem.sliceAsBytes(str.utf16SliceAligned())) }).toJS(globalThis); } return this.sink.writeLatin1(.{ .temporary = bun.ByteList.init(str.slice()) }).toJS(globalThis); } pub fn writeString(globalThis: *JSGlobalObject, callframe: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), fromJS(globalThis, callframe.this()) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); })); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } const args = callframe.arguments(); if (args.len == 0 or args[0].isEmptyOrUndefinedOrNull() or args[0].isNumber()) { const err = JSC.toTypeError( if (args.len == 0) JSC.Node.ErrorCode.ERR_MISSING_ARGS else JSC.Node.ErrorCode.ERR_INVALID_ARG_TYPE, "write() expects a string, ArrayBufferView, or ArrayBuffer", .{}, globalThis, ); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } const arg = args[0]; const str = arg.getZigString(globalThis); if (str.len == 0) { return JSC.JSValue.jsNumber(0); } if (str.is16Bit()) { return this.sink.writeUTF16(.{ .temporary = str.utf16SliceAligned() }).toJS(globalThis); } return this.sink.writeLatin1(.{ .temporary = str.slice() }).toJS(globalThis); } pub fn close(globalThis: *JSGlobalObject, sink_ptr: ?*anyopaque) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), sink_ptr) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); }); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } return this.sink.end(null).toJS(globalThis); } pub fn drain(globalThis: *JSGlobalObject, callframe: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), fromJS(globalThis, callframe.this()) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); })); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } if (comptime @hasDecl(SinkType, "drainFromJS")) { return this.sink.drainFromJS(globalThis).result; } return this.sink.drain().toJS(globalThis); } pub fn start(globalThis: *JSGlobalObject, callframe: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), fromJS(globalThis, callframe.this()) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); })); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } if (comptime @hasField(StreamStart, name_)) { return this.sink.start( if (callframe.argumentsCount() > 0) StreamStart.fromJSWithTag( globalThis, callframe.argument(0), comptime @field(StreamStart, name_), ) else StreamStart{ .empty = {} }, ).toJS(globalThis); } return this.sink.start( if (callframe.argumentsCount() > 0) StreamStart.fromJS(globalThis, callframe.argument(0)) else StreamStart{ .empty = {} }, ).toJS(globalThis); } pub fn end(globalThis: *JSGlobalObject, callframe: *JSC.CallFrame) callconv(.C) JSValue { JSC.markBinding(); var this = @ptrCast(*ThisSink, @alignCast(std.meta.alignment(ThisSink), fromJS(globalThis, callframe.this()) orelse { const err = JSC.toTypeError(JSC.Node.ErrorCode.ERR_INVALID_THIS, "Expected Sink", .{}, globalThis); globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); })); if (comptime @hasDecl(SinkType, "getPendingError")) { if (this.sink.getPendingError()) |err| { globalThis.vm().throwError(globalThis, err); return JSC.JSValue.jsUndefined(); } } return this.sink.endFromJS(globalThis).toJS(globalThis); } pub const Export = shim.exportFunctions(.{ .@"finalize" = finalize, .@"write" = write, .@"close" = close, .@"drain" = drain, .@"start" = start, .@"end" = end, .@"construct" = construct, }); comptime { if (!JSC.is_bindgen) { @export(finalize, .{ .name = Export[0].symbol_name }); @export(write, .{ .name = Export[1].symbol_name }); @export(close, .{ .name = Export[2].symbol_name }); @export(drain, .{ .name = Export[3].symbol_name }); @export(start, .{ .name = Export[4].symbol_name }); @export(end, .{ .name = Export[5].symbol_name }); @export(construct, .{ .name = Export[6].symbol_name }); } } pub const Extern = [_][]const u8{ "createObject", "fromJS" }; }; } pub fn WritableStreamSink( comptime Context: type, comptime onStart: ?fn (this: Context) void, comptime onWrite: fn (this: Context, bytes: []const u8) JSC.Maybe(Blob.SizeType), comptime onAbort: ?fn (this: Context) void, comptime onClose: ?fn (this: Context) void, comptime deinit: ?fn (this: Context) void, ) type { return struct { context: Context, closed: bool = false, deinited: bool = false, pending_err: ?JSC.Node.Syscall.Error = null, aborted: bool = false, abort_signaler: ?*anyopaque = null, onAbortCallback: ?fn (?*anyopaque) void = null, close_signaler: ?*anyopaque = null, onCloseCallback: ?fn (?*anyopaque) void = null, pub const This = @This(); pub fn write(this: *This, bytes: []const u8) JSC.Maybe(Blob.SizeType) { if (this.pending_err) |err| { this.pending_err = null; return .{ .err = err }; } if (this.closed or this.aborted or this.deinited) { return .{ .result = 0 }; } return onWrite(&this.context, bytes); } pub fn start(this: *This) StreamStart { return onStart(&this.context); } pub fn abort(this: *This) void { if (this.closed or this.deinited or this.aborted) { return; } this.aborted = true; onAbort(&this.context); } pub fn didAbort(this: *This) void { if (this.closed or this.deinited or this.aborted) { return; } this.aborted = true; if (this.onAbortCallback) |cb| { this.onAbortCallback = null; cb(this.abort_signaler); } } pub fn didClose(this: *This) void { if (this.closed or this.deinited or this.aborted) { return; } this.closed = true; if (this.onCloseCallback) |cb| { this.onCloseCallback = null; cb(this.close_signaler); } } pub fn close(this: *This) void { if (this.closed or this.deinited or this.aborted) { return; } this.closed = true; onClose(this.context); } pub fn deinit(this: *This) void { if (this.deinited) { return; } this.deinited = true; deinit(this.context); } pub fn getError(this: *This) ?JSC.Node.Syscall.Error { if (this.pending_err) |err| { this.pending_err = null; return err; } return null; } }; } pub fn HTTPServerWritable(comptime ssl: bool) type { return struct { pub const UWSResponse = uws.NewApp(ssl).Response; res: *UWSResponse, pending_chunk: []const u8 = "", is_listening_for_abort: bool = false, wrote: Blob.SizeType = 0, callback: anyframe->JSC.Maybe(Blob.SizeType) = undefined, writable: Writable, pub fn onWritable(this: *@This(), available: c_ulong, _: *UWSResponse) callconv(.C) bool { const to_write = @minimum(@truncate(Blob.SizeType, available), @truncate(Blob.SizeType, this.pending_chunk.len)); if (!this.res.write(this.pending_chunk[0..to_write])) { return true; } this.pending_chunk = this.pending_chunk[to_write..]; this.wrote += to_write; if (this.pending_chunk.len > 0) { this.res.onWritable(*@This(), onWritable, this); return true; } var callback = this.callback; this.callback = undefined; // TODO: clarify what the boolean means resume callback; bun.default_allocator.destroy(callback.*); return false; } pub fn onStart(this: *@This()) void { if (this.res.hasResponded()) { this.writable.didClose(); } } pub fn onWrite(this: *@This(), bytes: []const u8) JSC.Maybe(Blob.SizeType) { if (this.writable.aborted) { return .{ .result = 0 }; } if (this.pending_chunk.len > 0) { return JSC.Maybe(Blob.SizeType).retry; } if (this.res.write(bytes)) { return .{ .result = @truncate(Blob.SizeType, bytes.len) }; } this.pending_chunk = bytes; this.writable.pending_err = null; suspend { if (!this.is_listening_for_abort) { this.is_listening_for_abort = true; this.res.onAborted(*@This(), onAborted); } this.res.onWritable(*@This(), onWritable, this); var frame = bun.default_allocator.create(@TypeOf(@Frame(onWrite))) catch unreachable; this.callback = frame; frame.* = @frame().*; } const wrote = this.wrote; this.wrote = 0; if (this.writable.pending_err) |err| { this.writable.pending_err = null; return .{ .err = err }; } return .{ .result = wrote }; } // client-initiated pub fn onAborted(this: *@This(), _: *UWSResponse) void { this.writable.didAbort(); } // writer-initiated pub fn onAbort(this: *@This()) void { this.res.end("", true); } pub fn onClose(this: *@This()) void { this.res.end("", false); } pub fn deinit(_: *@This()) void {} pub const Writable = WritableStreamSink(@This(), onStart, onWrite, onAbort, onClose, deinit); }; } pub const HTTPSWriter = HTTPServerWritable(true); pub const HTTPWriter = HTTPServerWritable(false); pub fn ReadableStreamSource( comptime Context: type, comptime name_: []const u8, comptime onStart: anytype, comptime onPull: anytype, comptime onCancel: fn (this: *Context) void, comptime deinit: fn (this: *Context) void, ) type { return struct { context: Context, cancelled: bool = false, deinited: bool = false, pending_err: ?JSC.Node.Syscall.Error = null, close_handler: ?fn (*anyopaque) void = null, close_ctx: ?*anyopaque = null, close_jsvalue: JSValue = JSValue.zero, globalThis: *JSGlobalObject = undefined, const This = @This(); const ReadableStreamSourceType = @This(); pub fn pull(this: *This, buf: []u8) StreamResult { return onPull(&this.context, buf, JSValue.zero); } pub fn start( this: *This, ) StreamStart { return onStart(&this.context); } pub fn pullFromJS(this: *This, buf: []u8, view: JSValue) StreamResult { return onPull(&this.context, buf, view); } pub fn startFromJS(this: *This) StreamStart { return onStart(&this.context); } pub fn cancel(this: *This) void { if (this.cancelled or this.deinited) { return; } this.cancelled = true; onCancel(&this.context); } pub fn onClose(this: *This) void { if (this.cancelled or this.deinited) { return; } if (this.close_handler) |close| { this.close_handler = null; close(this.close_ctx); } } pub fn deinit(this: *This) void { if (this.deinited) { return; } this.deinited = true; deinit(&this.context); } pub fn getError(this: *This) ?JSC.Node.Syscall.Error { if (this.pending_err) |err| { this.pending_err = null; return err; } return null; } pub fn toJS(this: *ReadableStreamSourceType, globalThis: *JSGlobalObject) JSC.JSValue { return ReadableStream.fromNative(globalThis, Context.tag, this); } pub const JSReadableStreamSource = struct { pub const shim = JSC.Shimmer(std.mem.span(name_), "JSReadableStreamSource", @This()); pub const name = std.fmt.comptimePrint("{s}_JSReadableStreamSource", .{std.mem.span(name_)}); pub fn pull(globalThis: *JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSC.JSValue { var this = callFrame.argument(0).asPtr(ReadableStreamSourceType); const view = callFrame.argument(1); view.ensureStillAlive(); var buffer = view.asArrayBuffer(globalThis) orelse return JSC.JSValue.jsUndefined(); return processResult( globalThis, callFrame, this.pullFromJS(buffer.slice(), view), ); } pub fn start(globalThis: *JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSC.JSValue { var this = callFrame.argument(0).asPtr(ReadableStreamSourceType); switch (this.startFromJS()) { .empty => return JSValue.jsNumber(0), .ready => return JSValue.jsNumber(16384), .chunk_size => |size| return JSValue.jsNumber(size), .err => |err| { globalThis.vm().throwError(globalThis, err.toJSC(globalThis)); return JSC.JSValue.jsUndefined(); }, else => unreachable, } } pub fn processResult(globalThis: *JSGlobalObject, callFrame: *JSC.CallFrame, result: StreamResult) JSC.JSValue { switch (result) { .err => |err| { globalThis.vm().throwError(globalThis, err.toJSC(globalThis)); return JSValue.jsUndefined(); }, .temporary_and_done, .owned_and_done, .into_array_and_done => { JSC.C.JSObjectSetPropertyAtIndex(globalThis.ref(), callFrame.argument(2).asObjectRef(), 0, JSValue.jsBoolean(true).asObjectRef(), null); return result.toJS(globalThis); }, else => return result.toJS(globalThis), } } pub fn cancel(_: *JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSC.JSValue { var this = callFrame.argument(0).asPtr(ReadableStreamSourceType); this.cancel(); return JSC.JSValue.jsUndefined(); } pub fn setClose(globalThis: *JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSC.JSValue { var this = callFrame.argument(0).asPtr(ReadableStreamSourceType); this.close_ctx = this; this.close_handler = JSReadableStreamSource.onClose; this.globalThis = globalThis; this.close_jsvalue = callFrame.argument(1); return JSC.JSValue.jsUndefined(); } fn onClose(ptr: *anyopaque) void { var this = bun.cast(*ReadableStreamSourceType, ptr); _ = this.close_jsvalue.call(this.globalThis, &.{}); // this.closer } pub fn deinit(_: *JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSC.JSValue { var this = callFrame.argument(0).asPtr(ReadableStreamSourceType); this.deinit(); return JSValue.jsUndefined(); } pub fn load(globalThis: *JSGlobalObject) callconv(.C) JSC.JSValue { if (comptime JSC.is_bindgen) unreachable; if (comptime Environment.allow_assert) { // this should be cached per globals object const OnlyOnce = struct { pub threadlocal var last_globals: ?*JSGlobalObject = null; }; if (OnlyOnce.last_globals) |last_globals| { std.debug.assert(last_globals != globalThis); } OnlyOnce.last_globals = globalThis; } return JSC.JSArray.from(globalThis, &.{ JSC.NewFunction(globalThis, null, 1, JSReadableStreamSource.pull), JSC.NewFunction(globalThis, null, 1, JSReadableStreamSource.start), JSC.NewFunction(globalThis, null, 1, JSReadableStreamSource.cancel), JSC.NewFunction(globalThis, null, 1, JSReadableStreamSource.setClose), JSC.NewFunction(globalThis, null, 1, JSReadableStreamSource.deinit), }); } pub const Export = shim.exportFunctions(.{ .@"load" = load, }); comptime { if (!JSC.is_bindgen) { @export(load, .{ .name = Export[0].symbol_name }); _ = JSReadableStreamSource.pull; _ = JSReadableStreamSource.start; _ = JSReadableStreamSource.cancel; _ = JSReadableStreamSource.setClose; _ = JSReadableStreamSource.load; } } }; }; } pub const ByteBlobLoader = struct { offset: Blob.SizeType = 0, store: *Blob.Store, chunk_size: Blob.SizeType = 1024 * 1024 * 2, remain: Blob.SizeType = 1024 * 1024 * 2, done: bool = false, pub const tag = ReadableStream.Tag.Blob; pub fn setup( this: *ByteBlobLoader, blob: *const Blob, user_chunk_size: Blob.SizeType, ) void { blob.store.?.ref(); var blobe = blob.*; blobe.resolveSize(); this.* = ByteBlobLoader{ .offset = blobe.offset, .store = blobe.store.?, .chunk_size = if (user_chunk_size > 0) @minimum(user_chunk_size, blobe.size) else @minimum(1024 * 1024 * 2, blobe.size), .remain = blobe.size, .done = false, }; } pub fn onStart(this: *ByteBlobLoader) StreamStart { return .{ .chunk_size = this.chunk_size }; } pub fn onPull(this: *ByteBlobLoader, buffer: []u8, array: JSC.JSValue) StreamResult { array.ensureStillAlive(); defer array.ensureStillAlive(); if (this.done) { return .{ .done = {} }; } var temporary = this.store.sharedView(); temporary = temporary[this.offset..]; temporary = temporary[0..@minimum(buffer.len, @minimum(temporary.len, this.remain))]; if (temporary.len == 0) { this.store.deref(); this.done = true; return .{ .done = {} }; } const copied = @intCast(Blob.SizeType, temporary.len); this.remain -|= copied; this.offset +|= copied; @memcpy(buffer.ptr, temporary.ptr, temporary.len); if (this.remain == 0) { return .{ .into_array_and_done = .{ .value = array, .len = copied } }; } return .{ .into_array = .{ .value = array, .len = copied } }; } pub fn onCancel(_: *ByteBlobLoader) void {} pub fn deinit(this: *ByteBlobLoader) void { if (!this.done) { this.done = true; this.store.deref(); } bun.default_allocator.destroy(this); } pub const Source = ReadableStreamSource(@This(), "ByteBlob", onStart, onPull, onCancel, deinit); }; pub fn RequestBodyStreamer( comptime is_ssl: bool, ) type { return struct { response: *uws.NewApp(is_ssl).Response, pub const tag = if (is_ssl) ReadableStream.Tag.HTTPRequest else if (is_ssl) ReadableStream.Tag.HTTPSRequest; pub fn onStart(this: *ByteBlobLoader) StreamStart { return .{ .chunk_size = this.chunk_size }; } pub fn onPull(this: *ByteBlobLoader, buffer: []u8, array: JSC.JSValue) StreamResult { array.ensureStillAlive(); defer array.ensureStillAlive(); if (this.done) { return .{ .done = {} }; } var temporary = this.store.sharedView(); temporary = temporary[this.offset..]; temporary = temporary[0..@minimum(buffer.len, @minimum(temporary.len, this.remain))]; if (temporary.len == 0) { this.store.deref(); this.done = true; return .{ .done = {} }; } const copied = @intCast(Blob.SizeType, temporary.len); this.remain -|= copied; this.offset +|= copied; @memcpy(buffer.ptr, temporary.ptr, temporary.len); if (this.remain == 0) { return .{ .into_array_and_done = .{ .value = array, .len = copied } }; } return .{ .into_array = .{ .value = array, .len = copied } }; } pub fn onCancel(_: *ByteBlobLoader) void {} pub fn deinit(this: *ByteBlobLoader) void { if (!this.done) { this.done = true; this.store.deref(); } bun.default_allocator.destroy(this); } pub const label = if (is_ssl) "HTTPSRequestBodyStreamer" else "HTTPRequestBodyStreamer"; pub const Source = ReadableStreamSource(@This(), label, onStart, onPull, onCancel, deinit); }; } pub const FileBlobLoader = struct { buf: []u8 = &[_]u8{}, protected_view: JSC.JSValue = JSC.JSValue.zero, fd: JSC.Node.FileDescriptor = 0, auto_close: bool = false, loop: *JSC.EventLoop = undefined, mode: JSC.Node.Mode = 0, store: *Blob.Store, total_read: Blob.SizeType = 0, finalized: bool = false, callback: anyframe = undefined, pending: StreamResult.Pending = StreamResult.Pending{ .frame = undefined, .used = false, .result = .{ .done = {} }, }, cancelled: bool = false, user_chunk_size: Blob.SizeType = 0, scheduled_count: u32 = 0, concurrent: Concurrent = Concurrent{}, input_tag: StreamResult.Tag = StreamResult.Tag.done, const FileReader = @This(); const run_on_different_thread_size = bun.huge_allocator_threshold; pub const tag = ReadableStream.Tag.File; pub fn setup(this: *FileBlobLoader, store: *Blob.Store, chunk_size: Blob.SizeType) void { store.ref(); this.* = .{ .loop = JSC.VirtualMachine.vm.eventLoop(), .auto_close = store.data.file.pathlike == .path, .store = store, .user_chunk_size = chunk_size, }; } pub fn watch(this: *FileReader) void { _ = JSC.VirtualMachine.vm.poller.watch(this.fd, .read, this, callback); this.scheduled_count += 1; } const Concurrent = struct { read: Blob.SizeType = 0, task: NetworkThread.Task = .{ .callback = Concurrent.taskCallback }, completion: AsyncIO.Completion = undefined, read_frame: anyframe = undefined, chunk_size: Blob.SizeType = 0, main_thread_task: JSC.AnyTask = .{ .callback = onJSThread, .ctx = null }, pub fn taskCallback(task: *NetworkThread.Task) void { var this = @fieldParentPtr(FileBlobLoader, "concurrent", @fieldParentPtr(Concurrent, "task", task)); var frame = HTTPClient.getAllocator().create(@Frame(runAsync)) catch unreachable; _ = @asyncCall(std.mem.asBytes(frame), undefined, runAsync, .{this}); } pub fn onRead(this: *FileBlobLoader, completion: *HTTPClient.NetworkThread.Completion, result: AsyncIO.ReadError!usize) void { this.concurrent.read = @truncate(Blob.SizeType, result catch |err| { if (@hasField(HTTPClient.NetworkThread.Completion, "result")) { this.pending.result = .{ .err = JSC.Node.Syscall.Error{ .errno = @intCast(JSC.Node.Syscall.Error.Int, -completion.result), .syscall = .read, }, }; } else { this.pending.result = .{ .err = JSC.Node.Syscall.Error{ // this is too hacky .errno = @truncate(JSC.Node.Syscall.Error.Int, @intCast(u16, @maximum(1, @errorToInt(err)))), .syscall = .read, }, }; } this.concurrent.read = 0; resume this.concurrent.read_frame; return; }); resume this.concurrent.read_frame; } pub fn scheduleRead(this: *FileBlobLoader) void { if (comptime Environment.isMac) { var remaining = this.buf[this.concurrent.read..]; while (remaining.len > 0) { const to_read = @minimum(@as(usize, this.concurrent.chunk_size), remaining.len); switch (JSC.Node.Syscall.read(this.fd, remaining[0..to_read])) { .err => |err| { const retry = comptime if (Environment.isLinux) std.os.E.WOULDBLOCK else std.os.E.AGAIN; switch (err.getErrno()) { retry => break, else => {}, } this.pending.result = .{ .err = err }; scheduleMainThreadTask(this); return; }, .result => |result| { this.concurrent.read += @intCast(Blob.SizeType, result); remaining = remaining[result..]; if (result == 0) { remaining.len = 0; break; } }, } } if (remaining.len == 0) { scheduleMainThreadTask(this); return; } } AsyncIO.global.read( *FileBlobLoader, this, onRead, &this.concurrent.completion, this.fd, this.buf[this.concurrent.read..], null, ); suspend { var _frame = @frame(); var this_frame = HTTPClient.getAllocator().create(std.meta.Child(@TypeOf(_frame))) catch unreachable; this_frame.* = _frame.*; this.concurrent.read_frame = this_frame; } scheduleMainThreadTask(this); } pub fn onJSThread(task_ctx: *anyopaque) void { var this: *FileBlobLoader = bun.cast(*FileBlobLoader, task_ctx); const protected_view = this.protected_view; defer protected_view.unprotect(); this.protected_view = JSC.JSValue.zero; if (this.finalized and this.scheduled_count == 0) { if (!this.pending.used) { resume this.pending.frame; } this.scheduled_count -= 1; this.deinit(); return; } if (!this.pending.used and this.pending.result == .err and this.concurrent.read == 0) { resume this.pending.frame; this.scheduled_count -= 1; this.finalize(); return; } if (this.concurrent.read == 0) { this.pending.result = .{ .done = {} }; resume this.pending.frame; this.scheduled_count -= 1; this.finalize(); return; } this.pending.result = this.handleReadChunk(@as(usize, this.concurrent.read)); resume this.pending.frame; this.scheduled_count -= 1; if (this.pending.result.isDone()) { this.finalize(); } } pub fn scheduleMainThreadTask(this: *FileBlobLoader) void { this.concurrent.main_thread_task.ctx = this; this.loop.enqueueTaskConcurrent(JSC.Task.init(&this.concurrent.main_thread_task)); } fn runAsync(this: *FileBlobLoader) void { this.concurrent.read = 0; Concurrent.scheduleRead(this); suspend { HTTPClient.getAllocator().destroy(@frame()); } } }; pub fn scheduleAsync(this: *FileReader, chunk_size: Blob.SizeType) void { this.scheduled_count += 1; this.loop.virtual_machine.active_tasks +|= 1; NetworkThread.init() catch {}; this.concurrent.chunk_size = chunk_size; NetworkThread.global.pool.schedule(.{ .head = &this.concurrent.task, .tail = &this.concurrent.task, .len = 1 }); } const default_fifo_chunk_size = 1024; const default_file_chunk_size = 1024 * 1024 * 2; pub fn onStart(this: *FileBlobLoader) StreamStart { var file = &this.store.data.file; var file_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined; var auto_close = this.auto_close; defer this.auto_close = auto_close; var fd = if (!auto_close) file.pathlike.fd else switch (JSC.Node.Syscall.open(file.pathlike.path.sliceZ(&file_buf), std.os.O.RDONLY | std.os.O.NONBLOCK | std.os.O.CLOEXEC, 0)) { .result => |_fd| _fd, .err => |err| { this.deinit(); return .{ .err = err.withPath(file.pathlike.path.slice()) }; }, }; if (!auto_close) { // ensure we have non-blocking IO set const flags = std.os.fcntl(fd, std.os.F.GETFL, 0) catch return .{ .err = JSC.Node.Syscall.Error.fromCode(std.os.E.BADF, .fcntl) }; // if we do not, clone the descriptor and set non-blocking // it is important for us to clone it so we don't cause Weird Things to happen if ((flags & std.os.O.NONBLOCK) == 0) { auto_close = true; fd = @intCast(@TypeOf(fd), std.os.fcntl(fd, std.os.F.DUPFD, 0) catch return .{ .err = JSC.Node.Syscall.Error.fromCode(std.os.E.BADF, .fcntl) }); _ = std.os.fcntl(fd, std.os.F.SETFL, flags | std.os.O.NONBLOCK) catch return .{ .err = JSC.Node.Syscall.Error.fromCode(std.os.E.BADF, .fcntl) }; } } const stat: std.os.Stat = switch (JSC.Node.Syscall.fstat(fd)) { .result => |result| result, .err => |err| { if (auto_close) { _ = JSC.Node.Syscall.close(fd); } this.deinit(); return .{ .err = err.withPath(file.pathlike.path.slice()) }; }, }; if (std.os.S.ISDIR(stat.mode)) { const err = JSC.Node.Syscall.Error.fromCode(.ISDIR, .fstat); if (auto_close) { _ = JSC.Node.Syscall.close(fd); } this.deinit(); return .{ .err = err }; } if (std.os.S.ISSOCK(stat.mode)) { const err = JSC.Node.Syscall.Error.fromCode(.INVAL, .fstat); if (auto_close) { _ = JSC.Node.Syscall.close(fd); } this.deinit(); return .{ .err = err }; } file.seekable = std.os.S.ISREG(stat.mode); file.mode = @intCast(JSC.Node.Mode, stat.mode); this.mode = file.mode; if (file.seekable orelse false) file.max_size = @intCast(Blob.SizeType, stat.size); if ((file.seekable orelse false) and file.max_size == 0) { if (auto_close) { _ = JSC.Node.Syscall.close(fd); } this.deinit(); return .{ .empty = {} }; } this.fd = fd; this.auto_close = auto_close; const chunk_size = this.calculateChunkSize(std.math.maxInt(usize)); return .{ .chunk_size = @truncate(Blob.SizeType, chunk_size) }; } fn calculateChunkSize(this: *FileBlobLoader, available_to_read: usize) usize { const file = &this.store.data.file; const chunk_size: usize = if (this.user_chunk_size > 0) @as(usize, this.user_chunk_size) else if (file.seekable orelse false) @as(usize, default_file_chunk_size) else @as(usize, default_fifo_chunk_size); return if (file.max_size > 0) if (available_to_read != std.math.maxInt(usize)) @minimum(chunk_size, available_to_read) else @minimum(@maximum(this.total_read, file.max_size) - this.total_read, chunk_size) else @minimum(available_to_read, chunk_size); } pub fn onPullInto(this: *FileBlobLoader, buffer: []u8, view: JSC.JSValue) StreamResult { const chunk_size = this.calculateChunkSize(std.math.maxInt(usize)); this.input_tag = .into_array; switch (chunk_size) { 0 => { std.debug.assert(this.store.data.file.seekable orelse false); this.finalize(); return .{ .done = {} }; }, run_on_different_thread_size...std.math.maxInt(@TypeOf(chunk_size)) => { this.protected_view = view; this.protected_view.protect(); // should never be reached this.pending.result = .{ .err = JSC.Node.Syscall.Error.todo, }; this.buf = buffer; this.scheduleAsync(@truncate(Blob.SizeType, chunk_size)); return .{ .pending = &this.pending }; }, else => {}, } return this.read(buffer, view); } fn maybeAutoClose(this: *FileBlobLoader) void { if (this.auto_close) { _ = JSC.Node.Syscall.close(this.fd); this.auto_close = false; } } fn handleReadChunk(this: *FileBlobLoader, result: usize) StreamResult { this.total_read += @intCast(Blob.SizeType, result); const remaining: Blob.SizeType = if (this.store.data.file.seekable orelse false) this.store.data.file.max_size -| this.total_read else @as(Blob.SizeType, std.math.maxInt(Blob.SizeType)); // this handles: // - empty file // - stream closed for some reason if ((result == 0 and remaining == 0)) { this.finalize(); return .{ .done = {} }; } const has_more = remaining > 0; if (!has_more) { return .{ .into_array_and_done = .{ .len = @truncate(Blob.SizeType, result) } }; } return .{ .into_array = .{ .len = @truncate(Blob.SizeType, result) } }; } pub fn read( this: *FileBlobLoader, read_buf: []u8, view: JSC.JSValue, ) StreamResult { const rc = JSC.Node.Syscall.read(this.fd, read_buf); switch (rc) { .err => |err| { const retry = std.os.E.AGAIN; switch (err.getErrno()) { retry => { this.protected_view = view; this.protected_view.protect(); this.buf = read_buf; this.watch(); return .{ .pending = &this.pending, }; }, else => {}, } const sys = if (this.store.data.file.pathlike == .path and this.store.data.file.pathlike.path.slice().len > 0) err.withPath(this.store.data.file.pathlike.path.slice()) else err; this.finalize(); return .{ .err = sys }; }, .result => |result| { return this.handleReadChunk(result); }, } } pub fn callback(task: ?*anyopaque, sizeOrOffset: i64, _: u16) void { var this: *FileReader = bun.cast(*FileReader, task.?); this.scheduled_count -= 1; const protected_view = this.protected_view; defer protected_view.unprotect(); this.protected_view = JSValue.zero; var available_to_read: usize = std.math.maxInt(usize); if (comptime Environment.isMac) { if (std.os.S.ISREG(this.mode)) { // Returns when the file pointer is not at the end of // file. data contains the offset from current position // to end of file, and may be negative. available_to_read = @intCast(usize, @maximum(sizeOrOffset, 0)); } else if (std.os.S.ISCHR(this.mode) or std.os.S.ISFIFO(this.mode)) { available_to_read = @intCast(usize, @maximum(sizeOrOffset, 0)); } } if (this.finalized and this.scheduled_count == 0) { if (!this.pending.used) { // should never be reached this.pending.result = .{ .err = JSC.Node.Syscall.Error.todo, }; resume this.pending.frame; } this.deinit(); return; } if (this.cancelled) return; if (this.buf.len == 0) { return; } else { this.buf.len = @minimum(this.buf.len, available_to_read); } this.pending.result = this.read(this.buf, this.protected_view); resume this.pending.frame; } pub fn finalize(this: *FileBlobLoader) void { if (this.finalized) return; this.finalized = true; this.maybeAutoClose(); this.store.deref(); } pub fn onCancel(this: *FileBlobLoader) void { this.cancelled = true; this.deinit(); } pub fn deinit(this: *FileBlobLoader) void { this.finalize(); if (this.scheduled_count == 0 and !this.pending.used) { this.destroy(); } } pub fn destroy(this: *FileBlobLoader) void { bun.default_allocator.destroy(this); } pub const Source = ReadableStreamSource(@This(), "FileBlobLoader", onStart, onPullInto, onCancel, deinit); }; // pub const HTTPRequest = RequestBodyStreamer(false); // pub const HTTPSRequest = RequestBodyStreamer(true); // pub fn ResponseBodyStreamer(comptime is_ssl: bool) type { // return struct { // const Streamer = @This(); // pub fn onEnqueue(this: *Streamer, buffer: []u8, ): anytype, // pub fn onEnqueueMany(this: *Streamer): anytype, // pub fn onClose(this: *Streamer): anytype, // pub fn onError(this: *Streamer): anytype, // }; // }