<commit_before>/*
 * 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, getDeadlineTimeoutString, getRelativeTimeout, minDeadline } from "./deadline";
import { 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 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);

  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: ' + this.deadline);
    if (this.deadline !== Infinity) {
      const timeout = getRelativeTimeout(this.deadline);
      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;
      this.trace('ended with status: code=' + status.code + ' details="' + status.details + '"');
      this.statusWatchers.forEach(watcher => watcher(status));
      process.nextTick(() => {
        this.listener?.onReceiveStatus(status);
      });
    }
  }

  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);
    }<commit_msg>grpc-js: Restart deadline timer after getting timeout from service config<commit_after>/*
 * 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, getDeadlineTimeoutString, getRelativeTimeout, minDeadline } from "./deadline";
import { 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 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);

  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: ' + this.deadline);
    if (this.deadline !== Infinity) {
      const timeout = getRelativeTimeout(this.deadline);
      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;
      this.trace('ended with status: code=' + status.code + ' details="' + status.details + '"');
      this.statusWatchers.forEach(watcher => watcher(status));
      process.nextTick(() => {
        this.listener?.onReceiveStatus(status);
      });
    }
  }

  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();
    }
{'fraction_non_alphanumeric': 0.11744353676117252, 'fraction_numerical': 0.003267659778952427, 'mean_word_length': 3.1441656710473915, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1352470', 'n_tokens_mistral': 2925, 'n_tokens_neox': 2898, 'n_words': 750}