graphql over weboscket

client.js

import io from 'socket.io-client';
import { parse } from 'graphql';

let socket = null;
let queryId = 0;
const queryResolvers = new Map();
const queryInstancesCache = new Map();
const fragmentsCache = new Map();
const QueryStatus = {
    PENDING: 1,
    OK: 2,
    ERROR: 3
};

export const graphQLconnect = (host = window.location.hostname, port = 8080, protocol = 'http') => {
    socket = io.connect(`${protocol}://${host}:${port}`);

    socket.on('reconnect', () => {
        console.info('ws reconnected!');

        for(const [id, { query, variables }] of queryResolvers) {
            socket.emit('graphql/query', {
                data: { query, variables },
                id
            });
        }
    });

    socket.on('graphql/query/result', ({ id, data }) => {
        const {
            callbacks,
            mainOperation
        } = queryResolvers.get(id);

        for(const [clb] of callbacks) {
            let [resolve, reject] = clb;

            if (data.errors && typeof reject === "function") {
                reject(data.errors);
            }
            else if (typeof resolve === "function") resolve(data.data);
        }

        if (mainOperation !== 'subscription') {
            queryResolvers.delete(id);
        }
    });
};

class Query {
    constructor(query) {
        this.query = query;
        this.includeFragments = [];
        this.definedFragments = [];
        this.queryStatus = null;
        this.promise = null;
        this.mainOperation = null;
        this.variables = null;
        this.parse();
    }

    parse() {
        const { definitions } = parse(this.query);

        if (definitions.length === 1) {
            this.mainOperation = definitions[0].operation;
        }

        definitions.forEach(def => {
            if(def.kind === 'FragmentDefinition') {
                const { start, end } = def.loc;
                const fragment = this.query.substr(start, end - start);

                fragmentsCache.set(def.name.value, fragment);
                this.definedFragments.push(def.name.value);
            }

            if (def.selectionSet !== null) {
                this.findFragment(def.selectionSet.selections);
            }
        });

        this.includeFragments.forEach(fragmentName => {
            if (this.definedFragments.indexOf(fragmentName) !== -1) return;

            if (!fragmentsCache.has(fragmentName)) {
                console.error(`Fragment ${fragmentName} undefined`);
                return;
            }

            const fragment = fragmentsCache.get(fragmentName);
            this.query = `${fragment} \r\n ${this.query}`;
        });
    }

    findFragment(selections) {
        selections.forEach(sel => {
            if(sel.kind === 'FragmentSpread') {
                this.includeFragments.push(sel.name.value);
            }
            else if (sel.selectionSet !== null) {
                this.findFragment(sel.selectionSet.selections);
            }
        });
    }

    serVariables(variables) {
        this.variables = variables;
    }

    then(reject, resolve) {
        if (this.queryStatus === QueryStatus.PENDING) {
            this.promise.then(reject, resolve);
        }
        else {
            this.run().then(reject, resolve);
        }
        return this.promise;
    }

    catch(callback) {
        if (this.queryStatus === QueryStatus.PENDING) {
            this.promise.catch(callback);
        }
        else {
            this.run().catch(callback);
        }
        return this.promise;
    }

    run(variables = this.variables) {
        const callbacks = new Map();
        const id = ++queryId;
        const { query, mainOperation } = this;

        queryResolvers.set(id, { query, variables, callbacks, mainOperation, instance: this });

        this.promise = new Promise((resolver, reject) => {
            this.queryStatus = QueryStatus.PENDING;
            callbacks.set([resolver, reject]);

            socket.emit('graphql/query', {
                data: { query, variables },
                id
            });
            this.variables = null;
        });

        this.promise.onData = clb => {
            callbacks.set([clb]);

            return this.promise;
        };

        this.promise.then(data => {
            this.queryStatus = QueryStatus.OK;
            return data;
        }).catch(data => {
            this.queryStatus = QueryStatus.ERROR;
            return data;
        });

        return this.promise;
    }
}


export const gql = (query, variables) => {
    let instance = null;
    if (queryInstancesCache.has(query)) {
        instance = queryInstancesCache.get(query);
    }
    else {
        instance = new Query(query);
        queryInstancesCache.set(query, instance);
    }
    instance.serVariables(variables);
    return instance;
};

window.gq = gql;

export const unsubscribe = subName => {
    socket.emit("graphql/unsubscribe", {
        subName
    });
};

gistfile1.txt

import express from 'express';
import http from 'http';
import { connect } from '../src/gql/server';
import rootSchema from './rootSchema';
import cors from 'cors';

const app = express();
app.use(cors());

const server = http.Server(app);

server.listen(PORT);
connect(rootSchema, server);

server.js

import { graphql } from 'graphql';
import winston from 'winston';

class GQLWS {
    constructor(schema, http, ioConfig = {}) {
        this.schema = schema;
        this.http = http;
        this.ioConfig = ioConfig;
        this.init();

        this.onQuery = this.onQuery.bind(this);
        this.findSubByName = this.findSubByName.bind(this);
    }

    init() {
        this.io = require('socket.io')(this.http, this.ioConfig);
        winston.info(`[Start] ws://0.0.0.0:${this.port}`);

        this.io.on('connection', socket => {
            //socket.on('graphql/unsubscribe', data => this.onUnsubscribe(data.subName));
            socket.on('graphql/query', data => this.onQuery(data, socket));
        });
    }

    /*onUnsubscribe(subName) {
        winston.info(`[Unsubscribe] ${subName}`);
        const sub = this.findSubByName(subName);
        if(!sub || !sub.resolve) return;
        sub.resolve('unsubscribe');
    }*/

    onQuery({ id, data }, socket) {
        const {
            query,
            variables = {},
            operationName = null
        } = data;

        const rootValue = {
            resolve: data => {
                this.sendQueryResult(socket, { data }, id);
            },
            reject: err => {
                this.sendQueryError(socket, err, id);
            }
        };

        winston.info(`[Query] ${query} Variables: `, variables);

        graphql(
            this.schema,
            query,
            rootValue,
            undefined,
            variables,
            operationName
        ).then(result => {
            if (result === false) return;
            winston.info('[Result] ', result);
            this.sendQueryResult(socket, result, id);

        }).catch(result => {
            winston.info('[Query error] ', result);
            this.sendQueryError(socket, result, id);
        });
    }

    sendQueryError(socket, message, id) {
        socket.emit('graphql/query/result', {
            errors: [{
                message: message
            }],
            id
        });
    }

    sendQueryResult(socket, result, id) {
        socket.emit('graphql/query/result', {
            data: result,
            id: id
        });
    }

    findSubByName(name) {
        const schema = this.schema;

        if (!schema._subscriptionType || !name) return null;
        if (!schema._subscriptionType._fields) return null;
        if (!schema._subscriptionType._fields[name]) return null;

        return schema._subscriptionType._fields[name];
    }
}

export const connect = (schema, port, ioConfig) => {
    return new GQLWS(schema, port, ioConfig);
};


/*export const resolver = conf => {
    return (obj, args) => {
        try {
            if (obj === 'unsubscribe') {
                conf.unsubscribe(conf);
            }
            else {
                return conf.subscribe(obj, args, conf);
            }
        }
        catch(e) {
            obj.reject(e);
        }
    }
};*/