import { BehaviorSubject } from 'rxjs/BehaviorSubject' ; import { Buffer } from 'buffer' ; import { Observable } from 'rxjs/Observable' ; import { Subscriber } from 'rxjs/Subscriber' ; import * as BFSE from 'botframework-streaming' ; import fetch from 'cross-fetch' ; import { Activity , ConnectionStatus , Conversation , IBotConnection , Media , Message } from './directLine' ; const DIRECT_LINE_VERSION = 'DirectLine/3.0' ; const MAX_RETRY_COUNT = 3 ; const refreshTokenLifetime = 30 * 60 * 1000 ; const timeout = 20 * 1000 ; const refreshTokenInterval = refreshTokenLifetime / 2 ; interface DirectLineStreamingOptions { token : string ; conversationId ? : string ; domain : string ; botAgent ? : string ; } class StreamHandler implements BFSE . RequestHandler { private connectionStatus$ ; private subscriber : Subscriber < Activity > ; private shouldQueue : ( ) => boolean ; private activityQueue : Array < Activity > = [ ] ; constructor ( s , c$ , sq ) { this . subscriber = s ; this . connectionStatus$ = c$ ; this . shouldQueue = sq ; } public setSubscriber ( s ) { this . subscriber = s ; } async processRequest ( request , logger ? ) { const streams = [ ... request . streams ] ; const stream0 = streams . shift ( ) ; const activitySetJson = await stream0 . readAsString ( ) ; const activitySet = JSON . parse ( activitySetJson ) ; if ( activitySet . activities . length !== 1 ) { this . subscriber . error ( new Error ( 'there should be exactly one activity' ) ) ; return BFSE . StreamingResponse . create ( 500 ) ; } const activity = activitySet . activities [ 0 ] ; if ( streams . length > 0 ) { const attachments = [ ... activity . attachments ] ; let stream ; while ( stream = streams . shift ( ) ) { const attachment = await stream . readAsString ( ) ; const dataUri = "data:text/plain;base64," + attachment ; attachments . push ( { contentType : stream . contentType , contentUrl : dataUri } ) ; } activity . attachments = attachments ; } if ( this . shouldQueue ( ) ) { this . activityQueue . push ( activity ) ; } else { this . subscriber . next ( activity ) ; } return BFSE . StreamingResponse . create ( 200 ) ; } public flush ( ) { this . connectionStatus$ . subscribe ( cs => { } ) ; this . activityQueue . forEach ( ( a ) => this . subscriber . next ( a ) ) ; this . activityQueue = [ ] ; } } export class DirectLineStreaming implements IBotConnection { public connectionStatus$ = new BehaviorSubject ( ConnectionStatus . Uninitialized ) ; public activity$ : Observable < Activity > ; private activitySubscriber : Subscriber < Activity > ; private theStreamHandler : StreamHandler ; private domain : string ; private conversationId : string ; private token : string ; private streamConnection : BFSE . WebSocketClient ; private queueActivities : boolean ; private _botAgent = '' ; constructor ( options ) { this . token = options . token ; this . refreshToken ( ) ; this . domain = options . domain ; if ( options . conversationId ) { this . conversationId = options . conversationId ; } this . _botAgent = this . getBotAgent ( options . botAgent ) ; this . queueActivities = true ; this . activity$ = Observable . create ( async ( subscriber ) => { this . activitySubscriber = subscriber ; this . theStreamHandler = new StreamHandler ( subscriber , this . connectionStatus$ , ( ) => this . queueActivities ) ; this . connectWithRetryAsync ( ) ; } ) . share ( ) ; } public reconnect ( { conversationId , token } ) { this . conversationId = conversationId ; this . token = token ; this . connectAsync ( ) ; } end ( ) { this . connectionStatus$ . next ( ConnectionStatus . Ended ) ; this . streamConnection . disconnect ( ) ; } private commonHeaders ( ) { return { "Authorization" : `Bearer ${ this . token }` , "x-ms-bot-agent" : this . _botAgent } ; } private getBotAgent ( customAgent = '' ) { let clientAgent = 'directlineStreaming' ; if ( customAgent ) { clientAgent += `; ${ customAgent }` ; } return `${ DIRECT_LINE_VERSION } (${ clientAgent })` ; } private async refreshToken ( firstCall = true , retryCount = 0 ) { await this . waitUntilOnline ( ) ; let numberOfAttempts = 0 ; while ( numberOfAttempts < MAX_RETRY_COUNT ) { numberOfAttempts ++ ; await new Promise ( r => setTimeout ( r , refreshTokenInterval ) ) ; try { const res = await fetch ( `${ this . domain }/tokens/refresh` , { method : "POST" , headers : this . commonHeaders ( ) } ) ; if ( res . ok ) { numberOfAttempts = 0 ; const { token } = await res . json ( ) ; this . token = token ; } else { if ( res . status === 403 || res . status === 403 ) { console . error ( `Fatal error while refreshing the token: ${ res . status } ${ res . statusText }` ) ; this . streamConnection . disconnect ( ) ; } else { console . warn ( `Refresh attempt #${ numberOfAttempts } failed: ${ res . status } ${ res . statusText }` ) ; } } } catch ( e ) { console . warn ( `Refresh attempt #${ numberOfAttempts } threw an exception: ${ e }` ) ; } } console . error ( "Retries exhausted" ) ; this . streamConnection . disconnect ( ) ; } postActivity ( activity ) { if ( activity . type === "message" && activity . attachments && activity . attachments . length > 0 ) { return this . postMessageWithAttachments ( activity ) ; } const resp$ = Observable . create ( async ( subscriber ) => { const request = BFSE . StreamingRequest . create ( 'POST' , '/v3/directline/conversations/' + this . conversationId + '/activities' ) ; request . setBody ( JSON . stringify ( activity ) ) ; const resp = await this . streamConnection . send ( request ) ; try { if ( resp . statusCode !== 200 ) throw new Error ( "PostActivity returned " + resp . statusCode ) ; const numberOfStreams = resp . streams . length ; if ( numberOfStreams !== 1 ) throw new Error ( "Expected one stream but got " + numberOfStreams ) ; const idString = await resp . streams [ 0 ] . readAsString ( ) ; const { Id : id } = JSON . parse ( idString ) ; return subscriber . next ( id ) ; } catch ( e ) { console . warn ( e ) ; this . streamConnection . disconnect ( ) ; return subscriber . error ( e ) ; } } ) ; return resp$ ; } private postMessageWithAttachments ( message ) { const { attachments , ... messageWithoutAttachments } = message ; return Observable . create ( subscriber => { const httpContentList = [ ] ; ( async ( ) => { try { const arrayBuffers = await Promise . all ( attachments . map ( async ( attachment ) => { const media = attachment as Media ; const res = await fetch ( media . contentUrl ) ; if ( res . ok ) { return { arrayBuffer : await res . arrayBuffer ( ) , media } ; } else { throw new Error ( '...' ) ; } } ) ) ; arrayBuffers . forEach ( ( { arrayBuffer , media } ) => { const buffer = Buffer . from ( arrayBuffer ) ; console . log ( buffer ) ; const stream = new BFSE . SubscribableStream ( ) ; stream . write ( buffer ) ; const httpContent = new BFSE . HttpContent ( { type : media . contentType , contentLength : buffer . length } , stream ) ; httpContentList . push ( httpContent ) ; } ) ; const url = `/v3/directline/conversations/${ this . conversationId }/users/${ messageWithoutAttachments . from . id }/upload` ; const request = BFSE . StreamingRequest . create ( 'PUT' , url ) ; const activityStream = new BFSE . SubscribableStream ( ) ; activityStream . write ( JSON . stringify ( messageWithoutAttachments ) , 'utf-8' ) ; request . addStream ( new BFSE . HttpContent ( { type : "application/vnd.microsoft.activity" , contentLength : activityStream . length } , activityStream ) ) ; httpContentList . forEach ( e => request . addStream ( e ) ) ; const resp = await this . streamConnection . send ( request ) ; if ( resp . streams && resp . streams . length !== 1 ) { subscriber . error ( new Error ( `Invalid stream count ${ resp . streams . length }` ) ) ; } else { const { Id : id } = await resp . streams [ 0 ] . readAsJson ( ) ; subscriber . next ( id ) ; } } catch ( e ) { subscriber . error ( e ) ; } } ) ( ) ; } ) ; } private async waitUntilOnline ( ) { return new Promise < void > ( ( resolve , reject ) => { this . connectionStatus$ . subscribe ( ( cs ) => { if ( cs === ConnectionStatus . Online ) return resolve ( ) ; } , ( e ) => reject ( e ) ) ; } ) ; } private async connectAsync ( ) { const re = new RegExp ( '^http(s?)' ) ; if ( ! re . test ( this . domain ) ) throw ( "Domain must begin with http or https" ) ; const params = { token : this . token } ; if ( this . conversationId ) params [ 'conversationId' ] = this . conversationId ; const urlSearchParams = new URLSearchParams ( params ) . toString ( ) ; const wsUrl = `${ this . domain . replace ( re , 'ws$1' ) }/conversations/connect?${ urlSearchParams }` ; return new Promise ( async ( resolve , reject ) => { try { this . streamConnection = new BFSE . WebSocketClient ( { url : wsUrl , requestHandler : this . theStreamHandler , disconnectionHandler : ( e ) => resolve ( e ) } ) ; this . queueActivities = true ; await this . streamConnection . connect ( ) ; const request = BFSE . StreamingRequest . create ( 'POST' , '/v3/directline/conversations' ) ; const response = await this . streamConnection . send ( request ) ; if ( response . statusCode !== 200 ) throw new Error ( "Connection response code " + response . statusCode ) ; if ( response . streams . length !== 1 ) throw new Error ( "Expected 1 stream but got " + response . streams . length ) ; const responseString = await response . streams [ 0 ] . readAsString ( ) ; const conversation = JSON . parse ( responseString ) ; this . conversationId = conversation . conversationId ; this . connectionStatus$ . next ( ConnectionStatus . Online ) ; await this . waitUntilOnline ( ) ; this . theStreamHandler . flush ( ) ; this . queueActivities = false ; } catch ( e ) { reject ( e ) ; } } ) ; } private async connectWithRetryAsync ( ) { let numRetries = MAX_RETRY_COUNT ; while ( numRetries > 0 ) { numRetries -- ; const start = Date . now ( ) ; try { this . connectionStatus$ . next ( ConnectionStatus . Connecting ) ; const res = await this . connectAsync ( ) ; console . warn ( `Retrying connection ${ res }` ) ; if ( 60000 < Date . now ( ) - start ) { numRetries = MAX_RETRY_COUNT ; continue ; } } catch ( err ) { console . error ( `Failed to connect ${ err }` ) ; throw ( err ) ; } await new Promise ( r => setTimeout ( r , this . getRetryDelay ( ) ) ) ; } } private getRetryDelay ( ) { return Math . floor ( 3000 + Math . random ( ) * 12000 ) ; } }