/**
*
* client
*
*/
import { GRAPHQL_TRANSPORT_WS_PROTOCOL } from './protocol.mjs';
import { MessageType, parseMessage, stringifyMessage, } from './message.mjs';
import { isObject } from './utils.mjs';
// this file is the entry point for browsers, re-export relevant elements
export * from './message.mjs';
export * from './protocol.mjs';
/** Creates a disposable GraphQL over WebSocket client. */
export function createClient(options) {
const { url, connectionParams, lazy = true, onNonLazyError = console.error, keepAlive = 0, retryAttempts = 5, retryWait = async function randomisedExponentialBackoff(retries) {
let retryDelay = 1000; // start with 1s delay
for (let i = 0; i < retries; i++) {
retryDelay *= 2;
}
await new Promise((resolve) => setTimeout(resolve, retryDelay +
// add random timeout from 300ms to 3s
Math.floor(Math.random() * (3000 - 300) + 300)));
}, isFatalConnectionProblem = (errOrCloseEvent) =>
// non `CloseEvent`s are fatal by default
!isLikeCloseEvent(errOrCloseEvent), on, webSocketImpl,
/**
* Generates a v4 UUID to be used as the ID using `Math`
* as the random number generator. Supply your own generator
* in case you need more uniqueness.
*
* Reference: https://stackoverflow.com/a/2117523/709884
*/
generateID = function generateUUID() {
return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (c) => {
const r = (Math.random() * 16) | 0, v = c == 'x' ? r : (r & 0x3) | 0x8;
return v.toString(16);
});
}, } = options;
let ws;
if (webSocketImpl) {
if (!isWebSocket(webSocketImpl)) {
throw new Error('Invalid WebSocket implementation provided');
}
ws = webSocketImpl;
}
else if (typeof WebSocket !== 'undefined') {
ws = WebSocket;
}
else if (typeof global !== 'undefined') {
ws =
global.WebSocket ||
// @ts-expect-error: Support more browsers
global.MozWebSocket;
}
else if (typeof window !== 'undefined') {
ws =
window.WebSocket ||
// @ts-expect-error: Support more browsers
window.MozWebSocket;
}
if (!ws) {
throw new Error('WebSocket implementation missing');
}
const WebSocketImpl = ws;
// websocket status emitter, subscriptions are handled differently
const emitter = (() => {
const listeners = {
connecting: (on === null || on === void 0 ? void 0 : on.connecting) ? [on.connecting] : [],
connected: (on === null || on === void 0 ? void 0 : on.connected) ? [on.connected] : [],
closed: (on === null || on === void 0 ? void 0 : on.closed) ? [on.closed] : [],
error: (on === null || on === void 0 ? void 0 : on.error) ? [on.error] : [],
};
return {
on(event, listener) {
const l = listeners[event];
l.push(listener);
return () => {
l.splice(l.indexOf(listener), 1);
};
},
emit(event, ...args) {
for (const listener of listeners[event]) {
// @ts-expect-error: The args should fit
listener(...args);
}
},
};
})();
let connecting, locks = 0, retrying = false, retries = 0, disposed = false;
async function connect() {
locks++;
const socket = await (connecting !== null && connecting !== void 0 ? connecting : (connecting = new Promise((resolve, reject) => (async () => {
if (retrying) {
await retryWait(retries);
retries++;
}
emitter.emit('connecting');
const socket = new WebSocketImpl(url, GRAPHQL_TRANSPORT_WS_PROTOCOL);
socket.onerror = (err) => {
// we let the onclose reject the promise for correct retry handling
emitter.emit('error', err);
};
socket.onclose = (event) => {
connecting = undefined;
emitter.emit('closed', event);
reject(event);
};
socket.onopen = async () => {
try {
socket.send(stringifyMessage({
type: MessageType.ConnectionInit,
payload: typeof connectionParams === 'function'
? await connectionParams()
: connectionParams,
}));
}
catch (err) {
socket.close(4400, err instanceof Error ? err.message : new Error(err).message);
}
};
socket.onmessage = ({ data }) => {
socket.onmessage = null; // interested only in the first message
try {
const message = parseMessage(data);
if (message.type !== MessageType.ConnectionAck) {
throw new Error(`First message cannot be of type ${message.type}`);
}
emitter.emit('connected', socket, message.payload); // connected = socket opened + acknowledged
retries = 0; // reset the retries on connect
resolve(socket);
}
catch (err) {
socket.close(4400, err instanceof Error ? err.message : new Error(err).message);
}
};
})())));
let release = () => {
// releases this connection lock
};
const released = new Promise((resolve) => (release = resolve));
return [
socket,
release,
Promise.race([
released.then(() => {
if (--locks === 0) {
// if no more connection locks are present, complete the connection
const complete = () => socket.close(1000, 'Normal Closure');
if (isFinite(keepAlive) && keepAlive > 0) {
// if the keepalive is set, allow for the specified calmdown time and
// then complete. but only if no lock got created in the meantime and
// if the socket is still open
setTimeout(() => {
if (!locks && socket.readyState === WebSocketImpl.OPEN)
complete();
}, keepAlive);
}
else {
// otherwise complete immediately
complete();
}
}
}),
new Promise((_resolve, reject) => socket.addEventListener('close', reject, { once: true })),
]),
];
}
/**
* Checks the `connect` problem and evaluates if the client should retry.
*/
function shouldRetryConnectOrThrow(errOrCloseEvent) {
// some close codes are worth reporting immediately
if (isLikeCloseEvent(errOrCloseEvent) &&
[
1002,
1011,
4400,
4401,
4409,
4429,
].includes(errOrCloseEvent.code)) {
throw errOrCloseEvent;
}
// disposed or normal closure (completed), shouldnt try again
if (disposed ||
(isLikeCloseEvent(errOrCloseEvent) && errOrCloseEvent.code === 1000)) {
return false;
}
// retries are not allowed or we tried to many times, report error
if (!retryAttempts || retries >= retryAttempts) {
throw errOrCloseEvent;
}
// throw fatal connection problems immediately
if (isFatalConnectionProblem(errOrCloseEvent)) {
throw errOrCloseEvent;
}
// looks good, start retrying
retrying = true;
return true;
}
// in non-lazy (hot?) mode always hold one connection lock to persist the socket
if (!lazy) {
(async () => {
for (;;) {
try {
const [, , waitForReleaseOrThrowOnClose] = await connect();
await waitForReleaseOrThrowOnClose;
return; // completed, shouldnt try again
}
catch (errOrCloseEvent) {
try {
if (!shouldRetryConnectOrThrow(errOrCloseEvent))
return onNonLazyError === null || onNonLazyError === void 0 ? void 0 : onNonLazyError(errOrCloseEvent);
}
catch (_a) {
// report thrown error, no further retries
return onNonLazyError === null || onNonLazyError === void 0 ? void 0 : onNonLazyError(errOrCloseEvent);
}
}
}
})();
}
// to avoid parsing the same message in each
// subscriber, we memo one on the last received data
let lastData, lastMessage;
function memoParseMessage(data) {
if (data !== lastData) {
lastMessage = parseMessage(data);
lastData = data;
}
return lastMessage;
}
return {
on: emitter.on,
subscribe(payload, sink) {
const id = generateID();
let completed = false;
const releaserRef = {
current: () => {
// for handling completions before connect
completed = true;
},
};
function messageHandler({ data }) {
const message = memoParseMessage(data);
switch (message.type) {
case MessageType.Next: {
if (message.id === id) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
sink.next(message.payload);
}
return;
}
case MessageType.Error: {
if (message.id === id) {
completed = true;
sink.error(message.payload);
releaserRef.current();
// TODO-db-201025 calling releaser will complete the sink, meaning that both the `error` and `complete` will be
// called. neither promises or observables care; once they settle, additional calls to the resolvers will be ignored
}
return;
}
case MessageType.Complete: {
if (message.id === id) {
completed = true;
releaserRef.current(); // release completes the sink
}
return;
}
}
}
(async () => {
for (;;) {
try {
const [socket, release, waitForReleaseOrThrowOnClose,] = await connect();
// if completed while waiting for connect, release the connection lock right away
if (completed)
return release();
socket.addEventListener('message', messageHandler);
socket.send(stringifyMessage({
id: id,
type: MessageType.Subscribe,
payload,
}));
releaserRef.current = () => {
if (!completed && socket.readyState === WebSocketImpl.OPEN) {
// if not completed already and socket is open, send complete message to server on release
socket.send(stringifyMessage({
id: id,
type: MessageType.Complete,
}));
}
release();
};
// either the releaser will be called, connection completed and
// the promise resolved or the socket closed and the promise rejected
await waitForReleaseOrThrowOnClose;
socket.removeEventListener('message', messageHandler);
return; // completed, shouldnt try again
}
catch (errOrCloseEvent) {
if (!shouldRetryConnectOrThrow(errOrCloseEvent))
return;
}
}
})()
.catch(sink.error)
.then(sink.complete); // resolves on release or normal closure
return () => releaserRef.current();
},
async dispose() {
disposed = true;
if (connecting) {
// if there is a connection, close it
const socket = await connecting;
socket.close(1000, 'Normal Closure');
}
},
};
}
function isLikeCloseEvent(val) {
return isObject(val) && 'code' in val && 'reason' in val;
}
function isWebSocket(val) {
return (typeof val === 'function' &&
'constructor' in val &&
'CLOSED' in val &&
'CLOSING' in val &&
'CONNECTING' in val &&
'OPEN' in val);
}