index.js

/*
    This file is part of web3.js.
    web3.js is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
    web3.js is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.
    You should have received a copy of the GNU Lesser General Public License
    along with web3.js.  If not, see <http://www.gnu.org/licenses/>.
*/
/**
 * @file index.js
 * @author Fabian Vogelsteller <fabian@ethereum.org>
 * @date 2017
 */

"use strict";


const { callbackify } = require('util');
var errors = require('web3-core-helpers').errors;
var Jsonrpc = require('./jsonrpc.js');
var BatchManager = require('./batch.js');
var givenProvider = require('./givenProvider.js');

/**
 * It's responsible for passing messages to providers
 * It's also responsible for polling the ethereum node for incoming messages
 * Default poll timeout is 1 second
 * Singleton
 *
 * @param {string|Object}provider
 * @param {Net.Socket} net
 *
 * @constructor
 */
var RequestManager = function RequestManager(provider, net) {
    this.provider = null;
    this.providers = RequestManager.providers;

    this.setProvider(provider, net);
    this.subscriptions = new Map();
};


RequestManager.givenProvider = givenProvider;

RequestManager.providers = {
    WebsocketProvider: require('web3-providers-ws'),
    HttpProvider: require('web3-providers-http'),
    IpcProvider: require('web3-providers-ipc')
};


/**
 * Should be used to set provider of request manager
 *
 * @method setProvider
 *
 * @param {Object} provider
 * @param {net.Socket} net
 *
 * @returns void
 */
RequestManager.prototype.setProvider = function (provider, net) {
    var _this = this;

    // autodetect provider
    if (provider && typeof provider === 'string' && this.providers) {

        // HTTP
        if (/^http(s)?:\/\//i.test(provider)) {
            provider = new this.providers.HttpProvider(provider);

            // WS
        } else if (/^ws(s)?:\/\//i.test(provider)) {
            provider = new this.providers.WebsocketProvider(provider);

            // IPC
        } else if (provider && typeof net === 'object' && typeof net.connect === 'function') {
            provider = new this.providers.IpcProvider(provider, net);

        } else if (provider) {
            throw new Error('Can\'t autodetect provider for "' + provider + '"');
        }
    }


    // reset the old one before changing, if still connected
    if (this.provider && this.provider.connected)
        this.clearSubscriptions();

    this.provider = provider || null;

    // listen to incoming notifications
    if (this.provider && this.provider.on) {
        if (typeof provider.request === 'function') { // EIP-1193 provider
            this.provider.on('message', function (payload) {
                if (payload && payload.type === 'eth_subscription' && payload.data) {
                    const data = payload.data
                    if (data.subscription && _this.subscriptions.has(data.subscription)) {
                        _this.subscriptions.get(data.subscription).callback(null, data.result);
                    }
                }
            })
        } else { // legacy provider subscription event
            this.provider.on('data', function data(result, deprecatedResult) {
                result = result || deprecatedResult; // this is for possible old providers, which may had the error first handler

                // if result is a subscription, call callback for that subscription
                if (result.method && result.params && result.params.subscription && _this.subscriptions.has(result.params.subscription)) {
                    _this.subscriptions.get(result.params.subscription).callback(null, result.params.result);
                }
            });
        }

        // resubscribe if the provider has reconnected
        this.provider.on('connect', function connect() {
            _this.subscriptions.forEach(function (subscription) {
                subscription.subscription.resubscribe();
            });
        });

        // notify all subscriptions about the error condition
        this.provider.on('error', function error(error) {
            _this.subscriptions.forEach(function (subscription) {
                subscription.callback(error);
            });
        });

        // notify all subscriptions about bad close conditions
        const disconnect = function disconnect(event) {
            if (!_this._isCleanCloseEvent(event) || _this._isIpcCloseError(event)) {
                _this.subscriptions.forEach(function (subscription) {
                    subscription.callback(errors.ConnectionCloseError(event));
                    _this.subscriptions.delete(subscription.subscription.id);
                });

                if (_this.provider && _this.provider.emit) {
                    _this.provider.emit('error', errors.ConnectionCloseError(event));
                }
            }
            if (_this.provider && _this.provider.emit) {
                _this.provider.emit('end', event);
            }
        };
        this.provider.on('disconnect', disconnect);

        // TODO add end, timeout??
    }
};

/**
 * Asynchronously send request to provider.
 * Prefers to use the `request` method available on the provider as specified in [EIP-1193](https://eips.ethereum.org/EIPS/eip-1193).
 * If `request` is not available, falls back to `sendAsync` and `send` respectively.
 * @method send
 * @param {Object} data
 * @param {Function} callback
 */
RequestManager.prototype.send = function (data, callback) {
    callback = callback || function () { };

    if (!this.provider) {
        return callback(errors.InvalidProvider());
    }

    const { method, params } = data

    const jsonrpcPayload = Jsonrpc.toPayload(method, params);
    const jsonrpcResultCallback = this._jsonrpcResultCallback(callback, jsonrpcPayload)

    if (this.provider.request) {
        const callbackRequest = callbackify(this.provider.request.bind(this.provider))
        const requestArgs = { method, params }
        callbackRequest(requestArgs, callback);
    } else if (this.provider.sendAsync) {
        this.provider.sendAsync(jsonrpcPayload, jsonrpcResultCallback);
    } else if (this.provider.send) {
        this.provider.send(jsonrpcPayload, jsonrpcResultCallback);
    } else {
        throw new Error('Provider does not have a request or send method to use.');
    }
};

/**
 * Asynchronously send batch request.
 * Only works if provider supports batch methods through `sendAsync` or `send`.
 * @method sendBatch
 * @param {Array} data - array of payload objects
 * @param {Function} callback
 */
RequestManager.prototype.sendBatch = function (data, callback) {
    if (!this.provider) {
        return callback(errors.InvalidProvider());
    }

    var payload = Jsonrpc.toBatchPayload(data);
    this.provider[this.provider.sendAsync ? 'sendAsync' : 'send'](payload, function (err, results) {
        if (err) {
            return callback(err);
        }

        if (!Array.isArray(results)) {
            return callback(errors.InvalidResponse(results));
        }

        callback(null, results);
    });
};


/**
 * Waits for notifications
 *
 * @method addSubscription
 * @param {Subscription} subscription         the subscription
 * @param {String} type         the subscription namespace (eth, personal, etc)
 * @param {Function} callback   the callback to call for incoming notifications
 */
RequestManager.prototype.addSubscription = function (subscription, callback) {
    if (this.provider.on) {
        this.subscriptions.set(
            subscription.id,
            {
                callback: callback,
                subscription: subscription
            }
        );
    } else {
        throw new Error('The provider doesn\'t support subscriptions: ' + this.provider.constructor.name);
    }
};

/**
 * Waits for notifications
 *
 * @method removeSubscription
 * @param {String} id           the subscription id
 * @param {Function} callback   fired once the subscription is removed
 */
RequestManager.prototype.removeSubscription = function (id, callback) {
    if (this.subscriptions.has(id)) {
        var type = this.subscriptions.get(id).subscription.options.type;

        // remove subscription first to avoid reentry
        this.subscriptions.delete(id);

        // then, try to actually unsubscribe
        this.send({
            method: type + '_unsubscribe',
            params: [id]
        }, callback);

        return;
    }

    if (typeof callback === 'function') {
        // call the callback if the subscription was already removed
        callback(null);
    }
};

/**
 * Should be called to reset the subscriptions
 *
 * @method reset
 *
 * @returns {boolean}
 */
RequestManager.prototype.clearSubscriptions = function (keepIsSyncing) {
    try {
        var _this = this;

        // uninstall all subscriptions
        if (this.subscriptions.size > 0) {
            this.subscriptions.forEach(function (value, id) {
                if (!keepIsSyncing || value.name !== 'syncing')
                    _this.removeSubscription(id);
            });
        }

        //  reset notification callbacks etc.
        if (this.provider.reset)
            this.provider.reset();

        return true
    } catch (e) {
        throw new Error(`Error while clearing subscriptions: ${e}`)
    }
};

/**
 * Evaluates WS close event
 *
 * @method _isCleanClose
 *
 * @param {CloseEvent | boolean} event WS close event or exception flag
 *
 * @returns {boolean}
 */
RequestManager.prototype._isCleanCloseEvent = function (event) {
    return typeof event === 'object' && ([1000].includes(event.code) || event.wasClean === true);
};

/**
 * Detects Ipc close error. The node.net module emits ('close', isException)
 *
 * @method _isIpcCloseError
 *
 * @param {CloseEvent | boolean} event WS close event or exception flag
 *
 * @returns {boolean}
 */
RequestManager.prototype._isIpcCloseError = function (event) {
    return typeof event === 'boolean' && event;
};

/**
 * The jsonrpc result callback for RequestManager.send
 *
 * @method _jsonrpcResultCallback
 *
 * @param {Function} callback the callback to use
 * @param {Object} payload the jsonrpc payload
 *
 * @returns {Function} return callback of form (err, result)
 *
 */
RequestManager.prototype._jsonrpcResultCallback = function (callback, payload) {
    return function (err, result) {
        if (result && result.id && payload.id !== result.id) {
            return callback(new Error(`Wrong response id ${result.id} (expected: ${payload.id}) in ${JSON.stringify(payload)}`));
        }

        if (err) {
            return callback(err);
        }

        if (result && result.error) {
            return callback(errors.ErrorResponse(result));
        }

        if (!Jsonrpc.isValidResponse(result)) {
            return callback(errors.InvalidResponse(result));
        }

        callback(null, result.result);
    }
};

module.exports = {
    Manager: RequestManager,
    BatchManager: BatchManager
};