Stream

pub const Stream = struct {
    socket: Socket,

    const max_iovecs_len = 8;

    /// This is a low-level API that calls the `Io` interface function directly.
    /// For a higher level API, see `reader`.
    pub fn read(s: *const Stream, io: Io, data: [][]u8) Reader.Error!usize {
        return (try io.operate(.{ .net_read = .{
            .socket_handle = s.socket.handle,
            .data = data,
        } })).net_read;
    }

    pub fn close(s: *const Stream, io: Io) void {
        io.vtable.netClose(io.userdata, (&s.socket)[0..1]);
    }

    pub fn shutdown(s: *const Stream, io: Io, how: ShutdownHow) ShutdownError!void {
        return io.vtable.netShutdown(io.userdata, s.socket.handle, how);
    }

    pub const Reader = struct {
        io: Io,
        interface: Io.Reader,
        stream: Stream,
        err: ?Error,

        pub const Error = Io.Operation.NetRead.Error || Io.Cancelable;

        pub fn init(stream: Stream, io: Io, buffer: []u8) Reader {
            return .{
                .io = io,
                .interface = .{
                    .vtable = &.{
                        .stream = streamImpl,
                        .readVec = readVec,
                    },
                    .buffer = buffer,
                    .seek = 0,
                    .end = 0,
                },
                .stream = stream,
                .err = null,
            };
        }

        fn streamImpl(io_r: *Io.Reader, io_w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!usize {
            const dest = limit.slice(try io_w.writableSliceGreedy(1));
            var data: [1][]u8 = .{dest};
            const n = try readVec(io_r, &data);
            io_w.advance(n);
            return n;
        }

        fn readVec(io_r: *Io.Reader, data: [][]u8) Io.Reader.Error!usize {
            const r: *Reader = @alignCast(@fieldParentPtr("interface", io_r));
            const io = r.io;
            var iovecs_buffer: [max_iovecs_len][]u8 = undefined;
            const dest_n, const data_size = try io_r.writableVector(&iovecs_buffer, data);
            const dest = iovecs_buffer[0..dest_n];
            assert(dest[0].len > 0);
            const n = r.stream.read(io, dest) catch |err| {
                r.err = err;
                return error.ReadFailed;
            };
            if (n == 0) {
                return error.EndOfStream;
            }
            if (n > data_size) {
                r.interface.end += n - data_size;
                return data_size;
            }
            return n;
        }
    };

    pub const Writer = struct {
        io: Io,
        interface: Io.Writer,
        stream: Stream,
        err: ?Error = null,
        write_file_err: ?WriteFileError = null,

        pub const Error = Io.Operation.NetWrite.Error || Io.Cancelable;

        pub const WriteFileError = Error || error{
            /// The `Io` implementation cannot offer a more efficient
            /// file-to-socket path; the caller should fall back to read-based
            /// copying. See `Io.Writer.sendFile`.
            Unimplemented,
            /// Reached the end of the file being read.
            EndOfStream,
            /// The source `File.Reader` failed; detailed diagnostics are found
            /// on that struct.
            ReadFailed,
        };

        pub fn init(stream: Stream, io: Io, buffer: []u8) Writer {
            return .{
                .io = io,
                .stream = stream,
                .interface = .{
                    .vtable = &.{
                        .drain = drain,
                        .sendFile = sendFile,
                    },
                    .buffer = buffer,
                },
            };
        }

        fn drain(io_w: *Io.Writer, data: []const []const u8, splat: usize) Io.Writer.Error!usize {
            const w: *Writer = @alignCast(@fieldParentPtr("interface", io_w));
            const io = w.io;
            const buffered = io_w.buffered();
            const handle = w.stream.socket.handle;
            const result = io.operate(.{ .net_write = .{
                .socket_handle = handle,
                .header = buffered,
                .data = data,
                .splat = splat,
            } }) catch |err| {
                w.err = err;
                return error.WriteFailed;
            };
            const n = result.net_write catch |err| {
                w.err = err;
                return error.WriteFailed;
            };
            return io_w.consume(n);
        }

        fn sendFile(io_w: *Io.Writer, file_reader: *Io.File.Reader, limit: Io.Limit) Io.Writer.FileError!usize {
            const w: *Writer = @alignCast(@fieldParentPtr("interface", io_w));
            const io = w.io;
            const header = io_w.buffered();
            const handle = w.stream.socket.handle;
            const n = io.vtable.netWriteFile(io.userdata, handle, header, file_reader, limit) catch |err| switch (err) {
                error.Canceled => {
                    w.err = error.Canceled;
                    return error.WriteFailed;
                },
                error.EndOfStream,
                error.Unimplemented,
                error.ReadFailed,
                => |e| return e,
                else => |e| {
                    w.write_file_err = e;
                    return error.WriteFailed;
                },
            };
            return io_w.consume(n);
        }
    };

    pub fn reader(stream: Stream, io: Io, buffer: []u8) Reader {
        return .init(stream, io, buffer);
    }

    pub fn writer(stream: Stream, io: Io, buffer: []u8) Writer {
        return .init(stream, io, buffer);
    }
}