File Explorer

1/*2 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.3 * SPDX-License-Identifier: Apache-2.0.4 */5 6import {NativeResourceMixin} from "./native_resource";7import {BufferedEventEmitter} from "../common/event";8import {CrtError} from "./error";9import * as io from "./io";10import * as eventstream_utils from "./eventstream_utils";11import * as cancel from "../common/cancel";12import * as promise from "../common/promise";13import crt_native from "./binding";14 15/**16 * Node.js specific eventstream rpc native bindings17 *18 * @packageDocumentation19 * @module eventstream20 * @mergeTarget21 *22 */23 24/**25 * Supported types for the value within an eventstream message header26 */27export enum HeaderType {28 29    /** Value is True. No actual value is transmitted on the wire. */30    BooleanTrue = 0,31 32    /** Value is True. No actual value is transmitted on the wire. */33    BooleanFalse = 1,34 35    /** Value is signed 8-bit int. */36    Byte = 2,37 38    /** Value is signed 16-bit int. */39    Int16 = 3,40 41    /** Value is signed 32-bit int. */42    Int32 = 4,43 44    /** Value is signed 64-bit int. */45    Int64 = 5,46 47    /** Value is raw bytes. */48    ByteBuffer = 6,49 50    /** Value is a str.  Transmitted on the wire as utf-8. */51    String = 7,52 53    /** Value is a posix timestamp (seconds since Unix epoch).  Transmitted on the wire as a 64-bit int. */54    Timestamp = 8,55 56    /** Value is a UUID. Transmitted on the wire as 16 bytes. */57    UUID = 9,58}59 60/**61 * Union type for message payloads.62 *63 * Payloads are allowed to be any of the these types in an outbound message.64 * Payloads will always be ArrayBuffers when emitting received messages.65 */66export type Payload = string | ArrayBuffer | ArrayBufferView;67 68const AWS_MAXIMUM_EVENT_STREAM_HEADER_NAME_LENGTH : number = 127;69 70type HeaderValue =71    undefined |  /* BooleanTrue, BooleanFalse */72    number |  /* byte, int16, int32, timestamp */73    string |  /* string */74    Payload;  /* ByteBuffer, UUID (via ArrayBuffer), int64 */75 76/**77 * Wrapper class for event stream message headers.  Similar to HTTP, a header is a name-value pair.  Unlike HTTP, the78 * value's wire format varies depending on a type annotation.  We provide static builder functions to help79 * ensure correct type agreement (type annotation matches actual value) at construction time.  Getting the header80 * value requires the use of a safe conversion function.81 */82export class Header {83 84    /** @internal */85    constructor(public name: string, public type: HeaderType, public value?: HeaderValue) {}86 87    private static validateHeaderName(name: string) {88        if (name.length == 0 || name.length > AWS_MAXIMUM_EVENT_STREAM_HEADER_NAME_LENGTH) {89            throw new CrtError(`Event stream header name (${name}) is not valid`);90        }91    }92 93    /**94     * Create a new boolean-valued message header95     *96     * @param name name of the header97     * @param value value of the header98     */99    static newBoolean(name: string, value: boolean): Header {100        Header.validateHeaderName(name);101 102        if (value) {103            return new Header(name, HeaderType.BooleanTrue);104        } else {105            return new Header(name, HeaderType.BooleanFalse);106        }107    }108 109    /**110     * Create a new byte-valued message header111     *112     * @param name name of the header113     * @param value value of the header114     */115    static newByte(name: string, value: number): Header {116        Header.validateHeaderName(name);117 118        if (value >= eventstream_utils.MIN_INT8 && value <= eventstream_utils.MAX_INT8 && Number.isSafeInteger(value)) {119            return new Header(name, HeaderType.Byte, value);120        }121 122        throw new CrtError(`Illegal value for eventstream byte-valued header: ${value}`);123    }124 125    /**126     * Create a new 16-bit-integer-valued message header127     *128     * @param name name of the header129     * @param value value of the header130     */131    static newInt16(name: string, value: number): Header {132        Header.validateHeaderName(name);133 134        if (value >= eventstream_utils.MIN_INT16 && value <= eventstream_utils.MAX_INT16 && Number.isSafeInteger(value)) {135            return new Header(name, HeaderType.Int16, value);136        }137 138        throw new CrtError(`Illegal value for eventstream int16-valued header: ${value}`);139    }140 141    /**142     * Create a new 32-bit-integer-valued message header143     *144     * @param name name of the header145     * @param value value of the header146     */147    static newInt32(name: string, value: number): Header {148        Header.validateHeaderName(name);149 150        if (value >= eventstream_utils.MIN_INT32 && value <= eventstream_utils.MAX_INT32 && Number.isSafeInteger(value)) {151            return new Header(name, HeaderType.Int32, value);152        }153 154        throw new CrtError(`Illegal value for eventstream int32-valued header: ${value}`);155    }156 157    /**158     * Create a new 64-bit-integer-valued message header.  number cannot represent a full 64-bit integer range but159     * its usage is so common that this exists for convenience.  Internally, we always track 64 bit integers as160     * bigints.161     *162     * @param name name of the header163     * @param value value of the header164     */165    static newInt64FromNumber(name: string, value: number): Header {166        Header.validateHeaderName(name);167 168        if (Number.isSafeInteger(value)) {169            return new Header(name, HeaderType.Int64, eventstream_utils.marshalInt64BigintAsBuffer(BigInt(value)));170        }171 172        throw new CrtError(`Illegal value for eventstream int64-valued header: ${value}`);173    }174 175    /**176     * Create a new 64-bit-integer-valued message header from a big integer.177     *178     * @param name name of the header179     * @param value value of the header180     */181    static newInt64FromBigint(name: string, value: bigint): Header {182        Header.validateHeaderName(name);183 184        if (value >= eventstream_utils.MIN_INT64 && value <= eventstream_utils.MAX_INT64) {185            return new Header(name, HeaderType.Int64, eventstream_utils.marshalInt64BigintAsBuffer(value));186        }187 188        throw new CrtError(`Illegal value for eventstream int64-valued header: ${value}`);189    }190 191    /**192     * Create a new byte-buffer-valued message header193     *194     * @param name name of the header195     * @param value value of the header196     */197    static newByteBuffer(name: string, value: Payload): Header {198        Header.validateHeaderName(name);199 200        return new Header(name, HeaderType.ByteBuffer, value);201    }202 203    /**204     * Create a new string-valued message header205     *206     * @param name name of the header207     * @param value value of the header208     */209    static newString(name: string, value: string): Header {210        Header.validateHeaderName(name);211 212        return new Header(name, HeaderType.String, value);213    }214 215    /**216     * Create a new timestamp-valued message header from an integral value in seconds since epoch.217     *218     * @param name name of the header219     * @param value value of the header220     */221    static newTimeStampFromSecondsSinceEpoch(name: string, secondsSinceEpoch: number): Header {222        Header.validateHeaderName(name);223 224        if (Number.isSafeInteger(secondsSinceEpoch) && secondsSinceEpoch >= 0) {225            return new Header(name, HeaderType.Timestamp, secondsSinceEpoch);226        }227 228        throw new CrtError(`Illegal value for eventstream timestamp-valued header: ${secondsSinceEpoch}`);229    }230 231    /**232     * Create a new timestamp-valued message header from a date.233     *234     * @param name name of the header235     * @param value value of the header236     */237    static newTimeStampFromDate(name: string, date: Date): Header {238        Header.validateHeaderName(name);239 240        const secondsSinceEpoch: number = date.getTime();241        if (Number.isSafeInteger(secondsSinceEpoch)) {242            return new Header(name, HeaderType.Timestamp, secondsSinceEpoch);243        }244 245        throw new CrtError(`Illegal value for eventstream timestamp-valued header: ${date}`);246    }247 248    /**249     * Create a new UUID-valued message header.250     * WIP251     *252     * @param name name of the header253     * @param value value of the header254     */255    static newUUID(name: string, value: ArrayBuffer): Header {256        Header.validateHeaderName(name);257 258        if (value.byteLength == 16) {259            return new Header(name, HeaderType.UUID, value);260        }261 262        throw new CrtError(`Illegal value for eventstream uuid-valued header: ${value}`);263    }264 265    private toValue(type: HeaderType): any {266        if (type != this.type) {267            throw new CrtError(`Header of type (${this.type}) cannot be converted to type (${type})`);268        }269 270        return this.value;271    }272 273    /**274     * All conversion functions require the header's type to be appropriately matching.  There are no error-prone275     * flexible conversion helpers.276     */277 278    /**279     * Returns a boolean header's value.280     */281    asBoolean(): boolean {282        switch (this.type) {283            case HeaderType.BooleanFalse:284                return false;285            case HeaderType.BooleanTrue:286                return true;287            default:288                throw new CrtError(`Header of type (${this.type}) cannot be converted to type (boolean)`);289 290        }291    }292 293    /**294     * Returns a byte header's value.295     */296    asByte(): number {297        return this.toValue(HeaderType.Byte) as number;298    }299 300    /**301     * Returns a 16-bit integer header's value.302     */303    asInt16(): number {304        return this.toValue(HeaderType.Int16) as number;305    }306 307    /**308     * Returns a 32-bit integer header's value.309     */310    asInt32(): number {311        return this.toValue(HeaderType.Int32) as number;312    }313 314    /**315     * Returns a 64-bit integer header's value.316     */317    asInt64(): bigint {318        return eventstream_utils.unmarshalInt64BigintFromBuffer(this.toValue(HeaderType.Int64) as ArrayBuffer);319    }320 321    /**322     * Returns a byte buffer header's value.323     */324    asByteBuffer(): Payload {325        return this.toValue(HeaderType.ByteBuffer) as Payload;326    }327 328    /**329     * Returns a string header's value.330     */331    asString(): string {332        return this.toValue(HeaderType.String) as string;333    }334 335    /**336     * Returns a timestamp header's value (as seconds since epoch).337     */338    asTimestamp(): number {339        return this.toValue(HeaderType.Timestamp) as number;340    }341 342    /**343     * Returns a UUID header's value.344     */345    asUUID(): ArrayBuffer {346        return this.toValue(HeaderType.UUID) as ArrayBuffer;347    }348}349 350/**351 * Flags for messages in the event-stream RPC protocol.352 *353 * Flags may be XORed together.354 * Not all flags can be used with all message types, consult documentation.355 */356export enum MessageFlags {357 358    /** Nothing */359    None = 0,360 361    /**362     * Connection accepted363     *364     * If this flag is absent from a {@link MessageType.ConnectAck ConnectAck} message, the connection has been365     * rejected.366     */367    ConnectionAccepted = 0x1,368 369    /**370     * Terminate stream371     *372     * This message may be used with any message type.373     * The sender will close their connection after the message is written to the wire.374     * The receiver will close their connection after delivering the message to the user.375     */376    TerminateStream = 0x2,377}378 379/**380 *381 * Types of messages in the event-stream RPC protocol.382 * The {@link MessageType.ApplicationMessage Application} and {@link MessageType.ApplicationError Error} message types383 * may only be sent on streams, and will never arrive as a protocol message (stream-id 0).384 *385 * For all other message types, they may only be sent as protocol messages386 * (stream-id 0), and will never arrive as a stream message.387 *388 * Different message types expect specific headers and flags, consult documentation.389 */390export enum MessageType {391 392    /** Application message */393    ApplicationMessage = 0,394 395    /** Application error */396    ApplicationError = 1,397 398    /** Ping */399    Ping = 2,400 401    /** Ping response */402    PingResponse = 3,403 404    /** Connect */405    Connect = 4,406 407    /**408     * Connect acknowledgement409     *410     * If the {@link MessageFlags.ConnectionAccepted ConnectionAccepted} flag is not present, the connection has been rejected.411     */412    ConnectAck = 5,413 414    /**415     * Protocol error416     */417    ProtocolError = 6,418 419    /**420     * Internal error421     */422    InternalError = 7,423}424 425/**426 * Wrapper type for all event stream messages, whether they are protocol or application-level.427 */428export interface Message {429 430    /**431     * Type of message this is432     */433    type: MessageType,434 435    /**436     * Flags indicating additional boolean message properties437     */438    flags?: MessageFlags,439 440    /**441     * Message headers associated with this message442     */443    headers?: Array<Header>,444 445    /**446     * Actual message payload447     */448    payload?: Payload,449}450 451/** @internal */452function mapPodHeadersToJSHeaders(headers: Array<Header>) : Array<Header> {453    return Array.from(headers, (header) => {454        return new Header(header.name, header.type, header.value);455    });456}457 458/** @internal */459function mapPodMessageToJSMessage(message: Message) : Message {460    let jsMessage : Message = {461        type: message.type,462        flags: message.flags,463        payload: message.payload464    }465 466    if (message.headers) {467        jsMessage.headers = mapPodHeadersToJSHeaders(message.headers);468    }469 470    return jsMessage;471}472 473/**474 * Eventstream client connection configuration options.475 */476export interface ClientConnectionOptions {477 478    /**479     * Name of the host to connect to480     */481    hostName: string;482 483    /**484     * Port of the host to connect to485     */486    port: number;487 488    /**489     * Optional, additional socket options for the desired connection490     */491    socketOptions?: io.SocketOptions;492 493    /**494     * TLS context for the desired connection495     */496    tlsCtx?: io.ClientTlsContext;497}498 499/**500 * Options for opening a connection to an eventstream server501 */502export interface ConnectOptions {503    /**504     * Optional controller that allows the cancellation of asynchronous eventstream operations505     */506    cancelController?: cancel.ICancelController;507}508 509/**510 * Options for sending a protocol message over the client connection.511 */512export interface ProtocolMessageOptions {513 514    /**515     * Protocol message to send516     */517    message: Message;518 519    /**520     * Optional controller that allows the cancellation of asynchronous eventstream operations521     */522    cancelController?: cancel.ICancelController;523}524 525/**526 * Options for activating an event stream within the client connection.527 */528export interface ActivateStreamOptions {529 530    /**531     * Name of the operation that should be associated with this stream.532     */533    operation: string;534 535    /**536     * Application message to send as part of activating the stream.537     */538    message: Message;539 540    /**541     * Optional controller that allows the cancellation of asynchronous eventstream operations542     */543    cancelController?: cancel.ICancelController;544}545 546/**547 * Options for sending an application message within an event stream548 */549export interface StreamMessageOptions {550 551    /**552     * Application message to send.553     */554    message: Message;555 556    /**557     * Optional controller that allows the cancellation of asynchronous eventstream operations558     */559    cancelController?: cancel.ICancelController;560}561 562/**563 * Event emitted when an event stream connection has been fully shut down.  The connection is unusable afterwards, but564 * close() must still be called in order to release the native resources.565 */566export interface DisconnectionEvent {567 568    /**569     * Native error code.  Convert to a descriptive string with error_code_to_string()570     */571    errorCode: number;572}573 574/**575 * Event emitted when a message is received on an event stream connection.  When emitted by the connection, this576 * is a protocol message.  When emitted by a stream, it is an application message.577 */578export interface MessageEvent {579 580    /**581     * Event stream message received by the connection/stream.582     */583    message: Message;584}585 586/**587 * Signature for a handler that listens to event stream message events.588 */589export type MessageListener = (eventData: MessageEvent) => void;590 591/**592 * Signature for a handler that listens to event stream disconnection events.593 */594export type DisconnectionListener = (eventData: DisconnectionEvent) => void;595 596 597/**598 * @internal599 *600 * While not strictly necessary, the single-threaded nature of JS execution allows us to easily apply some601 * rigid constraints to the public API calls of our event stream objects.  This in turn reduces the complexity of the602 * binding cases we need to consider.603 *604 * This state value is the primary means by which we add and enforce these constraints to connection objects.605 *606 * Constraints enforced in the managed binding:607 *608 *  (1) close() may only be called once.  Once it has been called, nothing else may be called.609 *  (2) newStream() and sendMessage() may only be called after successful connection establishment and before the610 *      connection has been closed.611 *  (3) connect() may only be called once.  Combined with (1) and (2), this means that if connect() is called, it must612 *      be the first thing called.613 */614enum ClientConnectionState {615    None,616    Connecting,617    Connected,618    Disconnected,619    Closed,620}621 622/**623 * Wrapper for a network connection that fulfills the client-side event stream RPC protocol contract.624 *625 * The user **must** call close() on a connection once finished with it.  Once close() has been called, no more events626 * will be emitted and all public API invocations will trigger an exception.627 */628export class ClientConnection extends NativeResourceMixin(BufferedEventEmitter) {629 630    /**631     * Configures and creates a new ClientConnection instance632     *633     * @param config configuration options for the event stream connection634     */635    constructor(config: ClientConnectionOptions) {636        if (config === undefined) {637            throw new CrtError("Invalid configuration passed to eventstream ClientConnection constructor");638        }639 640        super();641 642        this.state = ClientConnectionState.None;643 644        this._super(crt_native.event_stream_client_connection_new(645            this,646            config,647            (connection: ClientConnection, errorCode: number) => { ClientConnection._s_on_disconnect(connection, errorCode); },648            (connection: ClientConnection, message: Message) => { ClientConnection._s_on_protocol_message(connection, message); },649            config.socketOptions ? config.socketOptions.native_handle() : null,650            config.tlsCtx ? config.tlsCtx.native_handle() : null651        ));652    }653 654    /**655     * Shuts down the connection (if active) and begins the process to release native resources associated with it by656     * having the native binding release the only reference to the extern object representing the connection.  Once657     * close() has been called, no more events will be emitted and all public API invocations will trigger an exception.658     *659     * Ultimately, the native resources will not be released until the connection has fully shut down and that660     * shutdown event has reached the libuv event loop.661     *662     * This function **must** be called for every ClientConnection instance or native resources will leak.663     */664    close() : void {665        if (this.state != ClientConnectionState.Closed) {666            this.state = ClientConnectionState.Closed;667 668            // invoke native binding close669            crt_native.event_stream_client_connection_close(this.native_handle());670        }671    }672 673    /**674     * Attempts to open a network connection to the configured remote endpoint.  Returned promise will be fulfilled if675     * the transport-level connection is successfully established, and rejected otherwise.676     *677     * connect() may only be called once.678     */679    async connect(options: ConnectOptions) : Promise<void> {680        let cleanupCancelListener : promise.PromiseCleanupFunctor | undefined = undefined;681 682        let connectPromise : Promise<void> = new Promise<void>((resolve, reject) => {683            if (!options) {684                reject(new CrtError("Invalid options passed to event stream ClientConnection.connect"));685                return;686            }687 688            if (this.state != ClientConnectionState.None) {689                reject(new CrtError(`Event stream connection in a state (${this.state}) where connect() is not allowed.`));690                return;691            }692 693            this.state = ClientConnectionState.Connecting;694 695            if (options.cancelController) {696                let cancel : () => void = () => {697                    reject(new CrtError(`Event stream connection connect() cancelled by external request.`));698                    setImmediate(() => { this.close(); });699                };700 701                cleanupCancelListener = options.cancelController.addListener(cancel);702                if (!cleanupCancelListener) {703                    return;704                }705            }706 707            function curriedPromiseCallback(connection: ClientConnection, errorCode: number){708                return ClientConnection._s_on_connection_setup(resolve, reject, connection, errorCode);709            }710 711            try {712                crt_native.event_stream_client_connection_connect(this.native_handle(), curriedPromiseCallback);713            } catch (e) {714                this.state = ClientConnectionState.Disconnected;715                reject(e);716            }717        });718 719        return promise.makeSelfCleaningPromise(connectPromise, cleanupCancelListener);720    }721 722    /**723     * Attempts to send an event stream protocol message over an open connection.724     *725     * @param options configuration -- including the message itself -- for sending a protocol message726     *727     * Returns a promise that will be fulfilled when the message is successfully flushed to the wire, and rejected if728     * an error occurs prior to that point.729     */730    async sendProtocolMessage(options: ProtocolMessageOptions) : Promise<void> {731        let cleanupCancelListener : promise.PromiseCleanupFunctor | undefined = undefined;732 733        let sendProtocolMessagePromise : Promise<void> = new Promise<void>((resolve, reject) => {734            try {735                if (!options) {736                    reject(new CrtError("Invalid options passed to event stream ClientConnection.sendProtocolMessage"));737                    return;738                }739 740                if (!this.isConnected()) {741                    reject(new CrtError(`Event stream connection in a state (${this.state}) where sending protocol messages is not allowed.`));742                    return;743                }744 745                if (options.cancelController) {746                    let cancel : () => void = () => {747                        reject(new CrtError(`Event stream connection sendProtocolMessage() cancelled by external request.`));748                        setImmediate(() => { this.close(); });749                    };750 751                    cleanupCancelListener = options.cancelController.addListener(cancel);752                    if (!cleanupCancelListener) {753                        return;754                    }755                }756 757                // invoke native binding send message;758                function curriedPromiseCallback(errorCode: number) {759                    return ClientConnection._s_on_connection_send_protocol_message_completion(resolve, reject, errorCode);760                }761 762                // invoke native binding send message;763                crt_native.event_stream_client_connection_send_protocol_message(this.native_handle(), options, curriedPromiseCallback);764            } catch (e) {765                reject(e);766            }767        });768 769        return promise.makeSelfCleaningPromise(sendProtocolMessagePromise, cleanupCancelListener);770    }771 772    /**773     * Returns true if the connection is currently open and ready-to-use, false otherwise.774     *775     * Internal note: Our notion of "connected" is intentionally not an invocation of776     * aws_event_stream_rpc_client_connection_is_open() (whose status is an out-of-sync race condition vs. our777     * well-defined client state)778     */779    isConnected() : boolean {780        return this.state == ClientConnectionState.Connected;781    }782 783    /**784     * Creates a new stream within the connection.785     */786    newStream() : ClientStream {787        if (!this.isConnected()) {788            throw new CrtError(`Event stream connection in a state (${this.state}) where creating new streams is forbidden.`);789        }790 791        return new ClientStream(this);792    }793 794    /**795     * Event emitted when the connection is closed for any reason.796     *797     * Listener type: {@link DisconnectionListener}798     *799     * @event800     */801    static DISCONNECTION : string = 'disconnection';802 803    /**804     * Event emitted when a protocol message is received from the remote endpoint805     *806     * Listener type: {@link MessageListener}807     *808     * @event809     */810    static PROTOCOL_MESSAGE : string = 'protocolMessage';811 812    on(event: 'disconnection', listener: DisconnectionListener): this;813 814    on(event: 'protocolMessage', listener: MessageListener): this;815 816    on(event: string | symbol, listener: (...args: any[]) => void): this {817        super.on(event, listener);818        return this;819    }820 821    private static _s_on_connection_setup(resolve : (value: (void | PromiseLike<void>)) => void, reject : (reason?: any) => void, connection: ClientConnection, errorCode: number) {822        if (errorCode == 0 && connection.state == ClientConnectionState.Connecting) {823            connection.state = ClientConnectionState.Connected;824            resolve();825        } else {826            if (connection.state != ClientConnectionState.Closed) {827                connection.state = ClientConnectionState.Disconnected;828            }829 830            reject(io.error_code_to_string(errorCode));831        }832    }833 834    private static _s_on_disconnect(connection: ClientConnection, errorCode: number) {835        if (connection.state != ClientConnectionState.Closed) {836            connection.state = ClientConnectionState.Disconnected;837        }838 839        process.nextTick(() => {840            connection.emit('disconnection', {errorCode: errorCode});841        });842    }843 844    private static _s_on_protocol_message(connection: ClientConnection, message: Message) {845        process.nextTick(() => {846            connection.emit('protocolMessage', {message: mapPodMessageToJSMessage(message)});847        });848    }849 850    private static _s_on_connection_send_protocol_message_completion(resolve : (value: (void | PromiseLike<void>)) => void, reject : (reason?: any) => void, errorCode: number) {851        if (errorCode == 0) {852            resolve();853        } else {854            reject(io.error_code_to_string(errorCode));855        }856    }857 858    private state : ClientConnectionState;859}860 861/**862 * Event emitted when the stream has ended.  At most one stream ended event will ever be emitted by a single863 * stream.864 */865export interface StreamEndedEvent {866}867 868/**869 * Signature for a handler that listens to stream ended events.870 */871export type StreamEndedListener = (eventData: StreamEndedEvent) => void;872 873/**874 * @internal875 *876 * While not strictly necessary, the single-threaded nature of JS execution allows us to easily apply some877 * rigid constraints to the public API calls of our event stream objects.  This in turn reduces the complexity of the878 * binding cases we need to consider.879 *880 * This state value is the primary means by which we add and enforce these constraints to stream objects.881 *882 * Constraints enforced in the managed binding:883 *884 *  (1) close() may only be called once.  Once it has been called, nothing else may be called.885 *  (2) sendMessage() may only be called after successful stream activation and before the886 *      stream has been closed.887 *  (3) activate() may only be called once.  Combined with (1) and (2), this means that if activate() is called, it must888 *      be the first thing called.889 */890enum ClientStreamState {891    None,892    Activating,893    Activated,894    Ended,895    Closed,896}897 898/**899 * Wrapper for an individual stream within an eventstream connection.900 *901 * The user **must** call close() on a stream once finished with it.  Once close() has been called, no more events902 * will be emitted and all public API invocations will trigger an exception.903 */904export class ClientStream extends NativeResourceMixin(BufferedEventEmitter) {905 906    constructor(connection: ClientConnection) {907        super();908 909        this._super(crt_native.event_stream_client_stream_new(910            this,911            connection.native_handle(),912            (stream: ClientStream) => { ClientStream._s_on_stream_ended(stream); },913            (stream: ClientStream, message: Message) => { ClientStream._s_on_stream_message(stream, message); },914        ));915 916        this.state = ClientStreamState.None;917    }918 919    /**920     * Shuts down the stream (if active) and begins the process to release native resources associated with it by921     * having the native binding release the only reference to the extern object representing the stream.  Once922     * close() has been called, no more events will be emitted and all public API invocations will trigger an exception.923     *924     * Ultimately, the native resources will not be released until the native stream has fully shut down and that925     * shutdown event has reached the libuv event loop.926     *927     * This function **must** be called for every ClientStream instance or native resources will leak.928     */929    close() : void {930        if (this.state != ClientStreamState.Closed) {931            this.state = ClientStreamState.Closed;932 933            crt_native.event_stream_client_stream_close(this.native_handle());934        }935    }936 937    /**938     * Activates the stream, allowing it to start sending and receiving messages.  The promise completes when939     * the activation message has been written to the wire.940     *941     * activate() may only be called once.942     *943     * @param options -- configuration data for stream activation, including operation name and initial message944     */945    async activate(options: ActivateStreamOptions) : Promise<void> {946        let cleanupCancelListener : promise.PromiseCleanupFunctor | undefined = undefined;947 948        let activatePromise : Promise<void> = new Promise<void>((resolve, reject) => {949            try {950                if (this.state != ClientStreamState.None) {951                    reject(new CrtError(`Event stream in a state (${this.state}) where activation is not allowed.`));952                    return;953                }954 955                /*956                 * Intentionally check this after the state check (so closed streams do not reach here).957                 * Intentionally mutate state the same way a failed synchronous call to native activate does.958                 */959                if (options === undefined) {960                    this.state = ClientStreamState.Ended;961                    reject(new CrtError("Invalid options passed to ClientStream.activate"));962                    return;963                }964 965                this.state = ClientStreamState.Activating;966 967                if (options.cancelController) {968                    let cancel : () => void = () => {969                        reject(new CrtError(`Event stream activate() cancelled by external request.`));970                        setImmediate(() => { this.close(); });971                    };972 973                    cleanupCancelListener = options.cancelController.addListener(cancel);974                    if (!cleanupCancelListener) {975                        return;976                    }977                }978 979                function curriedPromiseCallback(stream: ClientStream, errorCode: number){980                    return ClientStream._s_on_stream_activated(resolve, reject, stream, errorCode);981                }982 983                crt_native.event_stream_client_stream_activate(this.native_handle(), options, curriedPromiseCallback);984            } catch (e) {985                this.state = ClientStreamState.Ended;986                reject(e);987            }988        });989 990        return promise.makeSelfCleaningPromise<void>(activatePromise, cleanupCancelListener);991    }992 993    /**994     * Attempts to send an event stream message.995     *996     * @param options configuration -- including the message itself -- for sending a message997     *998     * Returns a promise that will be fulfilled when the message is successfully flushed to the wire, and rejected if999     * an error occurs prior to that point.1000     */1001    async sendMessage(options: StreamMessageOptions) : Promise<void> {1002        let cleanupCancelListener : promise.PromiseCleanupFunctor | undefined = undefined;1003 1004        let sendMessagePromise : Promise<void> = new Promise<void>((resolve, reject) => {1005            try {1006                if (!options) {1007                    reject(new CrtError("Invalid options passed to ClientStream.sendMessage"));1008                    return;1009                }1010 1011                if (this.state != ClientStreamState.Activated) {1012                    reject(new CrtError(`Event stream in a state (${this.state}) where sending messages is not allowed.`));1013                    return;1014                }1015 1016                if (options.cancelController) {1017                    let cancel : cancel.CancelListener = () => {1018                        reject(new CrtError(`Event stream sendMessage() cancelled by external request.`));1019                        setImmediate(() => { this.close(); });1020                    };1021 1022                    cleanupCancelListener = options.cancelController.addListener(cancel);1023                    if (!cleanupCancelListener) {1024                        return;1025                    }1026                }1027 1028                function curriedPromiseCallback(errorCode: number) {1029                    return ClientStream._s_on_stream_send_message_completion(resolve, reject, errorCode);1030                }1031 1032                // invoke native binding send message;1033                crt_native.event_stream_client_stream_send_message(this.native_handle(), options, curriedPromiseCallback);1034            } catch (e) {1035                reject(e);1036            }1037        });1038 1039        return promise.makeSelfCleaningPromise<void>(sendMessagePromise, cleanupCancelListener);1040    }1041 1042    /**1043     * Returns true if the stream is currently active and ready-to-use, false otherwise.1044     */1045    isActive() : boolean {1046        return this.state == ClientStreamState.Activated;1047    }1048 1049    /**1050     * Event emitted when the stream is shut down for any reason.1051     *1052     * Listener type: {@link StreamEndedListener}1053     *1054     * @event1055     */1056    static ENDED : string = 'ended';1057 1058    /**1059     * Event emitted when a stream message is received from the remote endpoint1060     *1061     * Listener type: {@link MessageListener}1062     *1063     * @event1064     */1065    static MESSAGE : string = 'message';1066 1067    on(event: 'ended', listener: StreamEndedListener): this;1068 1069    on(event: 'message', listener: MessageListener): this;1070 1071    on(event: string | symbol, listener: (...args: any[]) => void): this {1072        super.on(event, listener);1073        return this;1074    }1075 1076    private static _s_on_stream_activated(resolve : (value: (void | PromiseLike<void>)) => void, reject : (reason?: any) => void, stream: ClientStream, errorCode: number) {1077        if (errorCode == 0 && stream.state == ClientStreamState.Activating) {1078            stream.state = ClientStreamState.Activated;1079            resolve();1080        } else {1081            if (stream.state != ClientStreamState.Closed) {1082                stream.state = ClientStreamState.Ended;1083            }1084 1085            reject(io.error_code_to_string(errorCode));1086        }1087    }1088 1089    private static _s_on_stream_send_message_completion(resolve : (value: (void | PromiseLike<void>)) => void, reject : (reason?: any) => void, errorCode: number) {1090        if (errorCode == 0) {1091            resolve();1092        } else {1093            reject(io.error_code_to_string(errorCode));1094        }1095    }1096 1097    private static _s_on_stream_ended(stream: ClientStream) {1098        process.nextTick(() => {1099            stream.emit(ClientStream.ENDED, {});1100        });1101    }1102 1103    private static _s_on_stream_message(stream: ClientStream, message: Message) {1104        process.nextTick(() => {1105            stream.emit(ClientStream.MESSAGE, {message: mapPodMessageToJSMessage(message)});1106        });1107    }1108 1109    private state : ClientStreamState;1110}