1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012 |
- import * as http2 from 'http2';
- import { ChannelCredentials } from './channel-credentials';
- import { Metadata } from './metadata';
- import { Call, Http2CallStream, WriteObject } from './call-stream';
- import { ChannelOptions } from './channel-options';
- import { PeerCertificate, checkServerIdentity, TLSSocket, CipherNameAndProtocol } from 'tls';
- import { ConnectivityState } from './connectivity-state';
- import { BackoffTimeout, BackoffOptions } from './backoff-timeout';
- import { getDefaultAuthority } from './resolver';
- import * as logging from './logging';
- import { LogVerbosity, Status } from './constants';
- import { getProxiedConnection, ProxyConnectionResult } from './http_proxy';
- import * as net from 'net';
- import { GrpcUri, parseUri, splitHostPort, uriToString } from './uri-parser';
- import { ConnectionOptions } from 'tls';
- import { FilterFactory, Filter, BaseFilter } from './filter';
- import {
- stringToSubchannelAddress,
- SubchannelAddress,
- subchannelAddressToString,
- } from './subchannel-address';
- import { SubchannelRef, ChannelzTrace, ChannelzChildrenTracker, SubchannelInfo, registerChannelzSubchannel, ChannelzCallTracker, SocketInfo, SocketRef, unregisterChannelzRef, registerChannelzSocket, TlsInfo } from './channelz';
- import { ConnectivityStateListener } from './subchannel-interface';
- const clientVersion = require('../../package.json').version;
- const TRACER_NAME = 'subchannel';
- const FLOW_CONTROL_TRACER_NAME = 'subchannel_flowctrl';
- const MIN_CONNECT_TIMEOUT_MS = 20000;
- const INITIAL_BACKOFF_MS = 1000;
- const BACKOFF_MULTIPLIER = 1.6;
- const MAX_BACKOFF_MS = 120000;
- const BACKOFF_JITTER = 0.2;
- const KEEPALIVE_MAX_TIME_MS = ~(1 << 31);
- const KEEPALIVE_TIMEOUT_MS = 20000;
- export interface SubchannelCallStatsTracker {
- addMessageSent(): void;
- addMessageReceived(): void;
- }
- const {
- HTTP2_HEADER_AUTHORITY,
- HTTP2_HEADER_CONTENT_TYPE,
- HTTP2_HEADER_METHOD,
- HTTP2_HEADER_PATH,
- HTTP2_HEADER_TE,
- HTTP2_HEADER_USER_AGENT,
- } = http2.constants;
- function uniformRandom(min: number, max: number) {
- return Math.random() * (max - min) + min;
- }
- const tooManyPingsData: Buffer = Buffer.from('too_many_pings', 'ascii');
- export class Subchannel {
-
- private connectivityState: ConnectivityState = ConnectivityState.IDLE;
-
- private session: http2.ClientHttp2Session | null = null;
-
- private continueConnecting = false;
-
- private stateListeners: ConnectivityStateListener[] = [];
-
- private disconnectListeners: Set<() => void> = new Set();
- private backoffTimeout: BackoffTimeout;
-
- private userAgent: string;
-
- private keepaliveTimeMs: number = KEEPALIVE_MAX_TIME_MS;
-
- private keepaliveTimeoutMs: number = KEEPALIVE_TIMEOUT_MS;
-
- private keepaliveIntervalId: NodeJS.Timer;
-
- private keepaliveTimeoutId: NodeJS.Timer;
-
- private keepaliveWithoutCalls = false;
-
- private callRefcount = 0;
-
- private refcount = 0;
-
- private subchannelAddressString: string;
-
- private readonly channelzEnabled: boolean = true;
- private channelzRef: SubchannelRef;
- private channelzTrace: ChannelzTrace;
- private callTracker = new ChannelzCallTracker();
- private childrenTracker = new ChannelzChildrenTracker();
-
- private channelzSocketRef: SocketRef | null = null;
-
- private remoteName: string | null = null;
- private streamTracker = new ChannelzCallTracker();
- private keepalivesSent = 0;
- private messagesSent = 0;
- private messagesReceived = 0;
- private lastMessageSentTimestamp: Date | null = null;
- private lastMessageReceivedTimestamp: Date | null = null;
-
- constructor(
- private channelTarget: GrpcUri,
- private subchannelAddress: SubchannelAddress,
- private options: ChannelOptions,
- private credentials: ChannelCredentials
- ) {
-
- this.userAgent = [
- options['grpc.primary_user_agent'],
- `grpc-node-js/${clientVersion}`,
- options['grpc.secondary_user_agent'],
- ]
- .filter((e) => e)
- .join(' ');
- if ('grpc.keepalive_time_ms' in options) {
- this.keepaliveTimeMs = options['grpc.keepalive_time_ms']!;
- }
- if ('grpc.keepalive_timeout_ms' in options) {
- this.keepaliveTimeoutMs = options['grpc.keepalive_timeout_ms']!;
- }
- if ('grpc.keepalive_permit_without_calls' in options) {
- this.keepaliveWithoutCalls =
- options['grpc.keepalive_permit_without_calls'] === 1;
- } else {
- this.keepaliveWithoutCalls = false;
- }
- this.keepaliveIntervalId = setTimeout(() => {}, 0);
- clearTimeout(this.keepaliveIntervalId);
- this.keepaliveTimeoutId = setTimeout(() => {}, 0);
- clearTimeout(this.keepaliveTimeoutId);
- const backoffOptions: BackoffOptions = {
- initialDelay: options['grpc.initial_reconnect_backoff_ms'],
- maxDelay: options['grpc.max_reconnect_backoff_ms'],
- };
- this.backoffTimeout = new BackoffTimeout(() => {
- this.handleBackoffTimer();
- }, backoffOptions);
- this.subchannelAddressString = subchannelAddressToString(subchannelAddress);
- if (options['grpc.enable_channelz'] === 0) {
- this.channelzEnabled = false;
- }
- this.channelzTrace = new ChannelzTrace();
- this.channelzRef = registerChannelzSubchannel(this.subchannelAddressString, () => this.getChannelzInfo(), this.channelzEnabled);
- if (this.channelzEnabled) {
- this.channelzTrace.addTrace('CT_INFO', 'Subchannel created');
- }
- this.trace('Subchannel constructed with options ' + JSON.stringify(options, undefined, 2));
- }
- private getChannelzInfo(): SubchannelInfo {
- return {
- state: this.connectivityState,
- trace: this.channelzTrace,
- callTracker: this.callTracker,
- children: this.childrenTracker.getChildLists(),
- target: this.subchannelAddressString
- };
- }
- private getChannelzSocketInfo(): SocketInfo | null {
- if (this.session === null) {
- return null;
- }
- const sessionSocket = this.session.socket;
- const remoteAddress = sessionSocket.remoteAddress ? stringToSubchannelAddress(sessionSocket.remoteAddress, sessionSocket.remotePort) : null;
- const localAddress = sessionSocket.localAddress ? stringToSubchannelAddress(sessionSocket.localAddress, sessionSocket.localPort) : null;
- let tlsInfo: TlsInfo | null;
- if (this.session.encrypted) {
- const tlsSocket: TLSSocket = sessionSocket as TLSSocket;
- const cipherInfo: CipherNameAndProtocol & {standardName?: string} = tlsSocket.getCipher();
- const certificate = tlsSocket.getCertificate();
- const peerCertificate = tlsSocket.getPeerCertificate();
- tlsInfo = {
- cipherSuiteStandardName: cipherInfo.standardName ?? null,
- cipherSuiteOtherName: cipherInfo.standardName ? null : cipherInfo.name,
- localCertificate: (certificate && 'raw' in certificate) ? certificate.raw : null,
- remoteCertificate: (peerCertificate && 'raw' in peerCertificate) ? peerCertificate.raw : null
- };
- } else {
- tlsInfo = null;
- }
- const socketInfo: SocketInfo = {
- remoteAddress: remoteAddress,
- localAddress: localAddress,
- security: tlsInfo,
- remoteName: this.remoteName,
- streamsStarted: this.streamTracker.callsStarted,
- streamsSucceeded: this.streamTracker.callsSucceeded,
- streamsFailed: this.streamTracker.callsFailed,
- messagesSent: this.messagesSent,
- messagesReceived: this.messagesReceived,
- keepAlivesSent: this.keepalivesSent,
- lastLocalStreamCreatedTimestamp: this.streamTracker.lastCallStartedTimestamp,
- lastRemoteStreamCreatedTimestamp: null,
- lastMessageSentTimestamp: this.lastMessageSentTimestamp,
- lastMessageReceivedTimestamp: this.lastMessageReceivedTimestamp,
- localFlowControlWindow: this.session.state.localWindowSize ?? null,
- remoteFlowControlWindow: this.session.state.remoteWindowSize ?? null
- };
- return socketInfo;
- }
- private resetChannelzSocketInfo() {
- if (!this.channelzEnabled) {
- return;
- }
- if (this.channelzSocketRef) {
- unregisterChannelzRef(this.channelzSocketRef);
- this.childrenTracker.unrefChild(this.channelzSocketRef);
- this.channelzSocketRef = null;
- }
- this.remoteName = null;
- this.streamTracker = new ChannelzCallTracker();
- this.keepalivesSent = 0;
- this.messagesSent = 0;
- this.messagesReceived = 0;
- this.lastMessageSentTimestamp = null;
- this.lastMessageReceivedTimestamp = null;
- }
- private trace(text: string): void {
- logging.trace(LogVerbosity.DEBUG, TRACER_NAME, '(' + this.channelzRef.id + ') ' + this.subchannelAddressString + ' ' + text);
- }
- private refTrace(text: string): void {
- logging.trace(LogVerbosity.DEBUG, 'subchannel_refcount', '(' + this.channelzRef.id + ') ' + this.subchannelAddressString + ' ' + text);
- }
- private flowControlTrace(text: string): void {
- logging.trace(LogVerbosity.DEBUG, FLOW_CONTROL_TRACER_NAME, '(' + this.channelzRef.id + ') ' + this.subchannelAddressString + ' ' + text);
- }
- private internalsTrace(text: string): void {
- logging.trace(LogVerbosity.DEBUG, 'subchannel_internals', '(' + this.channelzRef.id + ') ' + this.subchannelAddressString + ' ' + text);
- }
- private keepaliveTrace(text: string): void {
- logging.trace(LogVerbosity.DEBUG, 'keepalive', '(' + this.channelzRef.id + ') ' + this.subchannelAddressString + ' ' + text);
- }
- private handleBackoffTimer() {
- if (this.continueConnecting) {
- this.transitionToState(
- [ConnectivityState.TRANSIENT_FAILURE],
- ConnectivityState.CONNECTING
- );
- } else {
- this.transitionToState(
- [ConnectivityState.TRANSIENT_FAILURE],
- ConnectivityState.IDLE
- );
- }
- }
-
- private startBackoff() {
- this.backoffTimeout.runOnce();
- }
- private stopBackoff() {
- this.backoffTimeout.stop();
- this.backoffTimeout.reset();
- }
- private sendPing() {
- if (this.channelzEnabled) {
- this.keepalivesSent += 1;
- }
- this.keepaliveTrace('Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms');
- this.keepaliveTimeoutId = setTimeout(() => {
- this.keepaliveTrace('Ping timeout passed without response');
- this.handleDisconnect();
- }, this.keepaliveTimeoutMs);
- this.keepaliveTimeoutId.unref?.();
- try {
- this.session!.ping(
- (err: Error | null, duration: number, payload: Buffer) => {
- this.keepaliveTrace('Received ping response');
- clearTimeout(this.keepaliveTimeoutId);
- }
- );
- } catch (e) {
-
- this.transitionToState(
- [ConnectivityState.READY],
- ConnectivityState.TRANSIENT_FAILURE
- );
- }
- }
- private startKeepalivePings() {
- this.keepaliveIntervalId = setInterval(() => {
- this.sendPing();
- }, this.keepaliveTimeMs);
- this.keepaliveIntervalId.unref?.();
-
- }
-
- private stopKeepalivePings() {
- clearInterval(this.keepaliveIntervalId);
- clearTimeout(this.keepaliveTimeoutId);
- }
- private createSession(proxyConnectionResult: ProxyConnectionResult) {
- if (proxyConnectionResult.realTarget) {
- this.remoteName = uriToString(proxyConnectionResult.realTarget);
- this.trace('creating HTTP/2 session through proxy to ' + proxyConnectionResult.realTarget);
- } else {
- this.remoteName = null;
- this.trace('creating HTTP/2 session');
- }
- const targetAuthority = getDefaultAuthority(
- proxyConnectionResult.realTarget ?? this.channelTarget
- );
- let connectionOptions: http2.SecureClientSessionOptions =
- this.credentials._getConnectionOptions() || {};
- connectionOptions.maxSendHeaderBlockLength = Number.MAX_SAFE_INTEGER;
- if ('grpc-node.max_session_memory' in this.options) {
- connectionOptions.maxSessionMemory = this.options[
- 'grpc-node.max_session_memory'
- ];
- } else {
-
- connectionOptions.maxSessionMemory = Number.MAX_SAFE_INTEGER;
- }
- let addressScheme = 'http://';
- if ('secureContext' in connectionOptions) {
- addressScheme = 'https://';
-
-
-
- if (this.options['grpc.ssl_target_name_override']) {
- const sslTargetNameOverride = this.options[
- 'grpc.ssl_target_name_override'
- ]!;
- connectionOptions.checkServerIdentity = (
- host: string,
- cert: PeerCertificate
- ): Error | undefined => {
- return checkServerIdentity(sslTargetNameOverride, cert);
- };
- connectionOptions.servername = sslTargetNameOverride;
- } else {
- const authorityHostname =
- splitHostPort(targetAuthority)?.host ?? 'localhost';
-
- connectionOptions.servername = authorityHostname;
- }
- if (proxyConnectionResult.socket) {
-
- connectionOptions.createConnection = (authority, option) => {
- return proxyConnectionResult.socket!;
- };
- }
- } else {
-
- connectionOptions.createConnection = (authority, option) => {
- if (proxyConnectionResult.socket) {
- return proxyConnectionResult.socket;
- } else {
-
- return net.connect(this.subchannelAddress);
- }
- };
- }
- connectionOptions = {
- ...connectionOptions,
- ...this.subchannelAddress,
- };
-
- const session = http2.connect(
- addressScheme + targetAuthority,
- connectionOptions
- );
- this.session = session;
- this.channelzSocketRef = registerChannelzSocket(this.subchannelAddressString, () => this.getChannelzSocketInfo()!, this.channelzEnabled);
- if (this.channelzEnabled) {
- this.childrenTracker.refChild(this.channelzSocketRef);
- }
- session.unref();
-
- session.once('connect', () => {
- if (this.session === session) {
- this.transitionToState(
- [ConnectivityState.CONNECTING],
- ConnectivityState.READY
- );
- }
- });
- session.once('close', () => {
- if (this.session === session) {
- this.trace('connection closed');
- this.transitionToState(
- [ConnectivityState.CONNECTING],
- ConnectivityState.TRANSIENT_FAILURE
- );
-
- this.transitionToState(
- [ConnectivityState.READY],
- ConnectivityState.IDLE
- );
- }
- });
- session.once(
- 'goaway',
- (errorCode: number, lastStreamID: number, opaqueData: Buffer) => {
- if (this.session === session) {
-
- if (
- errorCode === http2.constants.NGHTTP2_ENHANCE_YOUR_CALM &&
- opaqueData.equals(tooManyPingsData)
- ) {
- this.keepaliveTimeMs = Math.min(
- 2 * this.keepaliveTimeMs,
- KEEPALIVE_MAX_TIME_MS
- );
- logging.log(
- LogVerbosity.ERROR,
- `Connection to ${uriToString(this.channelTarget)} at ${
- this.subchannelAddressString
- } rejected by server because of excess pings. Increasing ping interval to ${
- this.keepaliveTimeMs
- } ms`
- );
- }
- this.trace(
- 'connection closed by GOAWAY with code ' +
- errorCode
- );
- this.transitionToState(
- [ConnectivityState.CONNECTING, ConnectivityState.READY],
- ConnectivityState.IDLE
- );
- }
- }
- );
- session.once('error', (error) => {
-
- this.trace(
- 'connection closed with error ' +
- (error as Error).message
- );
- });
- if (logging.isTracerEnabled(TRACER_NAME)) {
- session.on('remoteSettings', (settings: http2.Settings) => {
- this.trace(
- 'new settings received' +
- (this.session !== session ? ' on the old connection' : '') +
- ': ' +
- JSON.stringify(settings)
- );
- });
- session.on('localSettings', (settings: http2.Settings) => {
- this.trace(
- 'local settings acknowledged by remote' +
- (this.session !== session ? ' on the old connection' : '') +
- ': ' +
- JSON.stringify(settings)
- );
- });
- }
- }
- private startConnectingInternal() {
-
- const connectionOptions: ConnectionOptions =
- this.credentials._getConnectionOptions() || {};
- if ('secureContext' in connectionOptions) {
- connectionOptions.ALPNProtocols = ['h2'];
-
-
-
- if (this.options['grpc.ssl_target_name_override']) {
- const sslTargetNameOverride = this.options[
- 'grpc.ssl_target_name_override'
- ]!;
- connectionOptions.checkServerIdentity = (
- host: string,
- cert: PeerCertificate
- ): Error | undefined => {
- return checkServerIdentity(sslTargetNameOverride, cert);
- };
- connectionOptions.servername = sslTargetNameOverride;
- } else {
- if ('grpc.http_connect_target' in this.options) {
-
- const targetPath = getDefaultAuthority(
- parseUri(this.options['grpc.http_connect_target'] as string) ?? {
- path: 'localhost',
- }
- );
- const hostPort = splitHostPort(targetPath);
- connectionOptions.servername = hostPort?.host ?? targetPath;
- }
- }
- }
- getProxiedConnection(
- this.subchannelAddress,
- this.options,
- connectionOptions
- ).then(
- (result) => {
- this.createSession(result);
- },
- (reason) => {
- this.transitionToState(
- [ConnectivityState.CONNECTING],
- ConnectivityState.TRANSIENT_FAILURE
- );
- }
- );
- }
- private handleDisconnect() {
- this.transitionToState(
- [ConnectivityState.READY],
- ConnectivityState.TRANSIENT_FAILURE);
- for (const listener of this.disconnectListeners.values()) {
- listener();
- }
- }
-
- private transitionToState(
- oldStates: ConnectivityState[],
- newState: ConnectivityState
- ): boolean {
- if (oldStates.indexOf(this.connectivityState) === -1) {
- return false;
- }
- this.trace(
- ConnectivityState[this.connectivityState] +
- ' -> ' +
- ConnectivityState[newState]
- );
- if (this.channelzEnabled) {
- this.channelzTrace.addTrace('CT_INFO', ConnectivityState[this.connectivityState] + ' -> ' + ConnectivityState[newState]);
- }
- const previousState = this.connectivityState;
- this.connectivityState = newState;
- switch (newState) {
- case ConnectivityState.READY:
- this.stopBackoff();
- const session = this.session!;
- session.socket.once('close', () => {
- if (this.session === session) {
- this.handleDisconnect();
- }
- });
- if (this.keepaliveWithoutCalls) {
- this.startKeepalivePings();
- }
- break;
- case ConnectivityState.CONNECTING:
- this.startBackoff();
- this.startConnectingInternal();
- this.continueConnecting = false;
- break;
- case ConnectivityState.TRANSIENT_FAILURE:
- if (this.session) {
- this.session.close();
- }
- this.session = null;
- this.resetChannelzSocketInfo();
- this.stopKeepalivePings();
-
- if (!this.backoffTimeout.isRunning()) {
- process.nextTick(() => {
- this.handleBackoffTimer();
- });
- }
- break;
- case ConnectivityState.IDLE:
- if (this.session) {
- this.session.close();
- }
- this.session = null;
- this.resetChannelzSocketInfo();
- this.stopKeepalivePings();
- break;
- default:
- throw new Error(`Invalid state: unknown ConnectivityState ${newState}`);
- }
-
- for (const listener of [...this.stateListeners]) {
- listener(this, previousState, newState);
- }
- return true;
- }
-
- private checkBothRefcounts() {
-
- if (this.callRefcount === 0 && this.refcount === 0) {
- if (this.channelzEnabled) {
- this.channelzTrace.addTrace('CT_INFO', 'Shutting down');
- }
- this.transitionToState(
- [ConnectivityState.CONNECTING, ConnectivityState.READY],
- ConnectivityState.IDLE
- );
- if (this.channelzEnabled) {
- unregisterChannelzRef(this.channelzRef);
- }
- }
- }
- callRef() {
- this.refTrace(
- 'callRefcount ' +
- this.callRefcount +
- ' -> ' +
- (this.callRefcount + 1)
- );
- if (this.callRefcount === 0) {
- if (this.session) {
- this.session.ref();
- }
- this.backoffTimeout.ref();
- if (!this.keepaliveWithoutCalls) {
- this.startKeepalivePings();
- }
- }
- this.callRefcount += 1;
- }
- callUnref() {
- this.refTrace(
- 'callRefcount ' +
- this.callRefcount +
- ' -> ' +
- (this.callRefcount - 1)
- );
- this.callRefcount -= 1;
- if (this.callRefcount === 0) {
- if (this.session) {
- this.session.unref();
- }
- this.backoffTimeout.unref();
- if (!this.keepaliveWithoutCalls) {
- clearInterval(this.keepaliveIntervalId);
- }
- this.checkBothRefcounts();
- }
- }
- ref() {
- this.refTrace(
- 'refcount ' +
- this.refcount +
- ' -> ' +
- (this.refcount + 1)
- );
- this.refcount += 1;
- }
- unref() {
- this.refTrace(
- 'refcount ' +
- this.refcount +
- ' -> ' +
- (this.refcount - 1)
- );
- this.refcount -= 1;
- this.checkBothRefcounts();
- }
- unrefIfOneRef(): boolean {
- if (this.refcount === 1) {
- this.unref();
- return true;
- }
- return false;
- }
-
- startCallStream(
- metadata: Metadata,
- callStream: Http2CallStream,
- extraFilters: Filter[]
- ) {
- const headers = metadata.toHttp2Headers();
- headers[HTTP2_HEADER_AUTHORITY] = callStream.getHost();
- headers[HTTP2_HEADER_USER_AGENT] = this.userAgent;
- headers[HTTP2_HEADER_CONTENT_TYPE] = 'application/grpc';
- headers[HTTP2_HEADER_METHOD] = 'POST';
- headers[HTTP2_HEADER_PATH] = callStream.getMethod();
- headers[HTTP2_HEADER_TE] = 'trailers';
- let http2Stream: http2.ClientHttp2Stream;
-
- try {
- http2Stream = this.session!.request(headers);
- } catch (e) {
- this.transitionToState(
- [ConnectivityState.READY],
- ConnectivityState.TRANSIENT_FAILURE
- );
- throw e;
- }
- let headersString = '';
- for (const header of Object.keys(headers)) {
- headersString += '\t\t' + header + ': ' + headers[header] + '\n';
- }
- logging.trace(
- LogVerbosity.DEBUG,
- 'call_stream',
- 'Starting stream [' + callStream.getCallNumber() + '] on subchannel ' +
- '(' + this.channelzRef.id + ') ' +
- this.subchannelAddressString +
- ' with headers\n' +
- headersString
- );
- this.flowControlTrace(
- 'local window size: ' +
- this.session!.state.localWindowSize +
- ' remote window size: ' +
- this.session!.state.remoteWindowSize
- );
- const streamSession = this.session;
- this.internalsTrace(
- 'session.closed=' +
- streamSession!.closed +
- ' session.destroyed=' +
- streamSession!.destroyed +
- ' session.socket.destroyed=' +
- streamSession!.socket.destroyed);
- let statsTracker: SubchannelCallStatsTracker;
- if (this.channelzEnabled) {
- this.callTracker.addCallStarted();
- callStream.addStatusWatcher(status => {
- if (status.code === Status.OK) {
- this.callTracker.addCallSucceeded();
- } else {
- this.callTracker.addCallFailed();
- }
- });
- this.streamTracker.addCallStarted();
- callStream.addStreamEndWatcher(success => {
- if (streamSession === this.session) {
- if (success) {
- this.streamTracker.addCallSucceeded();
- } else {
- this.streamTracker.addCallFailed();
- }
- }
- });
- statsTracker = {
- addMessageSent: () => {
- this.messagesSent += 1;
- this.lastMessageSentTimestamp = new Date();
- },
- addMessageReceived: () => {
- this.messagesReceived += 1;
- }
- }
- } else {
- statsTracker = {
- addMessageSent: () => {},
- addMessageReceived: () => {}
- }
- }
- callStream.attachHttp2Stream(http2Stream, this, extraFilters, statsTracker);
- }
-
- startConnecting() {
-
- if (
- !this.transitionToState(
- [ConnectivityState.IDLE],
- ConnectivityState.CONNECTING
- )
- ) {
- if (this.connectivityState === ConnectivityState.TRANSIENT_FAILURE) {
- this.continueConnecting = true;
- }
- }
- }
-
- getConnectivityState() {
- return this.connectivityState;
- }
-
- addConnectivityStateListener(listener: ConnectivityStateListener) {
- this.stateListeners.push(listener);
- }
-
- removeConnectivityStateListener(listener: ConnectivityStateListener) {
- const listenerIndex = this.stateListeners.indexOf(listener);
- if (listenerIndex > -1) {
- this.stateListeners.splice(listenerIndex, 1);
- }
- }
- addDisconnectListener(listener: () => void) {
- this.disconnectListeners.add(listener);
- }
- removeDisconnectListener(listener: () => void) {
- this.disconnectListeners.delete(listener);
- }
-
- resetBackoff() {
- this.backoffTimeout.reset();
- this.transitionToState(
- [ConnectivityState.TRANSIENT_FAILURE],
- ConnectivityState.CONNECTING
- );
- }
- getAddress(): string {
- return this.subchannelAddressString;
- }
- getChannelzRef(): SubchannelRef {
- return this.channelzRef;
- }
- getRealSubchannel(): this {
- return this;
- }
- }
|