Fixed streaming bug in stream.ts (#285)

* fixed streaming bug in stream.ts

* wrapped whole function

Author: teddyam

package-lock.json

@@ -31,6 +31,7 @@ export class HttpStream implements IStreamer {
   private _result?: SentActivity;
   private _timeout?: NodeJS.Timeout;
   private _logger: ILogger;
+  private _flushing: boolean = false;
 
   constructor(client: Client, ref: ConversationReference, logger?: ILogger) {
     this.client = client;
@@ -95,60 +96,68 @@ export class HttpStream implements IStreamer {
   }
 
   protected async flush() {
-    if (!this.queue.length) return;
-    if (this._timeout) {
-      clearTimeout(this._timeout);
-      this._timeout = undefined;
-    }
+    // if locked or no queue, return early
+    if (!this.queue.length || this._flushing) return;
 
-    let i = 0;
+    this._flushing = true;
 
-    while (this.queue.length && i < 10) {
-      const activity = this.queue.shift();
+    try {
+      if (this._timeout) {
+        clearTimeout(this._timeout);
+        this._timeout = undefined;
+      }
 
-      if (!activity) continue;
+      let i = 0;
+      
+      while (this.queue.length && i < 10) {
+        const activity = this.queue.shift();
 
-      if (activity.text) {
-        this.text += activity.text;
-      }
+        if (!activity) continue;
 
-      if (activity.attachments) {
-        this.attachments = [...(this.attachments || []), ...activity.attachments];
-      }
+        if (activity.text) {
+          this.text += activity.text;
+        }
 
-      if (activity.channelData) {
-        this.channelData = {
-          ...this.channelData,
-          ...activity.channelData,
-        };
-      }
+        if (activity.attachments) {
+          this.attachments = [...(this.attachments || []), ...activity.attachments];
+        }
 
-      if (activity.entities) {
-        this.entities = [...(this.entities || []), ...activity.entities];
-      }
+        if (activity.channelData) {
+          this.channelData = {
+            ...this.channelData,
+            ...activity.channelData,
+          };
+        }
 
-      i++;
-    }
+        if (activity.entities) {
+          this.entities = [...(this.entities || []), ...activity.entities];
+        }
 
-    if (i === 0) return;
+        i++;
+      }
 
-    const activity = new TypingActivity({ id: this.id })
-      .withText(this.text)
-      .addStreamUpdate(this.index + 1);
+      if (i === 0) return;
 
-    const res = await promises.retry(() => this.send(activity), {
-      logger: this._logger
-    });
+      const activity = new TypingActivity({ id: this.id })
+        .withText(this.text)
+        .addStreamUpdate(this.index + 1);
 
-    this.events.emit('chunk', res);
-    this.index++;
+      const res = await promises.retry(() => this.send(activity), {
+        logger: this._logger
+      });
 
-    if (!this.id) {
-      this.id = res.id;
-    }
+      this.events.emit('chunk', res);
+      this.index++;
+
+      if (!this.id) {
+        this.id = res.id;
+      }
 
-    if (this.queue.length) {
-      this._timeout = setTimeout(this.flush.bind(this), 500);
+      if (this.queue.length) {
+        this._timeout = setTimeout(this.flush.bind(this), 500);
+      }
+    } finally {
+      this._flushing = false;
     }
   }