Skip to content

Commit 8a94755

Browse files
committed
quic: move getwriter to separate file
1 parent 69719cd commit 8a94755

2 files changed

Lines changed: 370 additions & 331 deletions

File tree

‎lib/internal/quic/quic.js‎

Lines changed: 4 additions & 331 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,11 @@ const {
1616
ObjectDefineProperties,
1717
ObjectKeys,
1818
PromisePrototypeThen,
19-
PromiseResolve,
2019
PromiseWithResolvers,
2120
SafeSet,
2221
Symbol,
2322
SymbolAsyncDispose,
2423
SymbolAsyncIterator,
25-
SymbolDispose,
2624
SymbolIterator,
2725
TypedArrayPrototypeGetByteLength,
2826
Uint8Array,
@@ -133,16 +131,9 @@ const {
133131
} = require('internal/blob');
134132

135133
const {
136-
drainableProtocol,
137134
kValidatedSource,
138135
} = require('internal/streams/iter/types');
139136

140-
const {
141-
convertChunks,
142-
getWriterSignal,
143-
toWriterUint8Array,
144-
} = require('internal/streams/iter/utils');
145-
146137
const {
147138
from: streamFrom,
148139
fromSync: streamFromSync,
@@ -231,6 +222,10 @@ const {
231222
QuicStreamState,
232223
} = require('internal/quic/state');
233224

225+
const {
226+
getWriter,
227+
} = require('internal/quic/writer');
228+
234229
const assert = require('internal/assert');
235230

236231
const {
@@ -1551,328 +1546,6 @@ function isSyncIterable(obj) {
15511546
return obj != null && typeof obj[SymbolIterator] === 'function';
15521547
}
15531548

1554-
function getWriter(obj) {
1555-
const {
1556-
setDrainCallback,
1557-
setStopSendingCallback,
1558-
writeDesiredSize,
1559-
writeEnded,
1560-
writeChunks,
1561-
resetStream,
1562-
endWrite,
1563-
nonWritableStream,
1564-
initStreamingSource,
1565-
} = obj;
1566-
// TODO, if this is not an internal function,
1567-
// the input should be checked for types
1568-
let closed = false;
1569-
let ending = false;
1570-
let errored = false;
1571-
let error = null;
1572-
let totalBytesWritten = 0;
1573-
let drainWakeup = null;
1574-
let pendingWrite = null;
1575-
let pendingEnd = null;
1576-
1577-
function raceWithSignal(promise, signal) {
1578-
if (signal === undefined) return promise;
1579-
signal.throwIfAborted();
1580-
1581-
const deferred = PromiseWithResolvers();
1582-
const onAbort = () => deferred.reject(signal.reason);
1583-
signal.addEventListener('abort', onAbort, {
1584-
__proto__: null,
1585-
once: true,
1586-
});
1587-
PromisePrototypeThen(
1588-
promise,
1589-
(value) => {
1590-
signal.removeEventListener('abort', onAbort);
1591-
deferred.resolve(value);
1592-
},
1593-
(reason) => {
1594-
signal.removeEventListener('abort', onAbort);
1595-
deferred.reject(reason);
1596-
});
1597-
return deferred.promise;
1598-
}
1599-
1600-
function waitForDrain(signal) {
1601-
drainWakeup ??= PromiseWithResolvers();
1602-
return raceWithSignal(drainWakeup.promise, signal);
1603-
}
1604-
1605-
// Drain callback - The implementation C++/js fires this when send buffer has space
1606-
setDrainCallback(() => {
1607-
if (drainWakeup) {
1608-
drainWakeup.resolve(true);
1609-
drainWakeup = null;
1610-
}
1611-
});
1612-
1613-
setStopSendingCallback((reason) => {
1614-
if (!closed && !errored) {
1615-
errored = true;
1616-
error = reason;
1617-
if (drainWakeup != null) {
1618-
markPromiseAsHandled(drainWakeup.promise);
1619-
drainWakeup.reject(error);
1620-
drainWakeup = null;
1621-
}
1622-
}
1623-
});
1624-
1625-
// A note on backpressure handling: per the stream/iter spec, the default
1626-
// backpressure policy for writers is strict. One async write may wait for
1627-
// capacity; additional writes are rejected until it settles.
1628-
1629-
function writeConvertedSync(chunk, token) {
1630-
const isPending = pendingWrite === token && token !== undefined;
1631-
// If the stream is closed, errored, or write-ended, we cannot accept
1632-
// more data. Refuse the sync write.
1633-
if (closed || errored || writeEnded() ||
1634-
(ending && !isPending) ||
1635-
(pendingWrite !== null && !isPending)) {
1636-
return false;
1637-
}
1638-
const len = TypedArrayPrototypeGetByteLength(chunk);
1639-
if (len === 0) return true;
1640-
// Refuse the write only when there is no available capacity at
1641-
// all. If we can write we allow the write even if the
1642-
// chunk is larger than the remaining capacity -- the source
1643-
// will accept the data into the underlying queues e.g. DataQueue and
1644-
// UpdateWriteDesiredSize() will drop writeDesiredSize toward 0,
1645-
// at which point the standard drain mechanism takes over.
1646-
// This follows the iter-streams model where writes beyond the
1647-
// budget succeed and backpressure applies to *subsequent* writes.
1648-
if (writeDesiredSize() === 0) return false;
1649-
const result = writeChunks([chunk]);
1650-
if (result === undefined) return false;
1651-
totalBytesWritten += len;
1652-
return true;
1653-
}
1654-
1655-
function writeSync(chunk) {
1656-
return writeConvertedSync(toWriterUint8Array(chunk));
1657-
}
1658-
1659-
function write(chunk, options = kEmptyObject) {
1660-
chunk = toWriterUint8Array(chunk);
1661-
const signal = getWriterSignal(options);
1662-
return writeAsync(chunk, signal);
1663-
}
1664-
1665-
async function writeAsync(chunk, signal) {
1666-
if (errored) throw error;
1667-
if (closed || ending || writeEnded()) {
1668-
throw new ERR_INVALID_STATE('Writer is closed');
1669-
}
1670-
signal?.throwIfAborted();
1671-
1672-
if (writeConvertedSync(chunk)) return;
1673-
await writeWhenDrained(chunk, writeConvertedSync, signal);
1674-
}
1675-
1676-
async function writeWhenDrained(chunks, writeConverted, signal) {
1677-
if (pendingWrite !== null) {
1678-
throw new ERR_INVALID_STATE.RangeError(
1679-
'Backpressure violation: too many pending writes. ' +
1680-
'Await each write() call to respect backpressure.');
1681-
}
1682-
if (writeDesiredSize() !== 0) {
1683-
throw new ERR_INVALID_STATE('Stream write buffer is full');
1684-
}
1685-
1686-
const done = PromiseWithResolvers();
1687-
const token = { __proto__: null, done };
1688-
pendingWrite = token;
1689-
try {
1690-
while (true) {
1691-
await waitForDrain(signal);
1692-
if (errored) throw error;
1693-
if (closed || writeEnded()) {
1694-
throw new ERR_INVALID_STATE('Writer is closed');
1695-
}
1696-
signal?.throwIfAborted();
1697-
if (writeConverted(chunks, token)) return;
1698-
if (writeDesiredSize() !== 0) {
1699-
throw new ERR_INVALID_STATE('Stream write buffer is full');
1700-
}
1701-
}
1702-
} finally {
1703-
if (pendingWrite === token) pendingWrite = null;
1704-
done.resolve();
1705-
}
1706-
}
1707-
1708-
function writevConvertedSync(chunks, token) {
1709-
const isPending = pendingWrite === token && token !== undefined;
1710-
if (closed || errored || writeEnded() ||
1711-
(ending && !isPending) ||
1712-
(pendingWrite !== null && !isPending)) {
1713-
return false;
1714-
}
1715-
let len = 0;
1716-
for (const c of chunks) len += TypedArrayPrototypeGetByteLength(c);
1717-
if (len === 0) return true;
1718-
if (writeDesiredSize() === 0) return false;
1719-
const result = writeChunks(chunks);
1720-
if (result === undefined) return false;
1721-
totalBytesWritten += len;
1722-
return true;
1723-
}
1724-
1725-
function writevSync(chunks) {
1726-
return writevConvertedSync(convertChunks(chunks));
1727-
}
1728-
1729-
function writev(chunks, options = kEmptyObject) {
1730-
chunks = convertChunks(chunks);
1731-
const signal = getWriterSignal(options);
1732-
return writevAsync(chunks, signal);
1733-
}
1734-
1735-
async function writevAsync(chunks, signal) {
1736-
if (errored) throw error;
1737-
if (closed || ending || writeEnded()) {
1738-
throw new ERR_INVALID_STATE('Writer is closed');
1739-
}
1740-
signal?.throwIfAborted();
1741-
1742-
if (writevConvertedSync(chunks)) return;
1743-
await writeWhenDrained(chunks, writevConvertedSync, signal);
1744-
}
1745-
1746-
function endSync() {
1747-
// Per the streams/iter spec, endSync and end follow a try-fallback
1748-
// pattern. That is, callers should try endSync first and if it returns
1749-
// -1, then they should call and await end(). This is a signal that sync
1750-
// end is not currently possible. However, we always support sync end
1751-
// here unless the stream is already errored.
1752-
if (errored) return -1;
1753-
1754-
// If we're already closed, just return the total bytes written.
1755-
if (closed) return totalBytesWritten;
1756-
1757-
// Accepted writes and existing drain waiters must settle before ending.
1758-
if (ending || pendingWrite !== null || drainWakeup !== null) return -1;
1759-
1760-
// Fantastic, we can end synchronously!
1761-
endWrite();
1762-
closed = true;
1763-
return totalBytesWritten;
1764-
}
1765-
1766-
function end(options = kEmptyObject) {
1767-
const signal = getWriterSignal(options);
1768-
return endAsync(signal);
1769-
}
1770-
1771-
async function endAsync(signal) {
1772-
if (errored) throw error;
1773-
if (closed) return totalBytesWritten;
1774-
signal?.throwIfAborted();
1775-
1776-
// Per the streams/iter spec, endSync and end follow a try-fallback
1777-
// pattern. That is, callers should try endSync first and if it returns
1778-
// -1, then they should call and await end(). This is a signal that sync
1779-
// end is not currently possible. However, we always support sync end
1780-
// here unless the stream is already errored.
1781-
// While the user should have already called endSync, we call it again
1782-
// here to actually process the end request. At worst it's called twice.
1783-
const n = endSync();
1784-
1785-
// A return value of -1 indicates that endSync was not yet able to
1786-
// process the end request, either because we are errored or because we
1787-
// are awaiting drain. If we're errored, throw the error. If we're waiting
1788-
// for drain, await it and then try ending again.
1789-
1790-
if (n >= 0) return n;
1791-
if (errored) throw error;
1792-
1793-
if (pendingEnd === null) {
1794-
ending = true;
1795-
pendingEnd = finishEnd();
1796-
}
1797-
return raceWithSignal(pendingEnd, signal);
1798-
}
1799-
1800-
async function finishEnd() {
1801-
if (pendingWrite !== null) {
1802-
await pendingWrite.done.promise;
1803-
}
1804-
if (errored) throw error;
1805-
1806-
if (drainWakeup !== null) {
1807-
await drainWakeup.promise;
1808-
}
1809-
if (errored) throw error;
1810-
1811-
if (!writeEnded()) endWrite();
1812-
closed = true;
1813-
ending = false;
1814-
return totalBytesWritten;
1815-
}
1816-
1817-
function fail(reason) {
1818-
if (closed || errored) return;
1819-
errored = true;
1820-
error = reason;
1821-
resetStream(error);
1822-
if (drainWakeup != null) {
1823-
markPromiseAsHandled(drainWakeup.promise);
1824-
drainWakeup.reject(error);
1825-
drainWakeup = null;
1826-
}
1827-
}
1828-
1829-
const writer = {
1830-
__proto__: null,
1831-
get canWrite() {
1832-
if (closed || ending || errored || writeEnded()) {
1833-
return null;
1834-
}
1835-
return writeDesiredSize() > 0;
1836-
},
1837-
writeSync,
1838-
write,
1839-
writevSync,
1840-
writev,
1841-
endSync,
1842-
end,
1843-
fail,
1844-
[drainableProtocol]() {
1845-
if (closed || ending || errored) return null;
1846-
// If a drain is already pending, return the existing promise.
1847-
if (drainWakeup != null) return drainWakeup.promise;
1848-
if (writeDesiredSize() > 0) return null;
1849-
drainWakeup = PromiseWithResolvers();
1850-
return drainWakeup.promise;
1851-
},
1852-
[SymbolAsyncDispose]() {
1853-
if (ending) return pendingEnd;
1854-
if (!closed && !errored) fail();
1855-
return PromiseResolve();
1856-
},
1857-
[SymbolDispose]() {
1858-
if (!closed && !errored) fail();
1859-
},
1860-
};
1861-
1862-
// Non-writable stream - return a pre-closed writer.
1863-
// A remote unidirectional stream is read-only and has no writable
1864-
// side. isLocal distinguishes locally-initiated (writable) from
1865-
// remotely-initiated (read-only) uni streams.
1866-
if (nonWritableStream() || writeEnded()) {
1867-
closed = true;
1868-
return writer;
1869-
}
1870-
1871-
// Initialize the outbound DataQueue for streaming writes
1872-
initStreamingSource();
1873-
return writer;
1874-
}
1875-
18761549
// Functions used specifically for internal or assertion purposes only.
18771550
let getQuicStreamState;
18781551
let getQuicSessionState;

0 commit comments

Comments
 (0)