grpc-js: Add a single-subchannel channel implementation

Author: murgatroid99

packages/grpc-js/src/single-subchannel-channel.ts

@@ -0,0 +1,177 @@
+/*
+ * Copyright 2025 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 { AuthContext } from "./auth-context";
+import { CallCredentials } from "./call-credentials";
+import { Call, CallStreamOptions, InterceptingListener, MessageContext, StatusObject } from "./call-interface";
+import { getNextCallNumber } from "./call-number";
+import { Channel } from "./channel";
+import { ChannelOptions } from "./channel-options";
+import { ChannelRef, ChannelzCallTracker, ChannelzChildrenTracker, ChannelzTrace, registerChannelzChannel, unregisterChannelzRef } from "./channelz";
+import { ConnectivityState } from "./connectivity-state";
+import { Propagate, Status } from "./constants";
+import { Deadline, getRelativeTimeout } from "./deadline";
+import { Metadata } from "./metadata";
+import { getDefaultAuthority } from "./resolver";
+import { Subchannel } from "./subchannel";
+import { SubchannelCall } from "./subchannel-call";
+import { GrpcUri, uriToString } from "./uri-parser";
+
+class SubchannelCallWrapper implements Call {
+  private childCall: SubchannelCall | null = null;
+  private pendingMessage: { context: MessageContext; message: Buffer } | null =
+    null;
+  private readPending = false;
+  private halfClosePending = false;
+  private pendingStatus: StatusObject | null = null;
+  constructor(private subchannel: Subchannel, private method: string, private options: CallStreamOptions, private callNumber: number) {
+    const timeout = getRelativeTimeout(options.deadline);
+    if (timeout !== Infinity) {
+      if (timeout <= 0) {
+        this.cancelWithStatus(Status.DEADLINE_EXCEEDED, 'Deadline exceeded');
+      } else {
+        setTimeout(() => {
+          this.cancelWithStatus(Status.DEADLINE_EXCEEDED, 'Deadline exceeded');
+        }, timeout);
+      }
+    }
+  }
+
+  cancelWithStatus(status: Status, details: string): void {
+    if (this.childCall) {
+      this.childCall.cancelWithStatus(status, details);
+    } else {
+      this.pendingStatus = {
+        code: status,
+        details: details,
+        metadata: new Metadata()
+      };
+    }
+
+  }
+  getPeer(): string {
+    return this.childCall?.getPeer() ?? this.subchannel.getAddress();
+  }
+  start(metadata: Metadata, listener: InterceptingListener): void {
+    if (this.pendingStatus) {
+      listener.onReceiveStatus(this.pendingStatus);
+      return;
+    }
+    if (this.subchannel.getConnectivityState() !== ConnectivityState.READY) {
+      listener.onReceiveStatus({
+        code: Status.UNAVAILABLE,
+        details: 'Subchannel not ready',
+        metadata: new Metadata()
+      });
+      return;
+    }
+    this.childCall = this.subchannel.createCall(metadata, this.options.host, this.method, listener);
+    if (this.readPending) {
+      this.childCall.startRead();
+    }
+    if (this.pendingMessage) {
+      this.childCall.sendMessageWithContext(this.pendingMessage.context, this.pendingMessage.message);
+    }
+    if (this.halfClosePending) {
+      this.childCall.halfClose();
+    }
+  }
+  sendMessageWithContext(context: MessageContext, message: Buffer): void {
+    if (this.childCall) {
+      this.childCall.sendMessageWithContext(context, message);
+    } else {
+      this.pendingMessage = { context, message };
+    }
+  }
+  startRead(): void {
+    if (this.childCall) {
+      this.childCall.startRead();
+    } else {
+      this.readPending = true;
+    }
+  }
+  halfClose(): void {
+    if (this.childCall) {
+      this.childCall.halfClose();
+    } else {
+      this.halfClosePending = true;
+    }
+  }
+  getCallNumber(): number {
+    return this.callNumber;
+  }
+  setCredentials(credentials: CallCredentials): void {
+    throw new Error("Method not implemented.");
+  }
+  getAuthContext(): AuthContext | null {
+    if (this.childCall) {
+      return this.childCall.getAuthContext();
+    } else {
+      return null;
+    }
+  }
+}
+
+export class SingleSubchannelChannel implements Channel {
+  private channelzRef: ChannelRef;
+  private channelzEnabled = false;
+  private channelzTrace = new ChannelzTrace();
+  private callTracker = new ChannelzCallTracker();
+  private childrenTracker = new ChannelzChildrenTracker();
+  constructor(private subchannel: Subchannel, private target: GrpcUri, options: ChannelOptions) {
+    this.channelzEnabled = options['grpc.enable_channelz'] !== 0;
+    this.channelzRef = registerChannelzChannel(uriToString(target),  () => ({
+      target: `${uriToString(target)} (${subchannel.getAddress()})`,
+      state: this.subchannel.getConnectivityState(),
+      trace: this.channelzTrace,
+      callTracker: this.callTracker,
+      children: this.childrenTracker.getChildLists()
+    }), this.channelzEnabled);
+    if (this.channelzEnabled) {
+      this.childrenTracker.refChild(subchannel.getChannelzRef());
+    }
+  }
+
+  close(): void {
+    if (this.channelzEnabled) {
+      this.childrenTracker.unrefChild(this.subchannel.getChannelzRef());
+    }
+    unregisterChannelzRef(this.channelzRef);
+  }
+
+  getTarget(): string {
+    return uriToString(this.target);
+  }
+  getConnectivityState(tryToConnect: boolean): ConnectivityState {
+    throw new Error("Method not implemented.");
+  }
+  watchConnectivityState(currentState: ConnectivityState, deadline: Date | number, callback: (error?: Error) => void): void {
+    throw new Error("Method not implemented.");
+  }
+  getChannelzRef(): ChannelRef {
+    return this.channelzRef;
+  }
+  createCall(method: string, deadline: Deadline): Call {
+    const callOptions: CallStreamOptions = {
+      deadline: deadline,
+      host: getDefaultAuthority(this.target),
+      flags: Propagate.DEFAULTS,
+      parentCall: null
+    };
+    return new SubchannelCallWrapper(this.subchannel, method, callOptions, getNextCallNumber());
+  }
+}