Update Stream code & Add AudioWorklet support (#45)

* Push work

* Pull stream work

* On Chrome browsers, use Audio Worklet to capture audio

* Fix bugs merge introduced

* Add script URL code

* CR Feedback

* Disable empty test case

gulpfile.js

@@ -65,7 +65,10 @@
         }))
         .pipe(change(renameHttpsAgent))
         .pipe(gulp.dest('distrib/browser'));
-  }));
+  }, function () {
+      return gulp.src('./src/audioworklet/speech-processor.js')
+        .pipe(gulp.dest('./distrib/browser'));
+   }));
 
   gulp.task('compress', gulp.series("bundle", function(cb) {
     return pump([

src/audioworklet/speech-processor.js

@@ -0,0 +1,22 @@
+/* Implementation of the AudioWorkletProcessor
+ https://webaudio.github.io/web-audio-api/#audioworklet
+ This file will be loaded only in recent browsers that supports Audio worklet it is
+ currently in js because it needs to be in es6 */
+class SpeechProcessor extends AudioWorkletProcessor {
+  constructor(options) {
+    // The super constructor call is required.
+    super(options);
+  }
+
+  process(inputs, outputs) {
+    const input = inputs[0];
+    const output = outputs[0];
+    for (let channel = 0; channel < input.length; channel += 1) {
+      output[channel] = input[channel];
+    }
+    this.port.postMessage(output[0]);
+    return true;
+  }
+}
+
+registerProcessor('speech-processor', SpeechProcessor);

src/common.browser/Exports.ts

@@ -5,7 +5,6 @@ export * from "./ConsoleLoggingListener";
 export * from "./IRecorder";
 export * from "./MicAudioSource";
 export * from "./FileAudioSource";
-export * from "./OpusRecorder";
 export * from "./PCMRecorder";
 export * from "./WebsocketConnection";
 export * from "./WebsocketMessageAdapter";

src/common.browser/IRecorder.ts

@@ -6,4 +6,5 @@ import { Stream } from "../common/Exports";
 export interface IRecorder {
     record(context: AudioContext, mediaStream: MediaStream, outputStream: Stream<ArrayBuffer>): void;
     releaseMediaResources(context: AudioContext): void;
+    setWorkletUrl(url: string): void;
 }

src/common.browser/MicAudioSource.ts

@@ -20,6 +20,7 @@ import {
     AudioStreamNodeAttachingEvent,
     AudioStreamNodeDetachedEvent,
     AudioStreamNodeErrorEvent,
+    ChunkedArrayBufferStream,
     createNoDashGuid,
     Deferred,
     Events,
@@ -41,6 +42,8 @@ interface INavigatorUserMedia extends NavigatorUserMedia {
     msGetUserMedia?: (constraints: MediaStreamConstraints, successCallback: NavigatorUserMediaSuccessCallback, errorCallback: NavigatorUserMediaErrorCallback) => void;
 }
 
+export const AudioWorkletSourceURLPropertyName = "MICROPHONE-WorkletSourceUrl";
+
 export class MicAudioSource implements IAudioSource {
 
     private static readonly AUDIOFORMAT: AudioStreamFormatImpl = AudioStreamFormat.getDefaultInputFormat() as AudioStreamFormatImpl;
@@ -59,7 +62,15 @@ export class MicAudioSource implements IAudioSource {
 
     private privMicrophoneLabel: string;
 
-    public constructor(private readonly privRecorder: IRecorder, audioSourceId?: string, private readonly deviceId?: string) {
+    private privOutputChunkSize: number;
+
+    public constructor(
+        private readonly privRecorder: IRecorder,
+        outputChunkSize: number,
+        audioSourceId?: string,
+        private readonly deviceId?: string) {
+
+        this.privOutputChunkSize = outputChunkSize;
         this.privId = audioSourceId ? audioSourceId : createNoDashGuid();
         this.privEvents = new EventSource<AudioSourceEvent>();
     }
@@ -208,6 +219,14 @@ export class MicAudioSource implements IAudioSource {
         });
     }
 
+    public setProperty(name: string, value: string): void {
+        if (name === AudioWorkletSourceURLPropertyName) {
+            this.privRecorder.setWorkletUrl(value);
+        } else {
+            throw new Error("Property '" + name + "' is not supported on Microphone.");
+        }
+    }
+
     private getMicrophoneLabel(): Promise<string> {
         const defaultMicrophoneName: string = "microphone";
 
@@ -252,7 +271,7 @@ export class MicAudioSource implements IAudioSource {
     private listen = (audioNodeId: string): Promise<StreamReader<ArrayBuffer>> => {
         return this.turnOn()
             .onSuccessContinueWith<StreamReader<ArrayBuffer>>((_: boolean) => {
-                const stream = new Stream<ArrayBuffer>(audioNodeId);
+                const stream = new ChunkedArrayBufferStream(this.privOutputChunkSize, audioNodeId);
                 this.privStreams[audioNodeId] = stream;
 
                 try {

src/common.browser/OpusRecorder.ts

@@ -1,63 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT license.
-
-import { Stream } from "../common/Exports";
-import { IRecorder } from "./IRecorder";
-
-// getting around the build error for MediaRecorder as Typescript does not have a definition for this one.
-declare var MediaRecorder: any;
-
-export class OpusRecorder implements IRecorder {
-    private privMediaResources: IMediaResources;
-    private privMediaRecorderOptions: any;
-
-    constructor(options?: { mimeType: string, bitsPerSecond: number }) {
-        this.privMediaRecorderOptions = options;
-    }
-
-    public record = (context: AudioContext, mediaStream: MediaStream, outputStream: Stream<ArrayBuffer>): void => {
-        const mediaRecorder: any = new MediaRecorder(mediaStream, this.privMediaRecorderOptions);
-        const timeslice = 100; // this is in ms - 100 ensures that the chunk doesn't exceed the max size of chunk allowed in WS connection
-        mediaRecorder.ondataavailable = (dataAvailableEvent: any) => {
-            if (outputStream) {
-                const reader = new FileReader();
-                reader.readAsArrayBuffer(dataAvailableEvent.data);
-                reader.onloadend = (event: ProgressEvent) => {
-                    outputStream.writeStreamChunk({
-                        buffer: reader.result as ArrayBuffer,
-                        isEnd: false,
-                        timeReceived: Date.now(),
-                    });
-                };
-            }
-        };
-
-        this.privMediaResources = {
-            recorder: mediaRecorder,
-            stream: mediaStream,
-        };
-        mediaRecorder.start(timeslice);
-    }
-
-    public releaseMediaResources = (context: AudioContext): void => {
-        if (this.privMediaResources.recorder.state !== "inactive") {
-            this.privMediaResources.recorder.stop();
-        }
-        this.privMediaResources.stream.getTracks().forEach((track: any) => track.stop());
-    }
-}
-
-interface IMediaResources {
-    stream: MediaStream;
-    recorder: any;
-}
-
-/* Declaring this inline to avoid compiler warnings
-declare class MediaRecorder {
-    constructor(mediaStream: MediaStream, options: any);
-
-    public state: string;
-
-    public ondataavailable(dataAvailableEvent: any): void;
-    public stop(): void;
-}*/

src/common.browser/PCMRecorder.ts

@@ -6,6 +6,7 @@ import { IRecorder } from "./IRecorder";
 
 export class PcmRecorder implements IRecorder {
     private privMediaResources: IMediaResources;
+    private privSpeechProcessorScript: string; // speech-processor.js Url
 
     public record = (context: AudioContext, mediaStream: MediaStream, outputStream: Stream<ArrayBuffer>): void => {
         const desiredSampleRate = 16000;
@@ -19,7 +20,7 @@ export class PcmRecorder implements IRecorder {
                 bufferSize = 2048;
                 let audioSampleRate = context.sampleRate;
                 while (bufferSize < 16384 && audioSampleRate >= (2 * desiredSampleRate)) {
-                    bufferSize <<= 1 ;
+                    bufferSize <<= 1;
                     audioSampleRate >>= 1;
                 }
                 return context.createScriptProcessor(bufferSize, 1, 1);
@@ -47,14 +48,55 @@ export class PcmRecorder implements IRecorder {
 
         const micInput = context.createMediaStreamSource(mediaStream);
 
-        this.privMediaResources = {
-            scriptProcessorNode: scriptNode,
-            source: micInput,
-            stream: mediaStream,
-        };
+        // https://webaudio.github.io/web-audio-api/#audioworklet
+        // Using AudioWorklet to improve audio quality and avoid audio glitches due to blocking the UI thread
 
-        micInput.connect(scriptNode);
-        scriptNode.connect(context.destination);
+        if (!!this.privSpeechProcessorScript && !!context.audioWorklet) {
+            context.audioWorklet
+                .addModule(this.privSpeechProcessorScript)
+                .then(() => {
+                    const workletNode = new AudioWorkletNode(context, "speech-processor");
+                    workletNode.port.onmessage = (ev: MessageEvent) => {
+                        const inputFrame: Float32Array = ev.data as Float32Array;
+
+                        if (outputStream && !outputStream.isClosed) {
+                            const waveFrame = waveStreamEncoder.encode(needHeader, inputFrame);
+                            if (!!waveFrame) {
+                                outputStream.writeStreamChunk({
+                                    buffer: waveFrame,
+                                    isEnd: false,
+                                    timeReceived: Date.now(),
+                                });
+                                needHeader = false;
+                            }
+                        }
+                    };
+                    micInput.connect(workletNode);
+                    workletNode.connect(context.destination);
+                    this.privMediaResources = {
+                        scriptProcessorNode: workletNode,
+                        source: micInput,
+                        stream: mediaStream,
+                    };
+                })
+                .catch(() => {
+                    micInput.connect(scriptNode);
+                    scriptNode.connect(context.destination);
+                    this.privMediaResources = {
+                        scriptProcessorNode: scriptNode,
+                        source: micInput,
+                        stream: mediaStream,
+                    };
+                });
+        } else {
+            micInput.connect(scriptNode);
+            scriptNode.connect(context.destination);
+            this.privMediaResources = {
+                scriptProcessorNode: scriptNode,
+                source: micInput,
+                stream: mediaStream,
+            };
+        }
     }
 
     public releaseMediaResources = (context: AudioContext): void => {
@@ -70,10 +112,14 @@ export class PcmRecorder implements IRecorder {
             }
         }
     }
+
+    public setWorkletUrl(url: string): void {
+        this.privSpeechProcessorScript = url;
+    }
 }
 
 interface IMediaResources {
     source: MediaStreamAudioSourceNode;
-    scriptProcessorNode: ScriptProcessorNode;
+    scriptProcessorNode: ScriptProcessorNode | AudioWorkletNode;
     stream: MediaStream;
 }

src/common/ChunkedArrayBufferStream.ts

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+import { IStreamChunk, Stream } from "./Exports";
+
+export class ChunkedArrayBufferStream extends Stream<ArrayBuffer> {
+    private privTargetChunkSize: number;
+    private privNextBufferToWrite: ArrayBuffer;
+    private privNextBufferStartTime: number;
+    private privNextBufferReadyBytes: number;
+
+    constructor(targetChunkSize: number, streamId?: string) {
+        super(streamId);
+        this.privTargetChunkSize = targetChunkSize;
+        this.privNextBufferReadyBytes = 0;
+    }
+
+    public writeStreamChunk(chunk: IStreamChunk<ArrayBuffer>): void {
+        // No pending write, and the buffer is the right size so write it.
+        if (chunk.isEnd ||
+            (0 === this.privNextBufferReadyBytes && chunk.buffer.byteLength === this.privTargetChunkSize)) {
+            super.writeStreamChunk(chunk);
+            return;
+        }
+
+        let bytesCopiedFromBuffer: number = 0;
+
+        while (bytesCopiedFromBuffer < chunk.buffer.byteLength) {
+            // Fill the next buffer.
+            if (undefined === this.privNextBufferToWrite) {
+                this.privNextBufferToWrite = new ArrayBuffer(this.privTargetChunkSize);
+                this.privNextBufferStartTime = chunk.timeReceived;
+            }
+
+            // Find out how many bytes we can copy into the read buffer.
+            const bytesToCopy: number = Math.min(chunk.buffer.byteLength - bytesCopiedFromBuffer, this.privTargetChunkSize - this.privNextBufferReadyBytes);
+            const targetView: Uint8Array = new Uint8Array(this.privNextBufferToWrite);
+            const sourceView: Uint8Array = new Uint8Array(chunk.buffer.slice(bytesCopiedFromBuffer, bytesToCopy + bytesCopiedFromBuffer));
+
+            targetView.set(sourceView, this.privNextBufferReadyBytes);
+            this.privNextBufferReadyBytes += bytesToCopy;
+            bytesCopiedFromBuffer += bytesToCopy;
+
+            // Are we ready to write?
+            if (this.privNextBufferReadyBytes === this.privTargetChunkSize) {
+                super.writeStreamChunk({
+                    buffer: this.privNextBufferToWrite,
+                    isEnd: false,
+                    timeReceived: this.privNextBufferStartTime,
+                });
+                this.privNextBufferReadyBytes = 0;
+                this.privNextBufferToWrite = undefined;
+            }
+        }
+    }
+
+    public close(): void {
+        // Send whatever is pending, then close the base class.
+        if (0 !== this.privNextBufferReadyBytes && !this.isClosed) {
+            super.writeStreamChunk({
+                buffer: this.privNextBufferToWrite.slice(0, this.privNextBufferReadyBytes),
+                isEnd: false,
+                timeReceived: this.privNextBufferStartTime,
+            });
+        }
+
+        super.close();
+    }
+}

src/common/Exports.ts

@@ -25,3 +25,4 @@ export * from "./RawWebsocketMessage";
 export * from "./RiffPcmEncoder";
 export * from "./Stream";
 export { TranslationStatus } from "../common.speech/TranslationStatus";
+export * from "./ChunkedArrayBufferStream";

src/common/IAudioSource.ts

@@ -18,6 +18,8 @@ export interface IAudioSource {
     events: EventSource<AudioSourceEvent>;
     format: AudioStreamFormat;
     deviceInfo: Promise<ISpeechConfigAudioDevice>;
+    setProperty?(name: string, value: string): void;
+    getProperty?(name: string, def?: string): string;
 }
 
 export interface IAudioStreamNode extends IDetachable {

src/common/Stream.ts

@@ -52,7 +52,7 @@ export class Stream<TBuffer> {
             });
     }
 
-    public close = (): void => {
+    public close(): void {
         if (!this.privIsEnded) {
             this.writeStreamChunk({
                 buffer: null,
@@ -63,7 +63,7 @@ export class Stream<TBuffer> {
         }
     }
 
-    public writeStreamChunk = (streamChunk: IStreamChunk<TBuffer>): void => {
+    public writeStreamChunk(streamChunk: IStreamChunk<TBuffer>): void {
         this.throwIfClosed();
         this.privStreambuffer.push(streamChunk);
         for (const readerId in this.privReaderQueues) {

src/sdk/Audio/AudioConfig.ts

@@ -5,8 +5,9 @@ import { AudioStreamFormat } from "../../../src/sdk/Exports";
 import { FileAudioSource, MicAudioSource, PcmRecorder } from "../../common.browser/Exports";
 import { ISpeechConfigAudioDevice } from "../../common.speech/Exports";
 import { AudioSourceEvent, EventSource, IAudioSource, IAudioStreamNode, Promise } from "../../common/Exports";
-import { AudioInputStream, PullAudioInputStreamCallback } from "../Exports";
-import { PullAudioInputStreamImpl, PushAudioInputStreamImpl } from "./AudioInputStream";
+import { Contracts } from "../Contracts";
+import { AudioInputStream, PropertyCollection, PropertyId, PullAudioInputStreamCallback } from "../Exports";
+import { bufferSize, PullAudioInputStreamImpl, PushAudioInputStreamImpl } from "./AudioInputStream";
 
 /**
  * Represents audio input configuration used for specifying what type of input to use (microphone, file, stream).
@@ -22,7 +23,7 @@ export abstract class AudioConfig {
      */
     public static fromDefaultMicrophoneInput(): AudioConfig {
         const pcmRecorder = new PcmRecorder();
-        return new AudioConfigImpl(new MicAudioSource(pcmRecorder));
+        return new AudioConfigImpl(new MicAudioSource(pcmRecorder, bufferSize));
     }
 
     /**
@@ -36,7 +37,7 @@ export abstract class AudioConfig {
      */
     public static fromMicrophoneInput(deviceId?: string): AudioConfig {
         const pcmRecorder = new PcmRecorder();
-        return new AudioConfigImpl(new MicAudioSource(pcmRecorder, deviceId));
+        return new AudioConfigImpl(new MicAudioSource(pcmRecorder, bufferSize, deviceId));
     }
 
     /**
@@ -81,6 +82,28 @@ export abstract class AudioConfig {
      * @public
      */
     public abstract close(): void;
+
+    /**
+     * Sets an arbitrary property.
+     * @member SpeechConfig.prototype.setProperty
+     * @function
+     * @public
+     * @param {string} name - The name of the property to set.
+     * @param {string} value - The new value of the property.
+     */
+    public abstract setProperty(name: string, value: string): void;
+
+    /**
+     * Returns the current value of an arbitrary property.
+     * @member SpeechConfig.prototype.getProperty
+     * @function
+     * @public
+     * @param {string} name - The name of the property to query.
+     * @param {string} def - The value to return in case the property is not known.
+     * @returns {string} The current value, or provided default, of the given property.
+     */
+    public abstract getProperty(name: string, def?: string): string;
+
 }
 
 /**
@@ -178,6 +201,27 @@ export class AudioConfigImpl extends AudioConfig implements IAudioSource {
         return this.privSource.events;
     }
 
+    public setProperty(name: string, value: string): void {
+        Contracts.throwIfNull(value, "value");
+
+        if (undefined !== this.privSource.setProperty) {
+            this.privSource.setProperty(name, value);
+        } else {
+            throw new Error("This AudioConfig instance does not support setting properties.");
+        }
+
+    }
+
+    public getProperty(name: string, def?: string): string {
+        if (undefined !== this.privSource.getProperty) {
+            return this.privSource.getProperty(name, def);
+        } else {
+            throw new Error("This AudioConfig instance does not support getting properties.");
+        }
+
+        return def;
+    }
+
     public get deviceInfo(): Promise<ISpeechConfigAudioDevice> {
         return this.privSource.deviceInfo;
     }

src/sdk/Audio/AudioInputStream.ts

@@ -14,6 +14,7 @@ import {
     AudioStreamNodeAttachedEvent,
     AudioStreamNodeAttachingEvent,
     AudioStreamNodeDetachedEvent,
+    ChunkedArrayBufferStream,
     Events,
     EventSource,
     IAudioSource,
@@ -27,7 +28,7 @@ import {
 import { AudioStreamFormat, PullAudioInputStreamCallback } from "../Exports";
 import { AudioStreamFormatImpl } from "./AudioStreamFormat";
 
-const bufferSize: number = 4096;
+export const bufferSize: number = 4096;
 
 /**
  * Represents audio input stream used for custom audio input configurations.
@@ -97,7 +98,7 @@ export abstract class PushAudioInputStream extends AudioInputStream {
      * @returns {PushAudioInputStream} The push audio input stream being created.
      */
     public static create(format?: AudioStreamFormat): PushAudioInputStream {
-        return new PushAudioInputStreamImpl(format);
+        return new PushAudioInputStreamImpl(bufferSize, format);
     }
 
     /**
@@ -129,14 +130,14 @@ export class PushAudioInputStreamImpl extends PushAudioInputStream implements IA
     private privFormat: AudioStreamFormatImpl;
     private privId: string;
     private privEvents: EventSource<AudioSourceEvent>;
-    private privStream: Stream<ArrayBuffer> = new Stream<ArrayBuffer>();
+    private privStream: Stream<ArrayBuffer>;
 
     /**
      * Creates and initalizes an instance with the given values.
      * @constructor
      * @param {AudioStreamFormat} format - The audio stream format.
      */
-    public constructor(format?: AudioStreamFormat) {
+    public constructor(chunkSize: number, format?: AudioStreamFormat) {
         super();
         if (format === undefined) {
             this.privFormat = AudioStreamFormatImpl.getDefaultInputFormat();
@@ -145,6 +146,7 @@ export class PushAudioInputStreamImpl extends PushAudioInputStream implements IA
         }
         this.privEvents = new EventSource<AudioSourceEvent>();
         this.privId = createNoDashGuid();
+        this.privStream = new ChunkedArrayBufferStream(chunkSize);
     }
 
     /**
@@ -162,25 +164,11 @@ export class PushAudioInputStreamImpl extends PushAudioInputStream implements IA
      * @param {ArrayBuffer} dataBuffer - The audio buffer of which this function will make a copy.
      */
     public write(dataBuffer: ArrayBuffer): void {
-        // Break the data up into smaller chunks if needed.
-        let i: number;
-        const time: number = Date.now();
-
-        for (i = bufferSize - 1; i < dataBuffer.byteLength; i += bufferSize) {
-            this.privStream.writeStreamChunk({
-                buffer: dataBuffer.slice(i - (bufferSize - 1), i + 1),
-                isEnd: false,
-                timeReceived: time,
-            });
-        }
-
-        if ((i - (bufferSize - 1)) !== dataBuffer.byteLength) {
-            this.privStream.writeStreamChunk({
-                buffer: dataBuffer.slice(i - (bufferSize - 1), dataBuffer.byteLength),
-                isEnd: false,
-                timeReceived: time
-            });
-        }
+        this.privStream.writeStreamChunk({
+            buffer: dataBuffer,
+            isEnd: false,
+            timeReceived: Date.now()
+        });
     }
 
     /**
@@ -381,18 +369,43 @@ export class PullAudioInputStreamImpl extends PullAudioInputStream implements IA
                         return audioNodeId;
                     },
                     read: (): Promise<IStreamChunk<ArrayBuffer>> => {
-                        const readBuff: ArrayBuffer = new ArrayBuffer(bufferSize);
-                        const pulledBytes: number = this.privCallback.read(readBuff);
+                        let totalBytes: number = 0;
+                        let transmitBuff: ArrayBuffer;
+
+                        // Until we have the minimum number of bytes to send in a transmission, keep asking for more.
+                        while (totalBytes < bufferSize) {
+                            // Sizing the read buffer to the delta between the perfect size and what's left means we won't ever get too much
+                            // data back.
+                            const readBuff: ArrayBuffer = new ArrayBuffer(bufferSize - totalBytes);
+                            const pulledBytes: number = this.privCallback.read(readBuff);
+
+                            // If there is no return buffer yet defined, set the return buffer to the that was just populated.
+                            // This was, if we have enough data there's no copy penalty, but if we don't we have a buffer that's the
+                            // preferred size allocated.
+                            if (undefined === transmitBuff) {
+                                transmitBuff = readBuff;
+                            } else {
+                                // Not the first bite at the apple, so fill the return buffer with the data we got back.
+                                const intView: Int8Array = new Int8Array(transmitBuff);
+                                intView.set(new Int8Array(readBuff), totalBytes);
+                            }
+
+                            // If there are no bytes to read, just break out and be done.
+                            if (0 === pulledBytes) {
+                                break;
+                            }
+
+                            totalBytes += pulledBytes;
+                        }
 
                         return PromiseHelper.fromResult<IStreamChunk<ArrayBuffer>>({
-                            buffer: readBuff.slice(0, pulledBytes),
-                            isEnd: this.privIsClosed,
+                            buffer: transmitBuff.slice(0, totalBytes),
+                            isEnd: this.privIsClosed || totalBytes === 0,
                             timeReceived: Date.now(),
                         });
                     },
                 };
             });
-
     }
 
     public detach(audioNodeId: string): void {

tests/IntentRecognizerTests.ts

@@ -890,56 +890,6 @@ describe.each([true, false])("Service based tests", (forceNodeWebSocket: boolean
     }, 35000);
 });
 
-test("Bad DataType for PushStreams results in error", (done: jest.DoneCallback) => {
-    // tslint:disable-next-line:no-console
-    console.info("Name: Bad DataType for PushStreams results in error");
-
-    const s: sdk.SpeechConfig = BuildSpeechConfig();
-    objsToClose.push(s);
-
-    const p: sdk.PushAudioInputStream = sdk.AudioInputStream.createPushStream();
-    const config: sdk.AudioConfig = sdk.AudioConfig.fromStreamInput(p);
-
-    // Wrong data type for ReadStreams
-    fs.createReadStream(Settings.WaveFile).on("data", (buffer: ArrayBuffer) => {
-        p.write(buffer);
-    }).on("end", () => {
-        p.close();
-    });
-
-    const r: sdk.IntentRecognizer = new sdk.IntentRecognizer(s, config);
-    objsToClose.push(r);
-
-    expect(r).not.toBeUndefined();
-    expect(r instanceof sdk.Recognizer);
-
-    r.canceled = (r: sdk.Recognizer, e: sdk.IntentRecognitionCanceledEventArgs) => {
-        try {
-            expect(e.errorDetails).not.toBeUndefined();
-            expect(sdk.CancellationReason[e.reason]).toEqual(sdk.CancellationReason[sdk.CancellationReason.Error]);
-            expect(sdk.CancellationErrorCode[e.errorCode]).toEqual(sdk.CancellationErrorCode[sdk.CancellationErrorCode.RuntimeError]);
-        } catch (error) {
-            done.fail(error);
-        }
-    };
-
-    r.recognizeOnceAsync(
-        (p2: sdk.SpeechRecognitionResult) => {
-            const res: sdk.SpeechRecognitionResult = p2;
-            try {
-                expect(res).not.toBeUndefined();
-                expect(res.errorDetails).not.toBeUndefined();
-                expect(sdk.ResultReason[res.reason]).toEqual(sdk.ResultReason[sdk.ResultReason.Canceled]);
-                done();
-            } catch (error) {
-                done.fail(error);
-            }
-        },
-        (error: string) => {
-            done.fail(error);
-        });
-});
-
 test("Ambiguous Speech default as expected", (done: jest.DoneCallback) => {
     // tslint:disable-next-line:no-console
     console.info("Name: Ambiguous Speech default as expected");

tests/PullInputStreamTests.ts

@@ -0,0 +1,169 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+import {
+    IAudioStreamNode,
+    IStreamChunk,
+} from "../src/common/Exports";
+import {
+    bufferSize,
+    PullAudioInputStreamImpl,
+} from "../src/sdk/Audio/AudioInputStream";
+import { Settings } from "./Settings";
+
+beforeAll(() => {
+    // Override inputs, if necessary
+    Settings.LoadSettings();
+});
+
+// Test cases are run linerally, the only other mechanism to demark them in the output is to put a console line in each case and
+// report the name.
+// tslint:disable-next-line:no-console
+beforeEach(() => console.info("---------------------------------------Starting test case-----------------------------------"));
+
+test("PullStream correctly reports bytes read", (done: jest.DoneCallback) => {
+
+    let readReturnVal: number = bufferSize;
+
+    const stream: PullAudioInputStreamImpl = new PullAudioInputStreamImpl({
+        close: (): void => {
+            return;
+        },
+        read: (dataBuffer: ArrayBuffer): number => {
+            return readReturnVal;
+        },
+    });
+
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readArray: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(readArray.buffer.byteLength).toEqual(readReturnVal);
+                expect(readArray.isEnd).toEqual(false);
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+
+    readReturnVal = bufferSize;
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readArray: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(readArray.buffer.byteLength).toEqual(readReturnVal);
+                expect(readArray.isEnd).toEqual(false);
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+
+    readReturnVal = bufferSize;
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readArray: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(readArray.buffer.byteLength).toEqual(readReturnVal);
+                expect(readArray.isEnd).toEqual(false);
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+});
+
+test("Returning 0 marks end of stream", (done: jest.DoneCallback) => {
+    const stream: PullAudioInputStreamImpl = new PullAudioInputStreamImpl({
+        close: (): void => {
+            return;
+        },
+        read: (dataBuffer: ArrayBuffer): number => {
+            return 0;
+        },
+    });
+
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readArray: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(readArray.buffer.byteLength).toEqual(0);
+                expect(readArray.isEnd).toEqual(true);
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+});
+
+// Validates that the pull stream will request more bytes until it has been satisfied.
+// Verifies no data is lost.
+test("Pull stream accumulates bytes", (done: jest.DoneCallback) => {
+    let bytesReturned: number = 0;
+    const stream: PullAudioInputStreamImpl = new PullAudioInputStreamImpl({
+        close: (): void => {
+            return;
+        },
+        read: (dataBuffer: ArrayBuffer): number => {
+            const returnArray: Uint8Array = new Uint8Array(dataBuffer);
+            returnArray[0] = bytesReturned++ % 256;
+            return 1;
+        },
+    });
+
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readBuffer: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(bytesReturned).toEqual(bufferSize);
+                expect(readBuffer.buffer.byteLength).toEqual(bufferSize);
+                const readArray: Uint8Array = new Uint8Array(readBuffer.buffer);
+
+                for (let i: number = 0; i < bytesReturned; i++) {
+                    expect(readArray[i]).toEqual(i % 256);
+                }
+
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+});
+
+// Validates that the pull stream will request more bytes until there are no more.
+// Verifies no data is lost.
+test("Pull stream accumulates bytes while available", (done: jest.DoneCallback) => {
+    let bytesReturned: number = 0;
+    const stream: PullAudioInputStreamImpl = new PullAudioInputStreamImpl({
+        close: (): void => {
+            return;
+        },
+        read: (dataBuffer: ArrayBuffer): number => {
+            const returnArray: Uint8Array = new Uint8Array(dataBuffer);
+            if (bytesReturned < bufferSize / 2) {
+                returnArray[0] = bytesReturned++ % 256;
+                return 1;
+            } else {
+                return 0;
+            }
+        },
+    });
+
+    stream.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        audioNode.read().onSuccessContinueWith((readBuffer: IStreamChunk<ArrayBuffer>) => {
+            try {
+                expect(bytesReturned).toEqual(bufferSize / 2);
+                expect(readBuffer.buffer.byteLength).toEqual(bufferSize / 2);
+                const readArray: Uint8Array = new Uint8Array(readBuffer.buffer);
+
+                for (let i: number = 0; i < bytesReturned; i++) {
+                    expect(readArray[i]).toEqual(i % 256);
+                }
+
+                done();
+            } catch (error) {
+                done.fail(error);
+            }
+        });
+    });
+});

tests/PushInputStreamTests.ts

@@ -0,0 +1,235 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+import { setTimeout } from "timers";
+import {
+    IAudioStreamNode,
+    IStreamChunk,
+} from "../src/common/Exports";
+import {
+    bufferSize,
+    PushAudioInputStreamImpl,
+} from "../src/sdk/Audio/AudioInputStream";
+import { Settings } from "./Settings";
+
+beforeAll(() => {
+    // Override inputs, if necessary
+    Settings.LoadSettings();
+});
+
+// Test cases are run linerally, the only other mechanism to demark them in the output is to put a console line in each case and
+// report the name.
+// tslint:disable-next-line:no-console
+beforeEach(() => console.info("---------------------------------------Starting test case-----------------------------------"));
+
+test("Push segments into small blocks", (done: jest.DoneCallback) => {
+    const ps: PushAudioInputStreamImpl = new PushAudioInputStreamImpl(bufferSize);
+
+    const ab: ArrayBuffer = new ArrayBuffer(bufferSize * 4);
+    const abView: Uint8Array = new Uint8Array(ab);
+    for (let i: number = 0; i < bufferSize * 4; i++) {
+        abView[i] = i % 256;
+    }
+
+    let j: number = 0;
+    for (j = 0; j < bufferSize * 4; j += 100) {
+        ps.write(ab.slice(j, j + 100));
+    }
+
+    ps.write(ab.slice(j));
+
+    ps.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+
+        let bytesRead: number = 0;
+
+        const readLoop = () => {
+            audioNode.read().onSuccessContinueWith((audioBuffer: IStreamChunk<ArrayBuffer>) => {
+                try {
+                    expect(audioBuffer.buffer.byteLength).toBeGreaterThanOrEqual(bufferSize);
+                    expect(audioBuffer.buffer.byteLength).toBeLessThanOrEqual(bufferSize);
+                    const readView: Uint8Array = new Uint8Array(audioBuffer.buffer);
+                    for (let i: number = 0; i < audioBuffer.buffer.byteLength; i++) {
+                        expect(readView[i]).toEqual(bytesRead++ % 256);
+                    }
+                } catch (error) {
+                    done.fail(error);
+                }
+
+                if (bytesRead < bufferSize * 4) {
+                    readLoop();
+                } else {
+                    done();
+                }
+            });
+        };
+
+        readLoop();
+    });
+});
+
+test("Stream returns all data when closed", (done: jest.DoneCallback) => {
+    const ps: PushAudioInputStreamImpl = new PushAudioInputStreamImpl(bufferSize);
+
+    const ab: ArrayBuffer = new ArrayBuffer(bufferSize * 4);
+    const abView: Uint8Array = new Uint8Array(ab);
+    for (let i: number = 0; i < bufferSize * 4; i++) {
+        abView[i] = i % 256;
+    }
+
+    let j: number = 0;
+    for (j = 0; j < bufferSize * 4; j += 100) {
+        ps.write(ab.slice(j, j + 100));
+    }
+
+    ps.write(ab.slice(j));
+    ps.close();
+
+    ps.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        let bytesRead: number = 0;
+
+        const readLoop = () => {
+            audioNode.read().onSuccessContinueWith((audioBuffer: IStreamChunk<ArrayBuffer>) => {
+                try {
+                    expect(audioBuffer).not.toBeUndefined();
+                    if (bytesRead === bufferSize * 4) {
+                        expect(audioBuffer.isEnd).toEqual(true);
+                        expect(audioBuffer.buffer).toEqual(null);
+                        done();
+                    } else {
+                        expect(audioBuffer.buffer).not.toBeUndefined();
+                        expect(audioBuffer.isEnd).toEqual(false);
+
+                        const readView: Uint8Array = new Uint8Array(audioBuffer.buffer);
+                        for (let i: number = 0; i < audioBuffer.buffer.byteLength; i++) {
+                            expect(readView[i]).toEqual(bytesRead++ % 256);
+                        }
+
+                        readLoop();
+                    }
+
+                } catch (error) {
+                    done.fail(error);
+                }
+
+            });
+        };
+
+        readLoop();
+    });
+});
+
+test("Stream blocks when not closed", (done: jest.DoneCallback) => {
+    const ps: PushAudioInputStreamImpl = new PushAudioInputStreamImpl(bufferSize);
+
+    const ab: ArrayBuffer = new ArrayBuffer(bufferSize * 4);
+    const abView: Uint8Array = new Uint8Array(ab);
+    for (let i: number = 0; i < bufferSize * 4; i++) {
+        abView[i] = i % 256;
+    }
+
+    let j: number = 0;
+    for (j = 0; j < bufferSize * 4; j += 100) {
+        ps.write(ab.slice(j, j + 100));
+    }
+
+    ps.write(ab.slice(j));
+
+    ps.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        let bytesRead: number = 0;
+        let readCallCount: number = 0;
+        let shouldBeEnd: boolean = false;
+
+        const readLoop = () => {
+            audioNode.read().onSuccessContinueWith((audioBuffer: IStreamChunk<ArrayBuffer>) => {
+                readCallCount++;
+                try {
+
+                    expect(audioBuffer).not.toBeUndefined();
+                    if (!shouldBeEnd) {
+                        expect(audioBuffer.buffer).not.toBeUndefined();
+
+                        const readView: Uint8Array = new Uint8Array(audioBuffer.buffer);
+                        for (let i: number = 0; i < audioBuffer.buffer.byteLength; i++) {
+                            expect(readView[i]).toEqual(bytesRead++ % 256);
+                        }
+
+                        if (bytesRead === bufferSize * 4) {
+                            // The next call should block.
+                            const currentReadCount: number = readCallCount;
+                            // Schedule a check that the number of calls has not increased.
+                            setTimeout(() => {
+                                try {
+                                    expect(readCallCount).toEqual(currentReadCount);
+                                    shouldBeEnd = true;
+                                    // Release the blocking read and finish when it does.
+                                    ps.close();
+                                } catch (error) {
+                                    done.fail(error);
+                                }
+                            }, 2000);
+                        }
+                        readLoop();
+
+                    } else {
+                        expect(audioBuffer.buffer).toEqual(null);
+                        expect(audioBuffer.isEnd).toEqual(true);
+                        done();
+                    }
+                } catch (error) {
+                    done.fail(error);
+                }
+
+            });
+        };
+
+        readLoop();
+    });
+});
+
+test("nonAligned data is fine", (done: jest.DoneCallback) => {
+    const ps: PushAudioInputStreamImpl = new PushAudioInputStreamImpl(bufferSize);
+
+    const dataSize: number = bufferSize * 1.25;
+    const ab: ArrayBuffer = new ArrayBuffer(dataSize);
+    const abView: Uint8Array = new Uint8Array(ab);
+    for (let i: number = 0; i < dataSize; i++) {
+        abView[i] = i % 256;
+    }
+
+    ps.write(ab);
+    ps.close();
+
+    ps.attach("id").onSuccessContinueWith((audioNode: IAudioStreamNode) => {
+        let bytesRead: number = 0;
+
+        const readLoop = () => {
+            audioNode.read().onSuccessContinueWith((audioBuffer: IStreamChunk<ArrayBuffer>) => {
+                try {
+                    expect(audioBuffer).not.toBeUndefined();
+
+                    if (bytesRead === dataSize) {
+                        expect(audioBuffer.isEnd).toEqual(true);
+                        expect(audioBuffer.buffer).toEqual(null);
+                        done();
+                    } else {
+                        expect(audioBuffer.buffer).not.toBeUndefined();
+                        expect(audioBuffer.isEnd).toEqual(false);
+
+                        const readView: Uint8Array = new Uint8Array(audioBuffer.buffer);
+                        for (let i: number = 0; i < audioBuffer.buffer.byteLength; i++) {
+                            expect(readView[i]).toEqual(bytesRead++ % 256);
+                        }
+
+                        readLoop();
+                    }
+
+                } catch (error) {
+                    done.fail(error);
+                }
+
+            });
+        };
+
+        readLoop();
+    });
+});

tests/SpeechRecognizerTests.ts

@@ -1076,7 +1076,7 @@ describe.each([true, false])("Service based tests", (forceNodeWebSocket: boolean
             });
     });
 
-    test("emptyFile", (done: jest.DoneCallback) => {
+    test.skip("emptyFile", (done: jest.DoneCallback) => {
         // tslint:disable-next-line:no-console
         console.info("Name: emptyFile");
         // Server Responses:
@@ -1771,56 +1771,6 @@ test("Push Stream Async", (done: jest.DoneCallback) => {
         });
 }, 10000);
 
-test("Bad DataType for PushStreams results in error", (done: jest.DoneCallback) => {
-    // tslint:disable-next-line:no-console
-    console.info("Name: Bad DataType for PushStreams results in error");
-
-    const s: sdk.SpeechConfig = BuildSpeechConfig();
-    objsToClose.push(s);
-
-    const p: sdk.PushAudioInputStream = sdk.AudioInputStream.createPushStream();
-    const config: sdk.AudioConfig = sdk.AudioConfig.fromStreamInput(p);
-
-    // Wrong data type for ReadStreams
-    fs.createReadStream(Settings.WaveFile).on("data", (buffer: ArrayBuffer) => {
-        p.write(buffer);
-    }).on("end", () => {
-        p.close();
-    });
-
-    const r: sdk.SpeechRecognizer = new sdk.SpeechRecognizer(s, config);
-    objsToClose.push(r);
-
-    expect(r).not.toBeUndefined();
-    expect(r instanceof sdk.Recognizer);
-
-    r.canceled = (r: sdk.Recognizer, e: sdk.SpeechRecognitionCanceledEventArgs) => {
-        try {
-            expect(e.errorDetails).not.toBeUndefined();
-            expect(sdk.CancellationReason[e.reason]).toEqual(sdk.CancellationReason[sdk.CancellationReason.Error]);
-            expect(sdk.CancellationErrorCode[e.errorCode]).toEqual(sdk.CancellationErrorCode[sdk.CancellationErrorCode.RuntimeError]);
-        } catch (error) {
-            done.fail(error);
-        }
-    };
-
-    r.recognizeOnceAsync(
-        (p2: sdk.SpeechRecognitionResult) => {
-            const res: sdk.SpeechRecognitionResult = p2;
-            try {
-                expect(res).not.toBeUndefined();
-                expect(res.errorDetails).not.toBeUndefined();
-                expect(sdk.ResultReason[res.reason]).toEqual(sdk.ResultReason[sdk.ResultReason.Canceled]);
-                done();
-            } catch (error) {
-                done.fail(error);
-            }
-        },
-        (error: string) => {
-            done.fail(error);
-        });
-});
-
 test("Connect / Disconnect", (done: jest.DoneCallback) => {
     // tslint:disable-next-line:no-console
     console.info("Name: Connect / Disconnect");

tests/TranslationRecognizerTests.ts

@@ -1400,59 +1400,5 @@ describe.each([true, false])("Service based tests", (forceNodeWebSocket: boolean
             (err: string) => {
                 done.fail(err);
             });
-    }, 70000);
-
-    test("Bad DataType for PushStreams results in error", (done: jest.DoneCallback) => {
-        // tslint:disable-next-line:no-console
-        console.info("Name: Bad DataType for PushStreams results in error");
-
-        const s: sdk.SpeechTranslationConfig = BuildSpeechConfig();
-        objsToClose.push(s);
-
-        s.addTargetLanguage("en-US");
-        s.speechRecognitionLanguage = "en-US";
-
-        const p: sdk.PushAudioInputStream = sdk.AudioInputStream.createPushStream();
-        const config: sdk.AudioConfig = sdk.AudioConfig.fromStreamInput(p);
-
-        // Wrong data type for ReadStreams
-        fs.createReadStream(Settings.WaveFile).on("data", (buffer: ArrayBuffer) => {
-            p.write(buffer);
-        }).on("end", () => {
-            p.close();
-        });
-
-        const r: sdk.TranslationRecognizer = new sdk.TranslationRecognizer(s, config);
-        objsToClose.push(r);
-
-        expect(r).not.toBeUndefined();
-        expect(r instanceof sdk.Recognizer);
-
-        r.canceled = (r: sdk.Recognizer, e: sdk.TranslationRecognitionCanceledEventArgs) => {
-            try {
-                expect(e.errorDetails).not.toBeUndefined();
-                expect(e.errorDetails).toContain("ArrayBuffer");
-                expect(sdk.CancellationReason[e.reason]).toEqual(sdk.CancellationReason[sdk.CancellationReason.Error]);
-                expect(sdk.CancellationErrorCode[e.errorCode]).toEqual(sdk.CancellationErrorCode[sdk.CancellationErrorCode.RuntimeError]);
-            } catch (error) {
-                done.fail(error);
-            }
-        };
-
-        r.recognizeOnceAsync(
-            (p2: sdk.TranslationRecognitionResult) => {
-                const res: sdk.TranslationRecognitionResult = p2;
-                try {
-                    expect(res).not.toBeUndefined();
-                    expect(res.errorDetails).not.toBeUndefined();
-                    expect(sdk.ResultReason[res.reason]).toEqual(sdk.ResultReason[sdk.ResultReason.Canceled]);
-                    done();
-                } catch (error) {
-                    done.fail(error);
-                }
-            },
-            (error: string) => {
-                done.fail(error);
-            });
-    });
+    }, 35000);
 });