New Case Study:See how Anthropic automated 95% of dependency reviews with Socket.Learn More →
Socket
Sign inDemoInstall
Socket
Sign inDemoInstall

@enisdenjo/graphql-transport-ws

Package Overview
Dependencies
Maintainers
1
Versions
6
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

@enisdenjo/graphql-transport-ws - npm Package Compare versions

Comparing version 0.0.2 to 1.0.0

CHANGELOG.md
lib/message.d.ts
lib/message.js
lib/protocol.d.ts
lib/protocol.js
lib/server.d.ts
lib/server.js
lib/utils.d.ts
lib/utils.js
PROTOCOL.md

94

lib/client.d.ts
/**
*
* GraphQL subscriptions over the WebSocket Protocol
* GraphQL over WebSocket Protocol
*

@@ -8,83 +8,12 @@ * Check out the `PROTOCOL.md` document for the transport specification.

*/
/**
* The shape of a GraphQL response as dictated by the
* [spec](https://graphql.github.io/graphql-spec/June2018/#sec-Response-Format)
*/
export interface GraphQLResponseWithData {
data: Record<string, any>;
errors?: {
message: string;
locations?: Array<{
line: number;
column: number;
}>;
}[];
path?: string[] | number[];
}
export interface GraphQLResponseWithoutData {
data?: Record<string, any>;
errors: {
message: string;
locations?: Array<{
line: number;
column: number;
}>;
}[];
path?: Array<string | number>;
}
export interface GraphQLError {
message: string;
}
export declare type GraphQLResponse = GraphQLResponseWithData | GraphQLResponseWithoutData | GraphQLError;
/** Used to indicate that the requestId is missing. */
declare const NO_REQUEST_ID = "NRID";
/**
* Is the raw message being sent through the WebSocket connection.
* Since the ID generation is done automatically, we have 2 separate
* types for the two possible messages.
*/
export interface MessageWithoutID {
type: MessageType;
payload?: GraphQLResponse | null;
}
export interface Message extends MessageWithoutID {
/**
* The message ID (internally represented as the `requestId`).
* Can be missing in cases when managing the subscription
* connection itself.
*/
id: string | typeof NO_REQUEST_ID;
}
/** Types of messages allowed to be sent by the client/server over the WS protocol. */
export declare enum MessageType {
ConnectionInit = "connection_init",
ConnectionAck = "connection_ack",
ConnectionError = "connection_error",
ConnectionKeepAlive = "ka",
ConnectionTerminate = "connection_terminate",
Start = "start",
Data = "data",
Error = "error",
Complete = "complete",
Stop = "stop"
}
/** The payload used for starting GraphQL subscriptions. */
export interface StartPayload {
operationName?: string;
query: string;
variables: Record<string, any>;
}
/** The sink to communicate the subscription through. */
export interface Sink<T = any> {
next(value: T): void;
error(error: Error): void;
complete(): void;
readonly closed: boolean;
}
import { Sink, Disposable } from './types';
import { SubscribePayload } from './message';
/** Configuration used for the `create` client function. */
export interface Config {
export interface ClientOptions {
/** URL of the GraphQL server to connect. */
url: string;
connectionParams?: Record<string, any> | (() => Record<string, any>);
/** Optional parameters that the client specifies when establishing a connection with the server. */
connectionParams?: Record<string, unknown> | (() => Record<string, unknown>);
}
export interface Client {
export interface Client extends Disposable {
/**

@@ -95,8 +24,5 @@ * Subscribes through the WebSocket following the config parameters. It

*/
subscribe<T>(payload: StartPayload, sink: Sink<T>): () => void;
/** Disposes of all active subscriptions, closes the WebSocket client and frees up memory. */
dispose(): Promise<void>;
subscribe<T = unknown>(payload: SubscribePayload, sink: Sink<T>): () => void;
}
/** Creates a disposable GQL subscriptions client. */
export declare function createClient({ url, connectionParams }: Config): Client;
export {};
export declare function createClient(options: ClientOptions): Client;

443

lib/client.js
"use strict";
/**
*
* GraphQL subscriptions over the WebSocket Protocol
* GraphQL over WebSocket Protocol
*

@@ -9,241 +9,214 @@ * Check out the `PROTOCOL.md` document for the transport specification.

*/
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.createClient = exports.MessageType = void 0;
const websocket_as_promised_1 = __importDefault(require("websocket-as-promised"));
/** Used to indicate that the requestId is missing. */
const NO_REQUEST_ID = 'NRID';
function isNoRequestId(val) {
return val === NO_REQUEST_ID;
}
/** Types of messages allowed to be sent by the client/server over the WS protocol. */
var MessageType;
(function (MessageType) {
MessageType["ConnectionInit"] = "connection_init";
MessageType["ConnectionAck"] = "connection_ack";
MessageType["ConnectionError"] = "connection_error";
// NOTE: The keep alive message type does not follow the standard due to connection optimizations
MessageType["ConnectionKeepAlive"] = "ka";
MessageType["ConnectionTerminate"] = "connection_terminate";
MessageType["Start"] = "start";
MessageType["Data"] = "data";
MessageType["Error"] = "error";
MessageType["Complete"] = "complete";
MessageType["Stop"] = "stop";
})(MessageType = exports.MessageType || (exports.MessageType = {}));
/** Checks if the value has a shape of a `Message`. */
function isMessage(val) {
if (typeof val !== 'object' || val == null) {
return false;
}
// TODO-db-200603 validate the type
if ('type' in val && Boolean(val.type)) {
return true;
}
return false;
}
/** Checks if the value has a shape of a `GraphQLResponse`. */
function isGraphQLResponse(val) {
if (typeof val !== 'object' || val == null) {
return false;
}
if (
// GraphQLResponseWithData
'data' in val ||
// GraphQLResponseWithoutData
'errors' in val ||
// GraphQLError
('message' in val && Object.keys(val).length === 1)) {
return true;
}
return false;
}
exports.createClient = void 0;
const protocol_1 = require("./protocol");
const message_1 = require("./message");
const utils_1 = require("./utils");
/** Creates a disposable GQL subscriptions client. */
function createClient({ url, connectionParams }) {
const ws = new websocket_as_promised_1.default(url, {
timeout: 2 * 1000,
createWebSocket: (url) => new WebSocket(url, 'graphql-ws'),
packMessage: (data) => JSON.stringify(data),
unpackMessage: (data) => {
if (typeof data !== 'string') {
throw new Error(`Unsupported message data type ${typeof data}`);
}
return JSON.parse(data);
},
// omits when receiving a no request id symbol to avoid confusion and reduce message size
attachRequestId: (data, requestId) => {
if (isNoRequestId(requestId)) {
return data;
}
return Object.assign(Object.assign({}, data), { id: String(requestId) });
},
// injecting no request id symbol allows us to request/response on id-less messages
extractRequestId: (data) => { var _a; return (_a = data === null || data === void 0 ? void 0 : data.id) !== null && _a !== void 0 ? _a : NO_REQUEST_ID; },
});
// connects on demand, already open connections are ignored
let isConnected = false, isConnecting = false, isDisconnecting = false;
async function waitForConnected() {
let waitedTimes = 0;
while (!isConnected) {
await new Promise((resolve) => setTimeout(resolve, 100));
// 100ms * 100 = 10s
if (waitedTimes >= 100) {
throw new Error('Waited 10 seconds but socket never connected.');
}
waitedTimes++;
}
function createClient(options) {
const { url, connectionParams } = options;
// holds all currently subscribed sinks, will use this map
// to dispatch messages to the correct destination
const subscribedSinks = {};
function errorAllSinks(err) {
Object.entries(subscribedSinks).forEach(([, sink]) => sink.error(err));
}
async function waitForDisconnected() {
let waitedTimes = 0;
while (isConnected) {
await new Promise((resolve) => setTimeout(resolve, 100));
// 100ms * 100 = 10s
if (waitedTimes >= 100) {
throw new Error('Waited 10 seconds but socket never disconnected.');
}
waitedTimes++;
}
function completeAllSinks() {
Object.entries(subscribedSinks).forEach(([, sink]) => sink.complete());
}
// Lazily uses the socket singleton to establishes a connection described by the protocol.
let socket = null, connected = false, connecting = false;
async function connect() {
if (isConnected)
if (connected) {
return;
if (isConnecting) {
return waitForConnected();
}
if (isDisconnecting) {
await waitForDisconnected();
if (connecting) {
let waitedTimes = 0;
while (!connected) {
await new Promise((resolve) => setTimeout(resolve, 100));
// 100ms * 50 = 5sec
if (waitedTimes >= 50) {
throw new Error('Waited 10 seconds but socket never connected');
}
waitedTimes++;
}
// connected === true
return;
}
// open and initialize a connection, send the start message and flag as connected
isConnected = false;
isConnecting = true;
await ws.open();
const ack = await request(MessageType.ConnectionInit, connectionParams && typeof connectionParams === 'function'
? connectionParams()
: connectionParams, NO_REQUEST_ID);
if (ack.type !== MessageType.ConnectionAck) {
await ws.close();
throw new Error('Connection not acknowledged');
}
isConnecting = false;
isConnected = true;
connected = false;
connecting = true;
return new Promise((resolve, reject) => {
let done = false; // used to avoid resolving/rejecting the promise multiple times
socket = new WebSocket(url, protocol_1.GRAPHQL_TRANSPORT_WS_PROTOCOL);
/**
* `onerror` handler is unnecessary because even if an error occurs, the `onclose` handler will be called
*
* From: https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_client_applications
* > If an error occurs while attempting to connect, first a simple event with the name error is sent to the
* > WebSocket object (thereby invoking its onerror handler), and then the CloseEvent is sent to the WebSocket
* > object (thereby invoking its onclose handler) to indicate the reason for the connection's closing.
*/
socket.onclose = ({ code, reason }) => {
const err = new Error(`Socket closed with event ${code}` + !reason ? '' : `: ${reason}`);
if (code === 1000 || code === 1001) {
// close event `1000: Normal Closure` is ok and so is `1001: Going Away` (maybe the server is restarting)
completeAllSinks();
}
else {
// all other close events are considered erroneous
errorAllSinks(err);
}
if (!done) {
done = true;
connecting = false;
connected = false; // the connection is lost
socket = null;
reject(err); // we reject here bacause the close is not supposed to be called during the connect phase
}
};
socket.onopen = () => {
try {
if (!socket) {
throw new Error('Opened a socket on nothing');
}
socket.send(message_1.stringifyMessage({
type: message_1.MessageType.ConnectionInit,
payload: typeof connectionParams === 'function'
? connectionParams()
: connectionParams,
}));
}
catch (err) {
errorAllSinks(err);
if (!done) {
done = true;
connecting = false;
if (socket) {
socket.close();
socket = null;
}
reject(err);
}
}
};
function handleMessage({ data }) {
try {
if (!socket) {
throw new Error('Received a message on nothing');
}
const message = message_1.parseMessage(data);
if (message.type !== message_1.MessageType.ConnectionAck) {
throw new Error(`First message cannot be of type ${message.type}`);
}
// message.type === MessageType.ConnectionAck
if (!done) {
done = true;
connecting = false;
connected = true; // only now is the connection ready
resolve();
}
}
catch (err) {
errorAllSinks(err);
if (!done) {
done = true;
connecting = false;
if (socket) {
socket.close();
socket = null;
}
reject(err);
}
}
finally {
if (socket) {
// this listener is not necessary anymore
socket.removeEventListener('message', handleMessage);
}
}
}
socket.addEventListener('message', handleMessage);
});
}
// disconnects on demand, already closed connections are ignored
async function disconnect() {
isDisconnecting = true;
if (isConnected) {
// sends a terminate message, then closes the websocket
send(MessageType.ConnectionTerminate);
}
await ws.close();
isDisconnecting = false;
isConnected = false;
}
// holds all currently subscribed sinks, will use this map
// to dispatch messages to the correct destination and
// as a decision system on when to unsubscribe
const requestIdSink = {};
function messageForSinkWithRequestId(requestId, message) {
let hasCompleted = false;
Object.entries(requestIdSink).some(([sinkRequestId, sink]) => {
if (requestId === sinkRequestId) {
return {
subscribe: (payload, sink) => {
const uuid = generateUUID();
if (subscribedSinks[uuid]) {
sink.error(new Error(`Sink with ID ${uuid} already registered`));
return utils_1.noop;
}
subscribedSinks[uuid] = sink;
function handleMessage({ data }) {
const message = message_1.parseMessage(data);
switch (message.type) {
case MessageType.Data: {
const err = checkServerPayload(message.payload);
if (err) {
sink.error(err);
hasCompleted = true;
return true;
case message_1.MessageType.Next: {
if (message.id === uuid) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
sink.next(message.payload);
}
sink.next(message.payload);
break;
}
case MessageType.Error: {
const err = checkServerPayload(message.payload);
if (err) {
sink.error(err);
case message_1.MessageType.Error: {
if (message.id === uuid) {
sink.error(message.payload);
}
else {
sink.error(new Error('Unkown error received from the subscription server'));
}
hasCompleted = true;
break;
}
case MessageType.Complete: {
sink.complete();
hasCompleted = true;
case message_1.MessageType.Complete: {
if (message.id === uuid) {
sink.complete();
}
break;
}
}
return true;
}
return false;
});
// if the sink got completed, remove it from the subscribed sinks
if (hasCompleted) {
delete requestIdSink[requestId];
}
// if there are no subscriptions left over, disconnect
if (Object.keys(requestIdSink).length === 0) {
// TODO-db-200603 report possible errors on disconnect
disconnect();
}
}
function errorAllSinks(error) {
Object.entries(requestIdSink).forEach(([, sink]) => sink.error(error));
}
// listens exclusively to messages with matching request ids
function responseListener(data) {
if (!isMessage(data)) {
return errorAllSinks(new Error('Received an invalid message from the subscription server'));
}
messageForSinkWithRequestId(data.id, data);
}
ws.onResponse.addListener(responseListener);
function subscribe(payload, sink) {
// generate a unique request id for this subscription
const requestId = randomString();
if (requestIdSink[requestId]) {
sink.error(new Error(`Sink already registered for ID: ${requestId}`));
return () => { };
}
requestIdSink[requestId] = sink;
connect()
// start the subscription on a connection
.then(() => send(MessageType.Start, payload, requestId))
// will also error this sink because its added to the map above
.catch(errorAllSinks);
return () => {
connect()
// stop the subscription, after the server acknowledges this the sink will complete
.then(() => send(MessageType.Stop, undefined, requestId))
// will also error this sink because its added to the map above
.catch(errorAllSinks);
};
}
function send(type, payload, requestId) {
if (requestId) {
return ws.sendPacked({ id: requestId, type, payload });
}
return ws.sendPacked({ type, payload });
}
async function request(type, payload, requestId) {
return await ws.sendRequest({ type, payload }, { requestId });
}
return {
subscribe(payload, sink) {
return subscribe(payload, sink);
(async () => {
try {
await connect();
if (!socket) {
throw new Error('Socket connected but empty');
}
socket.addEventListener('message', handleMessage);
socket.send(message_1.stringifyMessage({
id: uuid,
type: message_1.MessageType.Subscribe,
payload,
}));
}
catch (err) {
sink.error(err);
}
})();
return () => {
if (socket) {
socket.send(message_1.stringifyMessage({
id: uuid,
type: message_1.MessageType.Complete,
}));
socket.removeEventListener('message', handleMessage);
// equal to 1 because this sink is the last one.
// the deletion from the map happens afterwards
if (Object.entries(subscribedSinks).length === 1) {
socket.close(1000, 'Normal Closure');
socket = null;
}
}
sink.complete();
delete subscribedSinks[uuid];
};
},
async dispose() {
dispose: async () => {
// complete all sinks
Object.entries(requestIdSink).forEach(([, sink]) => sink.complete());
// remove all subscriptions
Object.keys(requestIdSink).forEach((key) => {
delete requestIdSink[key];
// TODO-db-200817 complete or error? the sinks should be completed BEFORE the client gets disposed
completeAllSinks();
// delete all sinks
Object.keys(subscribedSinks).forEach((uuid) => {
delete subscribedSinks[uuid];
});
// remove all listeners
ws.removeAllListeners();
// do disconnect
return disconnect();
// if there is an active socket, close it with a normal closure
if (socket && socket.readyState === WebSocket.OPEN) {
// TODO-db-200817 decide if `1001: Going Away` should be used instead
socket.close(1000, 'Normal Closure');
socket = null;
}
},

@@ -253,26 +226,16 @@ };

exports.createClient = createClient;
/**
* Takes in the payload received from the server, parses and validates it,
* checks for errors and returns a single error for all problematic cases.
*/
function checkServerPayload(payload) {
if (!payload) {
return new Error('Received empty payload from the subscription server');
/** Generates a new v4 UUID. Reference: https://stackoverflow.com/a/2117523/709884 */
function generateUUID() {
if (!window.crypto) {
// fallback to Math.random when crypto is not available
return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function (c) {
const r = (Math.random() * 16) | 0, v = c == 'x' ? r : (r & 0x3) | 0x8;
return v.toString(16);
});
}
if (!isGraphQLResponse(payload)) {
return new Error('Received invalid payload structure from the subscription server');
}
if ('errors' in payload && payload.errors) {
return new Error(payload.errors.map(({ message }) => message).join(', '));
}
if (Object.keys(payload).length === 1 &&
'message' in payload &&
payload.message) {
return new Error(payload.message);
}
return null;
return '10000000-1000-4000-8000-100000000000'.replace(/[018]/g, (s) => {
const c = Number.parseInt(s, 10);
return (c ^
(window.crypto.getRandomValues(new Uint8Array(1))[0] & (15 >> (c / 4)))).toString(16);
});
}
/** randomString does exactly what the name says. */
function randomString() {
return Math.random().toString(36).substr(2, 6);
}

41

package.json
{
"name": "@enisdenjo/graphql-transport-ws",
"version": "0.0.2",
"version": "1.0.0",
"description": "A WebSocket client for GraphQL subscriptions",

@@ -17,14 +17,24 @@ "license": "MIT",

"README.md",
"LICENSE.md"
"LICENSE.md",
"PROTOCOL.md"
],
"publishConfig": {
"access": "public"
},
"scripts": {
"gendocs": "typedoc --options typedoc.js src/",
"lint": "eslint 'src'",
"type-check": "tsc --noEmit",
"test": "jest",
"build": "tsc -b"
"test": "jest -i",
"build": "tsc -b",
"release": "semantic-release"
},
"peerDependencies": {
"graphql": ">=15.0.0"
},
"dependencies": {
"websocket-as-promised": "^1.0.1"
"ws": "^7.3.1"
},
"devDependencies": {
"@babel/core": "^7.11.0",
"@babel/core": "^7.11.1",
"@babel/plugin-proposal-class-properties": "^7.10.4",

@@ -36,8 +46,21 @@ "@babel/plugin-proposal-nullish-coalescing-operator": "^7.10.4",

"@babel/preset-typescript": "^7.10.4",
"@types/jest": "^26.0.8",
"babel-jest": "^26.2.2",
"jest": "^26.2.2",
"@semantic-release/changelog": "^5.0.1",
"@semantic-release/git": "^9.0.0",
"@types/jest": "^26.0.9",
"@types/ws": "^7.2.6",
"@typescript-eslint/eslint-plugin": "^3.9.0",
"@typescript-eslint/parser": "^3.9.0",
"babel-jest": "^26.3.0",
"eslint": "^7.6.0",
"eslint-config-prettier": "^6.11.0",
"eslint-plugin-prettier": "^3.1.4",
"graphql": "^15.3.0",
"graphql-subscriptions": "^1.1.0",
"jest": "^26.3.0",
"prettier": "^2.0.5",
"semantic-release": "^17.1.1",
"typedoc": "^0.18.0",
"typedoc-plugin-markdown": "^2.4.1",
"typescript": "^3.9.7"
}
}

61

README.md
# graphql-transport-ws
**Work in progress!**
> A coherent, zero-dependency, lazy, simple and easy to use server and client implementation of the [GraphQL over WebSocket Protocol](PROTODCOL.md).
A client for GraphQL subscriptions over WebSocket. _Server implementation coming soon!_
## Getting started

@@ -17,4 +15,6 @@

### Relay
### Examples
#### Client usage with [Relay](https://relay.dev)
```ts

@@ -63,9 +63,56 @@ import { createClient } from '@enisdenjo/graphql-transport-ws';

#### Client usage with [Apollo](https://www.apollographql.com)
```typescript
import { print } from 'graphql';
import { ApolloLink, Operation, FetchResult, Observable } from '@apollo/client';
import { createClient, Config, Client } from '@enisdenjo/graphql-transport-ws';
class WebSocketLink extends ApolloLink {
private client: Client;
constructor(config: Config) {
super();
this.client = createClient(config);
}
public request({
operationName,
query,
variables,
}: Operation): Observable<FetchResult> {
return new Observable((sink) => {
return this.client.subscribe<FetchResult>(
{ operationName, query: print(query), variables },
sink,
);
});
}
}
const link = new WebSocketLink({
url: 'wss://some.url/graphql',
connectionParams: () => {
const session = getSession();
if (!session) {
return null;
}
return {
Authorization: `Bearer ${session.token}`,
};
},
});
```
## Documentation
[TypeDoc](https://typedoc.org) generated documentation is located in the [docs folder](docs/).
## Protocol
Read more about it in the [PROTOCOL.md](PROTOCOL.md)
Read about the exact transport protocol used by the library in the [PROTOCOL.md](PROTOCOL.md) document.
## Want to help?
Want to file a bug, contribute some code, or improve documentation? Excellent! Read up on our
guidelines for [contributing](CONTRIBUTING.md).
File a bug, contribute with code, or improve documentation? Welcome 👋!
Read up on our guidelines for [contributing](CONTRIBUTING.md).
lib/tests/client.d.ts
lib/tests/client.js
SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc