async ingestFromStream()

in packages/azure-kusto-ingest/src/managedStreamingIngestClient.ts [94:119]

23 lines of code
8 McCabe index (conditional complexity)


    async ingestFromStream(
        stream: StreamDescriptor | Readable | ArrayBuffer,
        ingestionProperties?: IngestionPropertiesInput,
        clientRequestId?: string,
    ): Promise<KustoResponseDataSet | IngestionResult> {
        this.ensureOpen();
        const props = this._getMergedProps(ingestionProperties);
        let descriptor = stream instanceof StreamDescriptor ? stream : new StreamDescriptor(stream);
        let result = isNode ? await tryStreamToArray(descriptor.stream as Readable, maxStreamSize) : descriptor.stream;
        descriptor = new StreamDescriptor(result).merge(descriptor);
        let streamingResult: Promise<any> | null = null;
        // tryStreamToArray returns a Buffer in NodeJS impl if stream size is small enouph
        if ((isNode && result instanceof Buffer) || !isNode) {
            streamingResult = await this.streamWithRetries(
                isNode ? (descriptor.size ?? 0) : (descriptor.stream as ArrayBuffer).byteLength,
                descriptor,
                props,
                clientRequestId,
                result,
            );

            result = isNode ? readableToStream(result) : descriptor.stream;
        }

        return streamingResult ?? this.queuedIngestClient.ingestFromStream(new StreamDescriptor(result).merge(descriptor), props);
    }