mirror of
https://github.com/denoland/deno.git
synced 2024-11-22 15:06:54 -05:00
530 lines
11 KiB
JavaScript
530 lines
11 KiB
JavaScript
// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
|
|
|
|
import { core, primordials } from "ext:core/mod.js";
|
|
const { BadResourcePrototype, InterruptedPrototype, ops } = core;
|
|
import {
|
|
readableStreamForRidUnrefable,
|
|
readableStreamForRidUnrefableRef,
|
|
readableStreamForRidUnrefableUnref,
|
|
writableStreamForRid,
|
|
} from "ext:deno_web/06_streams.js";
|
|
import * as abortSignal from "ext:deno_web/03_abort_signal.js";
|
|
import { SymbolDispose } from "ext:deno_web/00_infra.js";
|
|
const {
|
|
op_dns_resolve,
|
|
op_net_accept_tcp,
|
|
op_net_accept_unix,
|
|
op_net_connect_tcp,
|
|
op_net_connect_unix,
|
|
op_net_recv_udp,
|
|
op_net_recv_unixpacket,
|
|
op_net_send_udp,
|
|
op_net_send_unixpacket,
|
|
op_net_set_multi_loopback_udp,
|
|
op_net_set_multi_ttl_udp,
|
|
op_net_join_multi_v4_udp,
|
|
op_net_join_multi_v6_udp,
|
|
op_net_leave_multi_v4_udp,
|
|
op_net_leave_multi_v6_udp,
|
|
} = core.ensureFastOps();
|
|
|
|
const {
|
|
Error,
|
|
Number,
|
|
ObjectPrototypeIsPrototypeOf,
|
|
PromiseResolve,
|
|
SafeSet,
|
|
SetPrototypeAdd,
|
|
SetPrototypeDelete,
|
|
SetPrototypeForEach,
|
|
SymbolAsyncIterator,
|
|
Symbol,
|
|
TypeError,
|
|
TypedArrayPrototypeSubarray,
|
|
Uint8Array,
|
|
} = primordials;
|
|
|
|
async function write(rid, data) {
|
|
return await core.write(rid, data);
|
|
}
|
|
|
|
function shutdown(rid) {
|
|
return core.shutdown(rid);
|
|
}
|
|
|
|
async function resolveDns(query, recordType, options) {
|
|
let cancelRid;
|
|
let abortHandler;
|
|
if (options?.signal) {
|
|
options.signal.throwIfAborted();
|
|
cancelRid = ops.op_cancel_handle();
|
|
abortHandler = () => core.tryClose(cancelRid);
|
|
options.signal[abortSignal.add](abortHandler);
|
|
}
|
|
|
|
try {
|
|
return await op_dns_resolve({
|
|
cancelRid,
|
|
query,
|
|
recordType,
|
|
options,
|
|
});
|
|
} finally {
|
|
if (options?.signal) {
|
|
options.signal[abortSignal.remove](abortHandler);
|
|
|
|
// always throw the abort error when aborted
|
|
options.signal.throwIfAborted();
|
|
}
|
|
}
|
|
}
|
|
|
|
class Conn {
|
|
#rid = 0;
|
|
#remoteAddr = null;
|
|
#localAddr = null;
|
|
#unref = false;
|
|
#pendingReadPromises = new SafeSet();
|
|
|
|
#readable;
|
|
#writable;
|
|
|
|
constructor(rid, remoteAddr, localAddr) {
|
|
this.#rid = rid;
|
|
this.#remoteAddr = remoteAddr;
|
|
this.#localAddr = localAddr;
|
|
}
|
|
|
|
get rid() {
|
|
return this.#rid;
|
|
}
|
|
|
|
get remoteAddr() {
|
|
return this.#remoteAddr;
|
|
}
|
|
|
|
get localAddr() {
|
|
return this.#localAddr;
|
|
}
|
|
|
|
write(p) {
|
|
return write(this.rid, p);
|
|
}
|
|
|
|
async read(buffer) {
|
|
if (buffer.length === 0) {
|
|
return 0;
|
|
}
|
|
const promise = core.read(this.rid, buffer);
|
|
if (this.#unref) core.unrefOpPromise(promise);
|
|
SetPrototypeAdd(this.#pendingReadPromises, promise);
|
|
let nread;
|
|
try {
|
|
nread = await promise;
|
|
} catch (e) {
|
|
throw e;
|
|
} finally {
|
|
SetPrototypeDelete(this.#pendingReadPromises, promise);
|
|
}
|
|
return nread === 0 ? null : nread;
|
|
}
|
|
|
|
close() {
|
|
core.close(this.rid);
|
|
}
|
|
|
|
closeWrite() {
|
|
return shutdown(this.rid);
|
|
}
|
|
|
|
get readable() {
|
|
if (this.#readable === undefined) {
|
|
this.#readable = readableStreamForRidUnrefable(this.rid);
|
|
if (this.#unref) {
|
|
readableStreamForRidUnrefableUnref(this.#readable);
|
|
}
|
|
}
|
|
return this.#readable;
|
|
}
|
|
|
|
get writable() {
|
|
if (this.#writable === undefined) {
|
|
this.#writable = writableStreamForRid(this.rid);
|
|
}
|
|
return this.#writable;
|
|
}
|
|
|
|
ref() {
|
|
this.#unref = false;
|
|
if (this.#readable) {
|
|
readableStreamForRidUnrefableRef(this.#readable);
|
|
}
|
|
|
|
SetPrototypeForEach(
|
|
this.#pendingReadPromises,
|
|
(promise) => core.refOpPromise(promise),
|
|
);
|
|
}
|
|
|
|
unref() {
|
|
this.#unref = true;
|
|
if (this.#readable) {
|
|
readableStreamForRidUnrefableUnref(this.#readable);
|
|
}
|
|
SetPrototypeForEach(
|
|
this.#pendingReadPromises,
|
|
(promise) => core.unrefOpPromise(promise),
|
|
);
|
|
}
|
|
|
|
[SymbolDispose]() {
|
|
core.tryClose(this.#rid);
|
|
}
|
|
}
|
|
|
|
class TcpConn extends Conn {
|
|
setNoDelay(noDelay = true) {
|
|
return ops.op_set_nodelay(this.rid, noDelay);
|
|
}
|
|
|
|
setKeepAlive(keepAlive = true) {
|
|
return ops.op_set_keepalive(this.rid, keepAlive);
|
|
}
|
|
}
|
|
|
|
class UnixConn extends Conn {}
|
|
|
|
class Listener {
|
|
#rid = 0;
|
|
#addr = null;
|
|
#unref = false;
|
|
#promise = null;
|
|
|
|
constructor(rid, addr) {
|
|
this.#rid = rid;
|
|
this.#addr = addr;
|
|
}
|
|
|
|
get rid() {
|
|
return this.#rid;
|
|
}
|
|
|
|
get addr() {
|
|
return this.#addr;
|
|
}
|
|
|
|
async accept() {
|
|
let promise;
|
|
switch (this.addr.transport) {
|
|
case "tcp":
|
|
promise = op_net_accept_tcp(this.rid);
|
|
break;
|
|
case "unix":
|
|
promise = op_net_accept_unix(this.rid);
|
|
break;
|
|
default:
|
|
throw new Error(`Unsupported transport: ${this.addr.transport}`);
|
|
}
|
|
this.#promise = promise;
|
|
if (this.#unref) core.unrefOpPromise(promise);
|
|
const { 0: rid, 1: localAddr, 2: remoteAddr } = await promise;
|
|
this.#promise = null;
|
|
if (this.addr.transport == "tcp") {
|
|
localAddr.transport = "tcp";
|
|
remoteAddr.transport = "tcp";
|
|
return new TcpConn(rid, remoteAddr, localAddr);
|
|
} else if (this.addr.transport == "unix") {
|
|
return new UnixConn(
|
|
rid,
|
|
{ transport: "unix", path: remoteAddr },
|
|
{ transport: "unix", path: localAddr },
|
|
);
|
|
} else {
|
|
throw new Error("unreachable");
|
|
}
|
|
}
|
|
|
|
async next() {
|
|
let conn;
|
|
try {
|
|
conn = await this.accept();
|
|
} catch (error) {
|
|
if (
|
|
ObjectPrototypeIsPrototypeOf(BadResourcePrototype, error) ||
|
|
ObjectPrototypeIsPrototypeOf(InterruptedPrototype, error)
|
|
) {
|
|
return { value: undefined, done: true };
|
|
}
|
|
throw error;
|
|
}
|
|
return { value: conn, done: false };
|
|
}
|
|
|
|
return(value) {
|
|
this.close();
|
|
return PromiseResolve({ value, done: true });
|
|
}
|
|
|
|
close() {
|
|
core.close(this.rid);
|
|
}
|
|
|
|
[SymbolDispose]() {
|
|
core.tryClose(this.#rid);
|
|
}
|
|
|
|
[SymbolAsyncIterator]() {
|
|
return this;
|
|
}
|
|
|
|
ref() {
|
|
this.#unref = false;
|
|
if (this.#promise !== null) {
|
|
core.refOpPromise(this.#promise);
|
|
}
|
|
}
|
|
|
|
unref() {
|
|
this.#unref = true;
|
|
if (this.#promise !== null) {
|
|
core.unrefOpPromise(this.#promise);
|
|
}
|
|
}
|
|
}
|
|
|
|
class Datagram {
|
|
#rid = 0;
|
|
#addr = null;
|
|
|
|
constructor(rid, addr, bufSize = 1024) {
|
|
this.#rid = rid;
|
|
this.#addr = addr;
|
|
this.bufSize = bufSize;
|
|
}
|
|
|
|
get rid() {
|
|
return this.#rid;
|
|
}
|
|
|
|
get addr() {
|
|
return this.#addr;
|
|
}
|
|
|
|
async joinMulticastV4(addr, multiInterface) {
|
|
await op_net_join_multi_v4_udp(
|
|
this.rid,
|
|
addr,
|
|
multiInterface,
|
|
);
|
|
|
|
return {
|
|
leave: () =>
|
|
op_net_leave_multi_v4_udp(
|
|
this.rid,
|
|
addr,
|
|
multiInterface,
|
|
),
|
|
setLoopback: (loopback) =>
|
|
op_net_set_multi_loopback_udp(
|
|
this.rid,
|
|
true,
|
|
loopback,
|
|
),
|
|
setTTL: (ttl) =>
|
|
op_net_set_multi_ttl_udp(
|
|
this.rid,
|
|
ttl,
|
|
),
|
|
};
|
|
}
|
|
|
|
async joinMulticastV6(addr, multiInterface) {
|
|
await op_net_join_multi_v6_udp(
|
|
this.rid,
|
|
addr,
|
|
multiInterface,
|
|
);
|
|
|
|
return {
|
|
leave: () =>
|
|
op_net_leave_multi_v6_udp(
|
|
this.rid,
|
|
addr,
|
|
multiInterface,
|
|
),
|
|
setLoopback: (loopback) =>
|
|
op_net_set_multi_loopback_udp(
|
|
this.rid,
|
|
false,
|
|
loopback,
|
|
),
|
|
};
|
|
}
|
|
|
|
async receive(p) {
|
|
const buf = p || new Uint8Array(this.bufSize);
|
|
let nread;
|
|
let remoteAddr;
|
|
switch (this.addr.transport) {
|
|
case "udp": {
|
|
({ 0: nread, 1: remoteAddr } = await op_net_recv_udp(
|
|
this.rid,
|
|
buf,
|
|
));
|
|
remoteAddr.transport = "udp";
|
|
break;
|
|
}
|
|
case "unixpacket": {
|
|
let path;
|
|
({ 0: nread, 1: path } = await op_net_recv_unixpacket(
|
|
this.rid,
|
|
buf,
|
|
));
|
|
remoteAddr = { transport: "unixpacket", path };
|
|
break;
|
|
}
|
|
default:
|
|
throw new Error(`Unsupported transport: ${this.addr.transport}`);
|
|
}
|
|
const sub = TypedArrayPrototypeSubarray(buf, 0, nread);
|
|
return [sub, remoteAddr];
|
|
}
|
|
|
|
async send(p, opts) {
|
|
switch (this.addr.transport) {
|
|
case "udp":
|
|
return await op_net_send_udp(
|
|
this.rid,
|
|
{ hostname: opts.hostname ?? "127.0.0.1", port: opts.port },
|
|
p,
|
|
);
|
|
case "unixpacket":
|
|
return await op_net_send_unixpacket(
|
|
this.rid,
|
|
opts.path,
|
|
p,
|
|
);
|
|
default:
|
|
throw new Error(`Unsupported transport: ${this.addr.transport}`);
|
|
}
|
|
}
|
|
|
|
close() {
|
|
core.close(this.rid);
|
|
}
|
|
|
|
async *[SymbolAsyncIterator]() {
|
|
while (true) {
|
|
try {
|
|
yield await this.receive();
|
|
} catch (err) {
|
|
if (
|
|
ObjectPrototypeIsPrototypeOf(BadResourcePrototype, err) ||
|
|
ObjectPrototypeIsPrototypeOf(InterruptedPrototype, err)
|
|
) {
|
|
break;
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
const listenOptionApiName = Symbol("listenOptionApiName");
|
|
|
|
function listen(args) {
|
|
switch (args.transport ?? "tcp") {
|
|
case "tcp": {
|
|
const { 0: rid, 1: addr } = ops.op_net_listen_tcp({
|
|
hostname: args.hostname ?? "0.0.0.0",
|
|
port: Number(args.port),
|
|
}, args.reusePort);
|
|
addr.transport = "tcp";
|
|
return new Listener(rid, addr);
|
|
}
|
|
case "unix": {
|
|
const { 0: rid, 1: path } = ops.op_net_listen_unix(
|
|
args.path,
|
|
args[listenOptionApiName] ?? "Deno.listen",
|
|
);
|
|
const addr = {
|
|
transport: "unix",
|
|
path,
|
|
};
|
|
return new Listener(rid, addr);
|
|
}
|
|
default:
|
|
throw new TypeError(`Unsupported transport: '${transport}'`);
|
|
}
|
|
}
|
|
|
|
function createListenDatagram(udpOpFn, unixOpFn) {
|
|
return function listenDatagram(args) {
|
|
switch (args.transport) {
|
|
case "udp": {
|
|
const { 0: rid, 1: addr } = udpOpFn(
|
|
{
|
|
hostname: args.hostname ?? "127.0.0.1",
|
|
port: args.port,
|
|
},
|
|
args.reuseAddress ?? false,
|
|
args.loopback ?? false,
|
|
);
|
|
addr.transport = "udp";
|
|
return new Datagram(rid, addr);
|
|
}
|
|
case "unixpacket": {
|
|
const { 0: rid, 1: path } = unixOpFn(args.path);
|
|
const addr = {
|
|
transport: "unixpacket",
|
|
path,
|
|
};
|
|
return new Datagram(rid, addr);
|
|
}
|
|
default:
|
|
throw new TypeError(`Unsupported transport: '${transport}'`);
|
|
}
|
|
};
|
|
}
|
|
|
|
async function connect(args) {
|
|
switch (args.transport ?? "tcp") {
|
|
case "tcp": {
|
|
const { 0: rid, 1: localAddr, 2: remoteAddr } = await op_net_connect_tcp(
|
|
{
|
|
hostname: args.hostname ?? "127.0.0.1",
|
|
port: args.port,
|
|
},
|
|
);
|
|
localAddr.transport = "tcp";
|
|
remoteAddr.transport = "tcp";
|
|
return new TcpConn(rid, remoteAddr, localAddr);
|
|
}
|
|
case "unix": {
|
|
const { 0: rid, 1: localAddr, 2: remoteAddr } = await op_net_connect_unix(
|
|
args.path,
|
|
);
|
|
return new UnixConn(
|
|
rid,
|
|
{ transport: "unix", path: remoteAddr },
|
|
{ transport: "unix", path: localAddr },
|
|
);
|
|
}
|
|
default:
|
|
throw new TypeError(`Unsupported transport: '${transport}'`);
|
|
}
|
|
}
|
|
|
|
export {
|
|
Conn,
|
|
connect,
|
|
createListenDatagram,
|
|
Datagram,
|
|
listen,
|
|
Listener,
|
|
listenOptionApiName,
|
|
resolveDns,
|
|
shutdown,
|
|
TcpConn,
|
|
UnixConn,
|
|
};
|