ravendb/dist/Http/RequestExecutor.js

"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
Object.defineProperty(exports, "__esModule", { value: true });
const os = require("os");
const BluebirdPromise = require("bluebird");
const semaphore = require("semaphore");
const SemaphoreUtil_1 = require("../Utility/SemaphoreUtil");
const LogUtil_1 = require("../Utility/LogUtil");
const Timer_1 = require("../Primitives/Timer");
const ServerNode_1 = require("./ServerNode");
const Topology_1 = require("./Topology");
const GetDatabaseTopologyCommand_1 = require("../ServerWide/Commands/GetDatabaseTopologyCommand");
const StatusCode_1 = require("./StatusCode");
const NodeSelector_1 = require("./NodeSelector");
const Certificate_1 = require("../Auth/Certificate");
const HttpCache_1 = require("./HttpCache");
const Exceptions_1 = require("../Exceptions");
const GetClientConfigurationOperation_1 = require("../Documents/Operations/Configuration/GetClientConfigurationOperation");
const Constants_1 = require("../Constants");
const Stopwatch_1 = require("../Utility/Stopwatch");
const PromiseUtil = require("../Utility/PromiseUtil");
const GetStatisticsOperation_1 = require("../Documents/Operations/GetStatisticsOperation");
const TypeUtil_1 = require("../Utility/TypeUtil");
const Serializer_1 = require("../Mapping/Json/Serializer");
const UriUtil_1 = require("../Utility/UriUtil");
const StreamUtil = require("../Utility/StreamUtil");
const HttpUtil_1 = require("../Utility/HttpUtil");
const PromiseUtil_1 = require("../Utility/PromiseUtil");
const DEFAULT_REQUEST_OPTIONS = {};
const log = LogUtil_1.getLogger({ module: "RequestExecutor" });
class IndexAndResponse {
    constructor(index, response) {
        this.index = index;
        this.response = response;
    }
}
class NodeStatus {
    constructor(nodeIndex, node, nodeStatusCallback) {
        this.nodeIndex = nodeIndex;
        this.node = node;
        this._timerPeriodInMs = 100;
        this._nodeStatusCallback = nodeStatusCallback;
    }
    _nextTimerPeriod() {
        if (this._timerPeriodInMs <= 5000) {
            return 5000;
        }
        this._timerPeriodInMs = this._timerPeriodInMs + 100;
        return this._timerPeriodInMs;
    }
    startTimer() {
        const that = this;
        this._timer = new Timer_1.Timer(function timerActionNodeStatusCallback() {
            return that._nodeStatusCallback(that);
        }, this._timerPeriodInMs);
    }
    updateTimer() {
        this._timer.change(this._nextTimerPeriod());
    }
    dispose() {
        this._timer.dispose();
    }
}
exports.NodeStatus = NodeStatus;
class RequestExecutor {
    constructor(database, authOptions, conventions) {
        this._updateDatabaseTopologySemaphore = semaphore();
        this._updateClientConfigurationSemaphore = semaphore();
        this._failedNodesTimers = new Map();
        this._certificate = null;
        this.aggressiveCaching = null;
        this.numberOfServerRequests = 0;
        this._clientConfigurationEtag = 0;
        this._topologyEtag = 0;
        this._log = LogUtil_1.getLogger({
            module: `${this.constructor.name}-${Math.floor(Math.random() * 10000)}`
        });
        this._cache = new HttpCache_1.HttpCache(conventions.maxHttpCacheSize);
        this._readBalanceBehavior = conventions.readBalanceBehavior;
        this._databaseName = database;
        this._lastReturnedResponse = new Date();
        this._conventions = conventions.clone();
        this._authOptions = authOptions;
        this._certificate = Certificate_1.Certificate.createFromOptions(this._authOptions);
        this._setDefaultRequestOptions();
    }
    get _firstTopologyUpdatePromise() {
        return this._firstTopologyUpdatePromiseInternal;
    }
    set _firstTopologyUpdatePromise(value) {
        this._firstTopologyUpdatePromiseInternal = value;
        if (value) {
            this._firstTopologyUpdateStatus = PromiseUtil_1.PromiseStatusTracker.track(value);
        }
    }
    get customHttpRequestOptions() {
        return this._customHttpRequestOptions;
    }
    set customHttpRequestOptions(value) {
        this._customHttpRequestOptions = value;
        this._setDefaultRequestOptions();
    }
    getAuthOptions() {
        return this._authOptions;
    }
    getTopologyEtag() {
        return this._topologyEtag;
    }
    get conventions() {
        return this._conventions;
    }
    getClientConfigurationEtag() {
        return this._clientConfigurationEtag;
    }
    get cache() {
        return this._cache;
    }
    get disposed() {
        return this._disposed;
    }
    getUrl() {
        if (!this._nodeSelector) {
            return null;
        }
        const preferredNode = this._nodeSelector.getPreferredNode();
        return preferredNode
            ? preferredNode.currentNode.url
            : null;
    }
    getTopology() {
        return this._nodeSelector
            ? this._nodeSelector.getTopology()
            : null;
    }
    getTopologyNodes() {
        const topology = this.getTopology();
        return topology
            ? [...topology.nodes]
            : null;
    }
    static create(initialUrls, database, opts) {
        const { authOptions, documentConventions } = opts || {};
        const executor = new RequestExecutor(database, authOptions, documentConventions);
        executor._firstTopologyUpdatePromise = executor._firstTopologyUpdate(initialUrls);
        executor._firstTopologyUpdatePromise.catch(TypeUtil_1.TypeUtil.NOOP);
        return executor;
    }
    static createForSingleNodeWithConfigurationUpdates(url, database, opts) {
        const executor = this.createForSingleNodeWithoutConfigurationUpdates(url, database, opts);
        executor._disableClientConfigurationUpdates = false;
        return executor;
    }
    static createForSingleNodeWithoutConfigurationUpdates(url, database, opts) {
        const { authOptions, documentConventions } = opts;
        const initialUrls = RequestExecutor._validateUrls([url], authOptions);
        const executor = new RequestExecutor(database, authOptions, documentConventions);
        const topology = new Topology_1.Topology();
        topology.etag = -1;
        const serverNode = new ServerNode_1.ServerNode({
            url: initialUrls[0],
            database
        });
        topology.nodes = [serverNode];
        executor._nodeSelector = new NodeSelector_1.NodeSelector(topology);
        executor._topologyEtag = -2;
        executor._disableTopologyUpdates = true;
        executor._disableClientConfigurationUpdates = true;
        return executor;
    }
    _ensureNodeSelector() {
        return __awaiter(this, void 0, void 0, function* () {
            if (this._firstTopologyUpdatePromise
                && (!this._firstTopologyUpdateStatus.isFullfilled()
                    || this._firstTopologyUpdateStatus.isRejected())) {
                yield this._firstTopologyUpdatePromise;
            }
            if (!this._nodeSelector) {
                const topology = new Topology_1.Topology(this._topologyEtag, this.getTopologyNodes().slice());
                this._nodeSelector = new NodeSelector_1.NodeSelector(topology);
            }
        });
    }
    getPreferredNode() {
        return __awaiter(this, void 0, void 0, function* () {
            yield this._ensureNodeSelector();
            return this._nodeSelector.getPreferredNode();
        });
    }
    getNodeBySessionId(sessionId) {
        return __awaiter(this, void 0, void 0, function* () {
            yield this._ensureNodeSelector();
            return this._nodeSelector.getNodeBySessionId(sessionId);
        });
    }
    getFastestNode() {
        return __awaiter(this, void 0, void 0, function* () {
            yield this._ensureNodeSelector();
            return this._nodeSelector.getFastestNode();
        });
    }
    _updateClientConfigurationInternal() {
        return __awaiter(this, void 0, void 0, function* () {
            const oldDisableClientConfigurationUpdates = this._disableClientConfigurationUpdates;
            this._disableClientConfigurationUpdates = true;
            try {
                if (this._disposed) {
                    return;
                }
                const command = new GetClientConfigurationOperation_1.GetClientConfigurationCommand();
                const { currentNode, currentIndex } = this.chooseNodeForRequest(command, null);
                yield this.execute(command, null, {
                    chosenNode: currentNode,
                    nodeIndex: currentIndex,
                    shouldRetry: false
                });
                const clientConfigOpResult = command.result;
                if (!clientConfigOpResult) {
                    return;
                }
                this._conventions.updateFrom(clientConfigOpResult.configuration);
                this._clientConfigurationEtag = clientConfigOpResult.etag;
            }
            catch (err) {
                this._log.error(err, "Error getting client configuration.");
            }
            finally {
                this._disableClientConfigurationUpdates = oldDisableClientConfigurationUpdates;
            }
        });
    }
    _updateClientConfiguration() {
        return __awaiter(this, void 0, void 0, function* () {
            if (this._disposed) {
                return;
            }
            let semAcquiredContext;
            try {
                semAcquiredContext = SemaphoreUtil_1.acquireSemaphore(this._updateClientConfigurationSemaphore);
                yield semAcquiredContext.promise;
                yield this._updateClientConfigurationInternal();
            }
            finally {
                if (semAcquiredContext) {
                    semAcquiredContext.dispose();
                }
            }
        });
    }
    updateTopology(node, timeout, forceUpdate = false) {
        if (this._disposed) {
            return Promise.resolve(false);
        }
        if (this._disableTopologyUpdates) {
            return Promise.resolve(false);
        }
        const acquiredSemContext = SemaphoreUtil_1.acquireSemaphore(this._updateDatabaseTopologySemaphore, { timeout });
        const result = BluebirdPromise.resolve(acquiredSemContext.promise)
            .then(() => {
            if (this._disposed) {
                return false;
            }
            this._log.info(`Update topology from ${node.url}.`);
            const getTopology = new GetDatabaseTopologyCommand_1.GetDatabaseTopologyCommand();
            const getTopologyPromise = this.execute(getTopology, null, {
                chosenNode: node,
                nodeIndex: null,
                shouldRetry: false,
            });
            return getTopologyPromise
                .then(() => {
                const topology = getTopology.result;
                if (!this._nodeSelector) {
                    this._nodeSelector = new NodeSelector_1.NodeSelector(topology);
                    if (this._readBalanceBehavior === "FastestNode") {
                        this._nodeSelector.scheduleSpeedTest();
                    }
                }
                else if (this._nodeSelector.onUpdateTopology(topology, forceUpdate)) {
                    this._disposeAllFailedNodesTimers();
                    if (this._readBalanceBehavior === "FastestNode") {
                        this._nodeSelector.scheduleSpeedTest();
                    }
                }
                this._topologyEtag = this._nodeSelector.getTopology().etag;
                return true;
            });
        }, (reason) => {
            if (reason.name === "TimeoutError") {
                return false;
            }
            throw reason;
        })
            .finally(() => {
            acquiredSemContext.dispose();
        });
        return Promise.resolve(result);
    }
    static _validateUrls(initialUrls, authOptions) {
        const cleanUrls = [...Array(initialUrls.length)];
        let requireHttps = !!authOptions;
        for (let index = 0; index < initialUrls.length; index++) {
            const url = initialUrls[index];
            UriUtil_1.validateUri(url);
            cleanUrls[index] = url.replace(/\/$/, "");
            requireHttps = requireHttps || url.startsWith("https://");
        }
        if (!requireHttps) {
            return cleanUrls;
        }
        for (const url of initialUrls) {
            if (!url.startsWith("http://")) {
                continue;
            }
            if (authOptions && authOptions.certificate) {
                Exceptions_1.throwError("InvalidOperationException", "The url " + url + " is using HTTP, but a certificate is specified, which require us to use HTTPS");
            }
            Exceptions_1.throwError("InvalidOperationException", "The url " + url
                + " is using HTTP, but other urls are using HTTPS, and mixing of HTTP and HTTPS is not allowed.");
        }
        return cleanUrls;
    }
    _initializeUpdateTopologyTimer() {
        if (this._updateTopologyTimer || this._disposed) {
            return;
        }
        this._log.info("Initialize update topology timer.");
        const minInMs = 60 * 1000;
        const that = this;
        this._updateTopologyTimer =
            new Timer_1.Timer(function timerActionUpdateTopology() {
                return that._updateTopologyCallback();
            }, minInMs, minInMs);
    }
    _updateTopologyCallback() {
        const time = new Date();
        const minInMs = 60 * 1000;
        if (time.valueOf() - this._lastReturnedResponse.valueOf() <= minInMs) {
            return;
        }
        let serverNode;
        try {
            const selector = this._nodeSelector;
            if (!selector) {
                return;
            }
            const preferredNode = selector.getPreferredNode();
            serverNode = preferredNode.currentNode;
        }
        catch (err) {
            this._log.warn(err, "Couldn't get preferred node Topology from _updateTopologyTimer");
            return;
        }
        return this.updateTopology(serverNode, 0)
            .catch(err => {
            this._log.error(err, "Couldn't update topology from _updateTopologyTimer");
            return null;
        });
    }
    _firstTopologyUpdate(inputUrls) {
        return __awaiter(this, void 0, void 0, function* () {
            const initialUrls = RequestExecutor._validateUrls(inputUrls, this._authOptions);
            const topologyUpdateErrors = [];
            const tryUpdateTopology = (url, database) => __awaiter(this, void 0, void 0, function* () {
                const serverNode = new ServerNode_1.ServerNode({ url, database });
                try {
                    yield this.updateTopology(serverNode, TypeUtil_1.TypeUtil.MAX_INT32);
                    this._initializeUpdateTopologyTimer();
                    this._topologyTakenFromNode = serverNode;
                    return true;
                }
                catch (error) {
                    if (error.name === "DatabaseDoesNotExistException") {
                        this._lastKnownUrls = initialUrls;
                        throw error;
                    }
                    if (initialUrls.length === 0) {
                        this._lastKnownUrls = initialUrls;
                        Exceptions_1.throwError("InvalidOperationException", `Cannot get topology from server: ${url}.`, error);
                    }
                    topologyUpdateErrors.push({ url, error });
                    return false;
                }
            });
            const tryUpdateTopologyOnAllNodes = () => __awaiter(this, void 0, void 0, function* () {
                for (const url of initialUrls) {
                    if (yield tryUpdateTopology(url, this._databaseName)) {
                        return;
                    }
                }
                return false;
            });
            yield tryUpdateTopologyOnAllNodes();
            const topology = new Topology_1.Topology();
            topology.etag = this._topologyEtag;
            let topologyNodes = this.getTopologyNodes();
            if (!topologyNodes) {
                topologyNodes = initialUrls.map(url => {
                    const serverNode = new ServerNode_1.ServerNode({
                        url,
                        database: this._databaseName
                    });
                    serverNode.clusterTag = "!";
                    return serverNode;
                });
            }
            topology.nodes = topologyNodes;
            this._nodeSelector = new NodeSelector_1.NodeSelector(topology);
            if (initialUrls && initialUrls.length > 0) {
                this._initializeUpdateTopologyTimer();
                return;
            }
            this._lastKnownUrls = initialUrls;
            const details = topologyUpdateErrors
                .map(x => `${x.url} -> ${x.error && x.error.stack ? x.error.stack : x.error}`)
                .join(", ");
            this._throwExceptions(details);
        });
    }
    _throwExceptions(details) {
        Exceptions_1.throwError("InvalidOperationException", "Failed to retrieve database topology from all known nodes"
            + os.EOL + details);
    }
    _disposeAllFailedNodesTimers() {
        for (const item of this._failedNodesTimers) {
            item[1].dispose();
        }
        this._failedNodesTimers.clear();
    }
    chooseNodeForRequest(cmd, sessionInfo) {
        if (!cmd.isReadRequest) {
            return this._nodeSelector.getPreferredNode();
        }
        switch (this._readBalanceBehavior) {
            case "None":
                return this._nodeSelector.getPreferredNode();
            case "RoundRobin":
                return this._nodeSelector.getNodeBySessionId(sessionInfo ? sessionInfo.sessionId : 0);
            case "FastestNode":
                return this._nodeSelector.getFastestNode();
            default:
                Exceptions_1.throwError("NotSupportedException", `Invalid read balance behavior: ${this._readBalanceBehavior}`);
        }
    }
    execute(command, sessionInfo, options) {
        if (options) {
            return this._executeOnSpecificNode(command, sessionInfo, options);
        }
        this._log.info(`Execute command ${command.constructor.name}`);
        const topologyUpdate = this._firstTopologyUpdatePromise;
        const topologyUpdateStatus = this._firstTopologyUpdateStatus;
        if ((topologyUpdate && topologyUpdateStatus.isResolved()) || this._disableTopologyUpdates) {
            const currentIndexAndNode = this.chooseNodeForRequest(command, sessionInfo);
            return this._executeOnSpecificNode(command, sessionInfo, {
                chosenNode: currentIndexAndNode.currentNode,
                nodeIndex: currentIndexAndNode.currentIndex,
                shouldRetry: true
            });
        }
        else {
            return this._unlikelyExecute(command, topologyUpdate, sessionInfo);
        }
    }
    _unlikelyExecute(command, topologyUpdate, sessionInfo) {
        return __awaiter(this, void 0, void 0, function* () {
            try {
                if (!this._firstTopologyUpdatePromise) {
                    if (!this._lastKnownUrls) {
                        Exceptions_1.throwError("InvalidOperationException", "No known topology and no previously known one, cannot proceed, likely a bug");
                    }
                    topologyUpdate = this._firstTopologyUpdate(this._lastKnownUrls);
                }
                yield topologyUpdate;
            }
            catch (reason) {
                if (this._firstTopologyUpdatePromise === topologyUpdate) {
                    this._firstTopologyUpdatePromise = null;
                }
                this._log.warn(reason, "Error doing topology update.");
                throw reason;
            }
            const currentIndexAndNode = this.chooseNodeForRequest(command, sessionInfo);
            return this._executeOnSpecificNode(command, sessionInfo, {
                chosenNode: currentIndexAndNode.currentNode,
                nodeIndex: currentIndexAndNode.currentIndex,
                shouldRetry: true
            });
        });
    }
    _getFromCache(command, useCache, url, cachedItemMetadataCallback) {
        if (useCache
            && command.canCache
            && command.isReadRequest
            && command.responseType === "Object") {
            return this._cache.get(url, cachedItemMetadataCallback);
        }
        cachedItemMetadataCallback({
            changeVector: null,
            response: null
        });
        return new HttpCache_1.ReleaseCacheItem(null);
    }
    _executeOnSpecificNode(command, sessionInfo = null, options = null) {
        return __awaiter(this, void 0, void 0, function* () {
            const { chosenNode, nodeIndex, shouldRetry } = options;
            this._log.info(`Actual execute ${command.constructor.name} on ${chosenNode.url}`
                + ` ${shouldRetry ? "with" : "without"} retry.`);
            const req = this._createRequest(chosenNode, command);
            const noCaching = sessionInfo ? sessionInfo.noCaching : false;
            let cachedChangeVector;
            let cachedValue;
            const cachedItem = this._getFromCache(command, !noCaching, req.uri.toString(), (cachedItemMetadata) => {
                cachedChangeVector = cachedItemMetadata.changeVector;
                cachedValue = cachedItemMetadata.response;
            });
            if (cachedChangeVector) {
                const aggressiveCacheOptions = this.aggressiveCaching;
                if (aggressiveCacheOptions
                    && cachedItem.age < aggressiveCacheOptions.duration
                    && !cachedItem.mightHaveBeenModified
                    && command.canCacheAggressively) {
                    return command.setResponseFromCache(cachedValue);
                }
                req.headers["If-None-Match"] = `"${cachedChangeVector}"`;
            }
            if (!this._disableClientConfigurationUpdates) {
                req.headers[Constants_1.HEADERS.CLIENT_CONFIGURATION_ETAG] = `"${this._clientConfigurationEtag}"`;
            }
            if (sessionInfo && sessionInfo.lastClusterTransactionIndex) {
                req.headers[Constants_1.HEADERS.LAST_KNOWN_CLUSTER_TRANSACTION_INDEX] =
                    sessionInfo.lastClusterTransactionIndex;
            }
            if (!this._disableTopologyUpdates) {
                req.headers[Constants_1.HEADERS.TOPOLOGY_ETAG] = `"${this._topologyEtag}"`;
            }
            const sp = Stopwatch_1.Stopwatch.createStarted();
            let response = null;
            let responseDispose = "Automatic";
            let bodyStream;
            this.numberOfServerRequests++;
            try {
                if (this._shouldExecuteOnAll(chosenNode, command)) {
                    response = yield this._executeOnAllToFigureOutTheFastest(chosenNode, command);
                }
                else {
                    const responseAndBody = yield command.send(req);
                    bodyStream = responseAndBody.bodyStream;
                    response = responseAndBody.response;
                }
                if (sessionInfo && sessionInfo.lastClusterTransactionIndex) {
                    const version = response.caseless.get(Constants_1.HEADERS.SERVER_VERSION);
                    if (version && "4.1" === version) {
                        Exceptions_1.throwError("ClientVersionMismatchException", "The server on " + chosenNode.url + " has an old version and can't perform "
                            + "the command since this command dependent on a cluster transaction "
                            + " which this node doesn't support.");
                    }
                }
                sp.stop();
            }
            catch (error) {
                this._log.warn(error, `Error executing '${command.constructor.name}' `
                    + `on specific node '${chosenNode.url}'`
                    + `${chosenNode.database ? "db " + chosenNode.database : ""}.`);
                if (!shouldRetry) {
                    throw error;
                }
                sp.stop();
                const serverDownHandledSuccessfully = yield this._handleServerDown(req.uri, chosenNode, nodeIndex, command, req, response, null, error, sessionInfo, shouldRetry);
                if (!serverDownHandledSuccessfully) {
                    this._throwFailedToContactAllNodes(command, req, error, null);
                }
                return;
            }
            command.statusCode = response.statusCode;
            const refreshTopology = response
                && response.caseless
                && response.caseless.get(Constants_1.HEADERS.REFRESH_TOPOLOGY);
            const refreshClientConfiguration = response
                && response.caseless
                && response.caseless.get(Constants_1.HEADERS.REFRESH_CLIENT_CONFIGURATION);
            try {
                if (response.statusCode === StatusCode_1.StatusCodes.NotModified) {
                    cachedItem.notModified();
                    if (command.responseType === "Object") {
                        yield command.setResponseFromCache(cachedValue);
                    }
                    return;
                }
                if (response.statusCode >= 400) {
                    const unsuccessfulResponseHandled = yield this._handleUnsuccessfulResponse(chosenNode, nodeIndex, command, req, response, bodyStream, req.uri, sessionInfo, shouldRetry);
                    if (!unsuccessfulResponseHandled) {
                        const dbMissingHeader = response.caseless.get(Constants_1.HEADERS.DATABASE_MISSING);
                        if (dbMissingHeader) {
                            Exceptions_1.throwError("DatabaseDoesNotExistException", dbMissingHeader);
                        }
                        if (command.failedNodes.size === 0) {
                            Exceptions_1.throwError("InvalidOperationException", "Received unsuccessful response and couldn't recover from it. "
                                + "Also, no record of exceptions per failed nodes. "
                                + "This is weird and should not happen.");
                        }
                        if (command.failedNodes.size === 1) {
                            const values = [...command.failedNodes.values()];
                            if (values && values.some(x => !!x)) {
                                const err = values.filter(x => !!x).map(x => x)[0];
                                Exceptions_1.throwError(err.name, err.message, err);
                            }
                        }
                        Exceptions_1.throwError("AllTopologyNodesDownException", "Received unsuccessful response from all servers"
                            + " and couldn't recover from it.");
                    }
                    return;
                }
                responseDispose = yield command.processResponse(this._cache, response, bodyStream, req.uri);
                this._lastReturnedResponse = new Date();
            }
            finally {
                if (responseDispose === "Automatic") {
                    HttpUtil_1.closeHttpResponse(response);
                }
                if (refreshTopology || refreshClientConfiguration) {
                    const serverNode = new ServerNode_1.ServerNode({
                        url: chosenNode.url,
                        database: this._databaseName
                    });
                    const topologyTask = refreshTopology
                        ? BluebirdPromise.resolve(this.updateTopology(serverNode, 0))
                            .tapCatch(err => this._log.warn(err, "Error refreshing topology."))
                        : BluebirdPromise.resolve(false);
                    const clientConfigurationTask = refreshClientConfiguration
                        ? BluebirdPromise.resolve(this._updateClientConfiguration())
                            .tapCatch(err => this._log.warn(err, "Error refreshing client configuration."))
                            .then(() => true)
                        : BluebirdPromise.resolve(false);
                    yield Promise.all([topologyTask, clientConfigurationTask]);
                }
            }
        });
    }
    _throwFailedToContactAllNodes(command, req, e, timeoutException) {
        let message = "Tried to send "
            + command.constructor.name
            + " request via "
            + (req.method || "GET") + " "
            + req.uri + " to all configured nodes in the topology, "
            + "all of them seem to be down or not responding. I've tried to access the following nodes: ";
        if (this._nodeSelector) {
            const topology = this._nodeSelector.getTopology();
            if (topology) {
                message += topology.nodes.map(x => x.url).join(", ");
            }
        }
        const tplFromNode = this._topologyTakenFromNode;
        if (tplFromNode && this._nodeSelector) {
            const topology = this._nodeSelector.getTopology();
            if (topology) {
                const nodesText = topology.nodes
                    .map(x => `( url: ${x.url}, clusterTag: ${x.clusterTag}, serverRole: ${x.serverRole})`)
                    .join(", ");
                message += os.EOL
                    + `I was able to fetch ${tplFromNode.database} topology from ${tplFromNode.url}.`
                    + os.EOL
                    + `Fetched topology: ${nodesText}`;
            }
        }
        const innerErr = timeoutException || e;
        Exceptions_1.throwError("AllTopologyNodesDownException", message, innerErr);
    }
    inSpeedTestPhase() {
        return this._nodeSelector
            && this._nodeSelector.inSpeedTestPhase();
    }
    _handleUnsuccessfulResponse(chosenNode, nodeIndex, command, req, response, responseBodyStream, url, sessionInfo, shouldRetry) {
        return __awaiter(this, void 0, void 0, function* () {
            responseBodyStream.resume();
            const readBody = () => StreamUtil.readToEnd(responseBodyStream);
            switch (response.statusCode) {
                case StatusCode_1.StatusCodes.NotFound:
                    this._cache.setNotFound(url);
                    switch (command.responseType) {
                        case "Empty":
                            return Promise.resolve(true);
                        case "Object":
                            return command.setResponseAsync(null, false)
                                .then(() => true);
                        default:
                            command.setResponseRaw(response, null);
                            break;
                    }
                    return true;
                case StatusCode_1.StatusCodes.Forbidden:
                    Exceptions_1.throwError("AuthorizationException", `Forbidden access to ${chosenNode.database}@${chosenNode.url}`
                        + `, ${req.method || "GET"} ${req.uri}`);
                    break;
                case StatusCode_1.StatusCodes.Gone:
                    if (!shouldRetry) {
                        return false;
                    }
                    return this.updateTopology(chosenNode, Number.MAX_VALUE, true)
                        .then(() => {
                        const currentIndexAndNode = this.chooseNodeForRequest(command, sessionInfo);
                        return this._executeOnSpecificNode(command, sessionInfo, {
                            chosenNode: currentIndexAndNode.currentNode,
                            nodeIndex: currentIndexAndNode.currentIndex,
                            shouldRetry: false
                        });
                    })
                        .then(() => true);
                case StatusCode_1.StatusCodes.GatewayTimeout:
                case StatusCode_1.StatusCodes.RequestTimeout:
                case StatusCode_1.StatusCodes.BadGateway:
                case StatusCode_1.StatusCodes.ServiceUnavailable:
                    return this._handleServerDown(url, chosenNode, nodeIndex, command, req, response, yield readBody(), null, sessionInfo, shouldRetry);
                case StatusCode_1.StatusCodes.Conflict:
                    RequestExecutor._handleConflict(response, yield readBody());
                    break;
                default:
                    command.onResponseFailure(response);
                    Exceptions_1.ExceptionDispatcher.throwException(response, yield readBody());
            }
        });
    }
    _executeOnAllToFigureOutTheFastest(chosenNode, command) {
        let preferredTask = null;
        const nodes = this._nodeSelector.getTopology().nodes;
        const tasks = nodes.map(x => null);
        let task;
        for (let i = 0; i < nodes.length; i++) {
            const taskNumber = i;
            this.numberOfServerRequests++;
            task = BluebirdPromise.resolve()
                .then(() => {
                const req = this._createRequest(nodes[taskNumber], command);
                return command.send(req)
                    .then(responseAndBodyStream => {
                    return responseAndBodyStream.response;
                });
            })
                .then(commandResult => new IndexAndResponse(taskNumber, commandResult))
                .catch(err => {
                tasks[taskNumber] = null;
                return BluebirdPromise.reject(err);
            });
            if (nodes[i].clusterTag === chosenNode.clusterTag) {
                preferredTask = task;
            }
            tasks[i] = task;
        }
        const result = PromiseUtil.raceToResolution(tasks)
            .then(fastest => {
            this._nodeSelector.recordFastest(fastest.index, nodes[fastest.index]);
        })
            .catch((err) => {
            this._log.warn(err, "Error executing on all to find fastest node.");
        })
            .then(() => preferredTask)
            .then(taskResult => taskResult.response);
        return Promise.resolve(result);
    }
    _shouldExecuteOnAll(chosenNode, command) {
        return this._readBalanceBehavior === "FastestNode" &&
            this._nodeSelector &&
            this._nodeSelector.inSpeedTestPhase() &&
            this._nodeSelectorHasMultipleNodes() &&
            command.isReadRequest &&
            command.responseType === "Object" &&
            !!chosenNode;
    }
    _nodeSelectorHasMultipleNodes() {
        const selector = this._nodeSelector;
        if (!selector) {
            return false;
        }
        const topology = selector.getTopology();
        return topology && topology.nodes && topology.nodes.length > 1;
    }
    _handleServerDown(url, chosenNode, nodeIndex, command, req, response, body, error, sessionInfo, shouldRetry) {
        return __awaiter(this, void 0, void 0, function* () {
            if (!command.failedNodes) {
                command.failedNodes = new Map();
            }
            RequestExecutor._addFailedResponseToCommand(chosenNode, command, req, response, body, error);
            if (nodeIndex === null) {
                return false;
            }
            this._spawnHealthChecks(chosenNode, nodeIndex);
            if (!this._nodeSelector) {
                return false;
            }
            this._nodeSelector.onFailedRequest(nodeIndex);
            const currentIndexAndNode = this._nodeSelector.getPreferredNode();
            if (command.failedNodes.has(currentIndexAndNode.currentNode)) {
                return false;
            }
            yield this._executeOnSpecificNode(command, sessionInfo, {
                chosenNode: currentIndexAndNode.currentNode,
                nodeIndex: currentIndexAndNode.currentIndex,
                shouldRetry: shouldRetry
            });
            return true;
        });
    }
    static _addFailedResponseToCommand(chosenNode, command, req, response, body, e) {
        if (response && body) {
            const responseJson = body;
            try {
                const resExceptionSchema = Serializer_1.JsonSerializer
                    .getDefaultForCommandPayload()
                    .deserialize(responseJson);
                const readException = Exceptions_1.ExceptionDispatcher.get(resExceptionSchema, response.statusCode);
                command.failedNodes.set(chosenNode, readException);
            }
            catch (_) {
                log.warn(_, "Error parsing server error.");
                const unrecognizedErrSchema = {
                    url: req.uri,
                    message: "Unrecognized response from the server",
                    error: responseJson,
                    type: "Unparsable Server Response"
                };
                const exceptionToUse = Exceptions_1.ExceptionDispatcher.get(unrecognizedErrSchema, response.statusCode);
                command.failedNodes.set(chosenNode, exceptionToUse);
            }
            return;
        }
        const exceptionSchema = {
            url: req.uri.toString(),
            message: e.message,
            error: `An exception occurred while contacting ${req.uri} . ${os.EOL + e.stack}`,
            type: e.name
        };
        command.failedNodes.set(chosenNode, Exceptions_1.ExceptionDispatcher.get(exceptionSchema, StatusCode_1.StatusCodes.ServiceUnavailable));
    }
    _createRequest(node, command) {
        const req = Object.assign(command.createRequest(node), this._defaultRequestOptions);
        req.headers = req.headers || {};
        if (this._authOptions) {
            const agentOptions = this._certificate.toAgentOptions();
            req.agentOptions = Object.assign(req.agentOptions || {}, agentOptions);
        }
        if (!req.headers[Constants_1.HEADERS.CLIENT_VERSION]) {
            req.headers[Constants_1.HEADERS.CLIENT_VERSION] = RequestExecutor.CLIENT_VERSION;
        }
        if (RequestExecutor.requestPostProcessor) {
            RequestExecutor.requestPostProcessor(req);
        }
        return req;
    }
    static _handleConflict(response, body) {
        Exceptions_1.ExceptionDispatcher.throwException(response, body);
    }
    _spawnHealthChecks(chosenNode, nodeIndex) {
        if (this._disposed) {
            return;
        }
        if (this._failedNodesTimers.has(chosenNode)) {
            return;
        }
        this._log.info(`Spawn health checks for node ${chosenNode.url}.`);
        const nodeStatus = new NodeStatus(nodeIndex, chosenNode, (nStatus) => this._checkNodeStatusCallback(nStatus));
        this._failedNodesTimers.set(chosenNode, nodeStatus);
        nodeStatus.startTimer();
    }
    _checkNodeStatusCallback(nodeStatus) {
        const copy = this.getTopologyNodes();
        if (nodeStatus.nodeIndex >= copy.length) {
            return;
        }
        const serverNode = copy[nodeStatus.nodeIndex];
        if (serverNode !== nodeStatus.node) {
            return;
        }
        return Promise.resolve()
            .then(() => {
            let status;
            return Promise.resolve(this._performHealthCheck(serverNode, nodeStatus.nodeIndex))
                .then(() => {
                status = this._failedNodesTimers[nodeStatus.nodeIndex];
                if (status) {
                    this._failedNodesTimers.delete(nodeStatus.node);
                    status.dispose();
                }
                if (this._nodeSelector) {
                    this._nodeSelector.restoreNodeIndex(nodeStatus.nodeIndex);
                }
            }, err => {
                this._log.error(err, `${serverNode.clusterTag} is still down`);
                status = this._failedNodesTimers.get(nodeStatus.node);
                if (status) {
                    nodeStatus.updateTimer();
                }
            });
        })
            .catch(err => {
            this._log.error(err, "Failed to check node topology, will ignore this node until next topology update.");
        });
    }
    _performHealthCheck(serverNode, nodeIndex) {
        return this._executeOnSpecificNode(RequestExecutor._failureCheckOperation.getCommand(this._conventions), null, {
            chosenNode: serverNode,
            nodeIndex,
            shouldRetry: false,
        });
    }
    _setDefaultRequestOptions() {
        this._defaultRequestOptions = Object.assign(DEFAULT_REQUEST_OPTIONS, {
            gzip: !(this._conventions.hasExplicitlySetCompressionUsage && !this._conventions.useCompression)
        }, this._customHttpRequestOptions);
    }
    dispose() {
        this._log.info("Dispose.");
        if (this._disposed) {
            return;
        }
        this._disposed = true;
        this._updateClientConfigurationSemaphore.take(TypeUtil_1.TypeUtil.NOOP);
        this._updateDatabaseTopologySemaphore.take(TypeUtil_1.TypeUtil.NOOP);
        this._cache.dispose();
        if (this._updateTopologyTimer) {
            this._updateTopologyTimer.dispose();
        }
        this._disposeAllFailedNodesTimers();
    }
}
RequestExecutor.CLIENT_VERSION = "4.1.0";
RequestExecutor._failureCheckOperation = new GetStatisticsOperation_1.GetStatisticsOperation("failure=check");
RequestExecutor.requestPostProcessor = null;
exports.RequestExecutor = RequestExecutor;