Method process

lib/writers.js:230–288

()

  }

  process() {
    if (this.error) {
      return;
    }

    const buffers = [];
    const callbacks = [];
    let totalLength = 0;

    while (this.queue.length > 0 && totalLength < this.coalescingThreshold) {
      const writeItem = this.queue.shift();
      if (!writeItem.operation.canBeWritten()) {
        // Invoke the write callback with an error that is not going to be yielded to user
        // as the operation has timed out or was cancelled.
        writeItem.callback(new Error('The operation was already cancelled or timeout elapsed'));
        continue;
      }
      let data;
      try {
        data = writeItem.operation.request.write(this.encoder, writeItem.operation.streamId);
      }
      catch (err) {
        writeItem.callback(err);
        continue;
      }
      totalLength += data.length;
      buffers.push(data);
      callbacks.push(writeItem.callback);
    }

    if (totalLength === 0) {
      this.isRunning = false;
      return;
    }

    // We have to invoke the callbacks to avoid race conditions.
    // There is a performance benefit from executing all of them in a loop
    for (let i = 0; i < callbacks.length; i++) {
      callbacks[i]();
    }

    // Concatenate buffers and write it to the socket
    // Further writes will be throttled until flushed
    this.canWrite = this.netClient.write(Buffer.concat(buffers, totalLength), err => {
      if (err) {
        this.setWriteError(err);
        return;
      }

      if (this.queue.length === 0 || !this.canWrite) {
        // It will start running once we get the next message or has drained
        this.isRunning = false;
        return;
      }

      // Allow IO between writes
      setImmediate(() => this.process());
    });