Add per-conversation streaming switch (deliveryMode='stream') (#443)

* add deliveryMode support

* deliveryMode support to enable streaming from ABS side

* comment resolved

---------

Co-authored-by: Pranav Joshi <pranavjoshi@microsoft.com>

Author: pranavjoshi001

src/directLine.test.ts

@@ -243,4 +243,151 @@ describe('MockSuite', () => {
         expect(actualError.status).toStrictEqual(429);
         expect(endTime - startTime).toStrictEqual(10);
     });
+
+    describe('StreamingMode', () => {
+
+        test('Setting streaming adds deliveryMode=stream to outgoing activity', () => {
+            // override directline with streaming enabled
+            directline = new DirectLineExport.DirectLine({ ...services, streaming: true });
+
+            const streamingActivity = DirectLineMock.mockActivity('streaming-test');
+            const scenario = function* (): IterableIterator<Observable<unknown>> {
+                yield Observable.timer(200, scheduler);
+                yield directline.postActivity(streamingActivity);
+            };
+
+            subscriptions.push(lazyConcat(scenario()).observeOn(scheduler).subscribe());
+
+            const actual: Array<DirectLineExport.Activity> = [];
+            subscriptions.push(directline.activity$.subscribe(a => actual.push(a)));
+
+            scheduler.flush();
+
+            expect(streamingActivity.deliveryMode).toStrictEqual('stream');
+            expect(actual[0].deliveryMode).toStrictEqual('stream');
+        });
+
+        test('Not setting streaming does not add deliveryMode at all to outgoing activity', () => {
+            const normalActivity = DirectLineMock.mockActivity('normal-test');
+            const scenario = function* (): IterableIterator<Observable<unknown>> {
+                yield Observable.timer(200, scheduler);
+                yield directline.postActivity(normalActivity);
+            };
+
+            subscriptions.push(lazyConcat(scenario()).observeOn(scheduler).subscribe());
+
+            const actual: Array<DirectLineExport.Activity> = [];
+            subscriptions.push(directline.activity$.subscribe(a => actual.push(a)));
+
+            scheduler.flush();
+
+            expect(normalActivity.deliveryMode).toBeUndefined();
+            expect(actual[0].deliveryMode).toBeUndefined();
+        });
+
+        test('Setting streaming overrides passed deliveryMode "normal" in activity to "stream"', () => {
+            directline = new DirectLineExport.DirectLine({ ...services, streaming: true });
+
+            const presetActivity: DirectLineExport.Message = {
+                type: 'message',
+                from: { id: 'sender' },
+                text: 'preset',
+                deliveryMode: 'normal'
+            };
+
+            const scenario = function* (): IterableIterator<Observable<unknown>> {
+                yield Observable.timer(200, scheduler);
+                yield directline.postActivity(presetActivity);
+            };
+
+            subscriptions.push(lazyConcat(scenario()).observeOn(scheduler).subscribe());
+
+            const actual: Array<DirectLineExport.Activity> = [];
+            subscriptions.push(directline.activity$.subscribe(a => actual.push(a)));
+
+            scheduler.flush();
+
+            expect(presetActivity.deliveryMode).toStrictEqual('stream');
+            expect(actual[0].deliveryMode).toStrictEqual('stream');
+        });
+
+        test('Not setting streaming preserves passed deliveryMode "normal" in activity', () => {
+            const presetActivity: DirectLineExport.Message = {
+                type: 'message',
+                from: { id: 'sender' },
+                text: 'preset-nonstream',
+                deliveryMode: 'normal'
+            };
+
+            const scenario = function* (): IterableIterator<Observable<unknown>> {
+                yield Observable.timer(200, scheduler);
+                yield directline.postActivity(presetActivity);
+            };
+
+            subscriptions.push(lazyConcat(scenario()).observeOn(scheduler).subscribe());
+
+            const actual: Array<DirectLineExport.Activity> = [];
+            subscriptions.push(directline.activity$.subscribe(a => actual.push(a)));
+
+            scheduler.flush();
+
+            expect(presetActivity.deliveryMode).toStrictEqual('normal');
+            expect(actual[0].deliveryMode).toStrictEqual('normal');
+        });
+
+        test.each([
+            { streaming: true, expectedDeliveryMode: 'stream', testName: 'Streaming' },
+            { streaming: false, expectedDeliveryMode: undefined, testName: 'Non-streaming' }
+        ])('$testName + 403 post returns retry and preserves deliveryMode', ({ streaming, expectedDeliveryMode }) => {
+            services.ajax = DirectLineMock.mockAjax(server, (urlOrRequest) => {
+                if (typeof urlOrRequest === 'string') {
+                    throw new Error();
+                }
+
+                if (urlOrRequest.url && urlOrRequest.url.indexOf('/conversations') > 0 && !/activities/u.test(urlOrRequest.url)) {
+                    // start conversation
+                    const response: Partial<AjaxResponse> = {
+                        response: server.conversation,
+                        status: 201,
+                        xhr: { getResponseHeader: () => 'n/a' } as unknown as XMLHttpRequest
+                    };
+                    return response as AjaxResponse;
+                }
+
+                if (urlOrRequest.url && /activities/u.test(urlOrRequest.url)) {
+                    const response: Partial<AjaxResponse> = {
+                        status: 403,
+                        xhr: { getResponseHeader: () => 'n/a' } as unknown as XMLHttpRequest
+                    };
+                    const error = new Error('Forbidden');
+                    throw Object.assign(error, response);
+                }
+
+                throw new Error();
+            });
+
+            directline = new DirectLineExport.DirectLine({
+                ...services,
+                ...(streaming ? { streaming: true } : {})
+            });
+
+            const retryActivity = DirectLineMock.mockActivity('will-retry');
+            const scenario = function* (): IterableIterator<Observable<unknown>> {
+                yield Observable.timer(200, scheduler);
+                yield directline.postActivity(retryActivity);
+            };
+
+            let postResult: string | undefined;
+            subscriptions.push(lazyConcat(scenario()).observeOn(scheduler).subscribe({
+                next: v => { postResult = v as string; },
+                error: () => {},
+                complete: () => {}
+            }));
+
+            scheduler.flush();
+
+            expect(retryActivity.deliveryMode).toStrictEqual(expectedDeliveryMode);
+            expect(postResult).toStrictEqual('retry');
+        });
+    });
 });

src/directLine.ts

@@ -305,6 +305,8 @@ export interface User {
     role?: UserRole
 }
 
+export type DeliveryMode = "normal" | "stream";
+
 export interface IActivity {
     type: string,
     channelData?: any,
@@ -313,7 +315,8 @@ export interface IActivity {
     eTag?: string,
     from: User,
     id?: string,
-    timestamp?: string
+    timestamp?: string,
+    deliveryMode?: DeliveryMode
 }
 
 export type AttachmentLayout = "list" | "carousel";
@@ -372,7 +375,13 @@ export interface DirectLineOptions {
     timeout?: number,
     // Attached to all requests to identify requesting agent.
     botAgent?: string,
-    conversationStartProperties?: any
+    conversationStartProperties?: any,
+    /**
+     * Per-conversation switch for streaming delivery mode.
+     * If true, every outgoing activity will include deliveryMode: 'stream'.
+     * If false/omitted, deliveryMode is not sent (defaults to 'normal' in ABS).
+     */
+    streaming?: boolean
 }
 
 export interface Services {
@@ -477,12 +486,17 @@ export class DirectLine implements IBotConnection {
     private pollingInterval: number = 1000; //ms
 
     private tokenRefreshSubscription: Subscription;
+    private streaming: boolean;
 
     constructor(options: DirectLineOptions & Partial<Services>) {
         this.secret = options.secret;
         this.token = options.secret || options.token;
         this.webSocket = (options.webSocket === undefined ? true : options.webSocket) && typeof WebSocket !== 'undefined' && WebSocket !== undefined;
 
+        if (options.streaming) {
+            this.streaming = options.streaming;
+        }
+
         if (options.conversationStartProperties && options.conversationStartProperties.locale) {
             if (Object.prototype.toString.call(options.conversationStartProperties.locale) === '[object String]') {
                 this.localeOnStartConversation = options.conversationStartProperties.locale;
@@ -754,6 +768,12 @@ export class DirectLine implements IBotConnection {
     }
 
     postActivity(activity: Activity) {
+        // If streaming is enabled for this DirectLine instance, always set deliveryMode to 'stream'
+        // default would be 'normal' and not passing anything meaning 'normal' as well on ABS side
+        if (this.streaming) {
+            activity.deliveryMode = 'stream';
+        }
+
         // If user id is set, check if it match activity.from.id and always override it in activity
         if (this.userIdOnStartConversation && activity.from && activity.from.id !== this.userIdOnStartConversation) {
             console.warn('DirectLineJS: Activity.from.id does not match with user id, ignoring activity.from.id');