Method _executeOne

lib/concurrent/index.js:205–246

(params)

  }

  _executeOne(params) {
    if (!Array.isArray(params)) {
      return this._setReadEnded(new TypeError('Stream should be in objectMode and should emit Array instances'));
    }

    if (this._readEnded) {
      // Read ended abruptly because of incorrect format or error event being emitted.
      // We shouldn't consider additional items.
      return;
    }

    const index = this._index++;
    this._inFlight++;

    this._client.execute(this._query, params, this._queryOptions)
      .then(rs => {
        this._result.setResultItem(index, rs);
        this._inFlight--;
      })
      .catch(err => {
        this._inFlight--;
        this._setError(index, err);
      })
      .then(() => {
        if (this._stream.isPaused()) {
          this._stream.resume();
        }

        if (this._readEnded && this._inFlight === 0) {
          // When read ended and there are no more in-flight requests
          // We yield the result to the user.
          // It could have ended prematurely when there is a read error
          // or there was an execution error and raiseOnFirstError is true
          // In that case, calling the resolve callback has no effect
          this._resolveCallback(this._result);
        }
      });

    if (this._inFlight >= this._concurrencyLevel) {
      this._stream.pause();
    }
  }

  /**
   * Marks the stream read process as ended.