@grpc/grpc-js/src/resolving-call.ts

/*
 * Copyright 2022 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

import { CallCredentials } from "./call-credentials";
import { Call, CallStreamOptions, InterceptingListener, MessageContext, StatusObject } from "./call-interface";
import { LogVerbosity, Propagate, Status } from "./constants";
import { Deadline, deadlineToString, getDeadlineTimeoutString, getRelativeTimeout, minDeadline } from "./deadline";
import { FilterStack, FilterStackFactory } from "./filter-stack";
import { InternalChannel } from "./internal-channel";
import { Metadata } from "./metadata";
import * as logging from './logging';
import { restrictControlPlaneStatusCode } from "./control-plane-status";

const TRACER_NAME = 'resolving_call';

export class ResolvingCall implements Call {
  private child: Call | null = null;
  private readPending = false;
  private pendingMessage: {context: MessageContext, message: Buffer} | null = null;
  private pendingHalfClose = false;
  private ended = false;
  private readFilterPending = false;
  private writeFilterPending = false;
  private pendingChildStatus: StatusObject | null = null;
  private metadata: Metadata | null = null;
  private listener: InterceptingListener | null = null;
  private deadline: Deadline;
  private host: string;
  private statusWatchers: ((status: StatusObject) => void)[] = [];
  private deadlineTimer: NodeJS.Timer = setTimeout(() => {}, 0);
  private filterStack: FilterStack | null = null;

  constructor(
    private readonly channel: InternalChannel,
    private readonly method: string,
    options: CallStreamOptions,
    private readonly filterStackFactory: FilterStackFactory,
    private credentials: CallCredentials,
    private callNumber: number
  ) {
    this.deadline = options.deadline;
    this.host = options.host;
    if (options.parentCall) {
      if (options.flags & Propagate.CANCELLATION) {
        options.parentCall.on('cancelled', () => {
          this.cancelWithStatus(Status.CANCELLED, 'Cancelled by parent call');
        });
      }
      if (options.flags & Propagate.DEADLINE) {
        this.trace('Propagating deadline from parent: ' + options.parentCall.getDeadline());
        this.deadline = minDeadline(this.deadline, options.parentCall.getDeadline());
      }
    }
    this.trace('Created');
    this.runDeadlineTimer();
  }

  private trace(text: string): void {
    logging.trace(
      LogVerbosity.DEBUG,
      TRACER_NAME,
      '[' + this.callNumber + '] ' + text
    );
  }

  private runDeadlineTimer() {
    clearTimeout(this.deadlineTimer);
    this.trace('Deadline: ' + deadlineToString(this.deadline));
    const timeout = getRelativeTimeout(this.deadline);
    if (timeout !== Infinity) {
      this.trace('Deadline will be reached in ' + timeout + 'ms');
      const handleDeadline = () => {
        this.cancelWithStatus(
          Status.DEADLINE_EXCEEDED,
          'Deadline exceeded'
        );
      }
      if (timeout <= 0) {
        process.nextTick(handleDeadline);
      } else {
        this.deadlineTimer = setTimeout(handleDeadline, timeout);
      }
    }
  }

  private outputStatus(status: StatusObject) {
    if (!this.ended) {
      this.ended = true;
      if (!this.filterStack) {
        this.filterStack = this.filterStackFactory.createFilter();
      }
      clearTimeout(this.deadlineTimer);
      const filteredStatus = this.filterStack.receiveTrailers(status);
      this.trace('ended with status: code=' + filteredStatus.code + ' details="' + filteredStatus.details + '"');
      this.statusWatchers.forEach(watcher => watcher(filteredStatus));
      process.nextTick(() => {
        this.listener?.onReceiveStatus(filteredStatus);
      });
    }
  }

  private sendMessageOnChild(context: MessageContext, message: Buffer): void {
    if (!this.child) {
      throw new Error('sendMessageonChild called with child not populated');
    }
    const child = this.child;
    this.writeFilterPending = true;
    this.filterStack!.sendMessage(Promise.resolve({message: message, flags: context.flags})).then((filteredMessage) => {
      this.writeFilterPending = false;
      child.sendMessageWithContext(context, filteredMessage.message);
      if (this.pendingHalfClose) {
        child.halfClose();
      }
    }, (status: StatusObject) => {
      this.cancelWithStatus(status.code, status.details);
    });
  }

  getConfig(): void {
    if (this.ended) {
      return;
    }
    if (!this.metadata || !this.listener) {
      throw new Error('getConfig called before start');
    }
    const configResult = this.channel.getConfig(this.method, this.metadata);
    if (configResult.type === 'NONE') {
      this.channel.queueCallForConfig(this);
      return;
    } else if (configResult.type === 'ERROR') {
      if (this.metadata.getOptions().waitForReady) {
        this.channel.queueCallForConfig(this);
      } else {
        this.outputStatus(configResult.error);
      }
      return;
    }
    // configResult.type === 'SUCCESS'
    const config = configResult.config;
    if (config.status !== Status.OK) {
      const {code, details} = restrictControlPlaneStatusCode(config.status, 'Failed to route call to method ' + this.method);
      this.outputStatus({
        code: code,
        details: details,
        metadata: new Metadata()
      });
      return;
    }

    if (config.methodConfig.timeout) {
      const configDeadline = new Date();
      configDeadline.setSeconds(
        configDeadline.getSeconds() + config.methodConfig.timeout.seconds
      );
      configDeadline.setMilliseconds(
        configDeadline.getMilliseconds() +
          config.methodConfig.timeout.nanos / 1_000_000
      );
      this.deadline = minDeadline(this.deadline, configDeadline);
      this.runDeadlineTimer();
    }

    this.filterStackFactory.push(config.dynamicFilterFactories);
    this.filterStack = this.filterStackFactory.createFilter();
    this.filterStack.sendMetadata(Promise.resolve(this.metadata)).then(filteredMetadata => {
      this.child = this.channel.createInnerCall(config, this.method, this.host, this.credentials, this.deadline);
      this.trace('Created child [' + this.child.getCallNumber() + ']')
      this.child.start(filteredMetadata, {
        onReceiveMetadata: metadata => {
          this.trace('Received metadata')
          this.listener!.onReceiveMetadata(this.filterStack!.receiveMetadata(metadata));
        },
        onReceiveMessage: message => {
          this.trace('Received message');
          this.readFilterPending = true;
          this.filterStack!.receiveMessage(message).then(filteredMesssage => {
            this.trace('Finished filtering received message');
            this.readFilterPending = false;
            this.listener!.onReceiveMessage(filteredMesssage);
            if (this.pendingChildStatus) {
              this.outputStatus(this.pendingChildStatus);
            }
          }, (status: StatusObject) => {
            this.cancelWithStatus(status.code, status.details);
          });
        },
        onReceiveStatus: status => {
          this.trace('Received status');
          if (this.readFilterPending) {
            this.pendingChildStatus = status;
          } else {
            this.outputStatus(status);
          }
        }
      });
      if (this.readPending) {
        this.child.startRead();
      }
      if (this.pendingMessage) {
        this.sendMessageOnChild(this.pendingMessage.context, this.pendingMessage.message);
      } else if (this.pendingHalfClose) {
        this.child.halfClose();
      }
    }, (status: StatusObject) => {
      this.outputStatus(status);
    })
  }

  reportResolverError(status: StatusObject) {
    if (this.metadata?.getOptions().waitForReady) {
      this.channel.queueCallForConfig(this);
    } else {
      this.outputStatus(status);
    }
  }
  cancelWithStatus(status: Status, details: string): void {
    this.trace('cancelWithStatus code: ' + status + ' details: "' + details + '"');
    this.child?.cancelWithStatus(status, details);
    this.outputStatus({code: status, details: details, metadata: new Metadata()});
  }
  getPeer(): string {
    return this.child?.getPeer() ?? this.channel.getTarget();
  }
  start(metadata: Metadata, listener: InterceptingListener): void {
    this.trace('start called');
    this.metadata = metadata.clone();
    this.listener = listener;
    this.getConfig();
  }
  sendMessageWithContext(context: MessageContext, message: Buffer): void {
    this.trace('write() called with message of length ' + message.length);
    if (this.child) {
      this.sendMessageOnChild(context, message);
    } else {
      this.pendingMessage = {context, message};
    }
  }
  startRead(): void {
    this.trace('startRead called');
    if (this.child) {
      this.child.startRead();
    } else {
      this.readPending = true;
    }
  }
  halfClose(): void {
    this.trace('halfClose called');
    if (this.child && !this.writeFilterPending) {
      this.child.halfClose();
    } else {
      this.pendingHalfClose = true;
    }
  }
  setCredentials(credentials: CallCredentials): void {
    this.credentials = this.credentials.compose(credentials);
  }

  addStatusWatcher(watcher: (status: StatusObject) => void) {
    this.statusWatchers.push(watcher);
  }

  getCallNumber(): number {
    return this.callNumber;
  }
}