An open socket connection with a network protocol that guarantees sequencing, delivery, and prevents repetition. Typically TCP or UNIX domain socket.
pub const Stream = structpub 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.handle)[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 = error{
/// Another TCP Fast Open is already in progress.
FastOpenAlreadyInProgress,
/// Network session was unexpectedly closed by recipient.
ConnectionResetByPeer,
/// The output queue for a network interface was full. This generally indicates that the
/// interface has stopped sending, but may be caused by transient congestion. (Normally,
/// this does not occur in Linux. Packets are just silently dropped when a device queue
/// overflows.)
///
/// This is also caused when there is not enough kernel memory available.
SystemResources,
/// No route to network.
NetworkUnreachable,
/// Network reached but no route to host.
HostUnreachable,
/// The local network interface used to reach the destination is down.
NetworkDown,
/// The destination address is not listening.
ConnectionRefused,
/// The passed address didn't have the correct address family in its sa_family field.
AddressFamilyUnsupported,
/// Local end has been shut down on a connection-oriented socket, or
/// the socket was never connected.
SocketUnconnected,
SocketNotBound,
} || Io.UnexpectedError || 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 n = io.vtable.netWrite(io.userdata, handle, buffered, data, splat) 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);
}
}