import { HostEvent } from '../shared/models/host-event' ; import { Disposable } from 'app/shared/models/disposable' ; import { Observable } from 'rxjs/Observable' ; import { FunctionAppContext } from './function-app-context' ; import { Subject } from 'rxjs/Subject' ; import { UserService } from './services/user.service' ; import { Subscription } from 'rxjs/Subscription' ; import { ArmUtil } from './Utilities/arm-utils' ; export class HostEventClient implements Disposable { private events : Subject < HostEvent > ; private processedEvents : Observable < HostEvent > ; private tokenSubscription : Subscription ; private req : XMLHttpRequest ; private timeouts : number [ ] = [ ] ; private currentPosition = 0 ; private functionEventsMap : { [ key ] : HostEvent ; } = { } ; constructor ( private context , private userService ) { this . events = new Subject < HostEvent > ( ) ; this . processedEvents = this . events . do ( e => { this . functionEventsMap [ e . functionName ] = e ; } ) ; this . readHostEvents ( ) ; } get id ( ) : string { return this . context . site . id ; } dispose ( ) { if ( this . req ) { if ( this . timeouts ) { this . timeouts . forEach ( window . clearTimeout ) ; this . timeouts = [ ] ; } this . req . abort ( ) ; } if ( this . tokenSubscription ) { this . tokenSubscription . unsubscribe ( ) ; } delete this . functionEventsMap ; this . events . complete ( ) ; } getEvents ( functionName ) { if ( this . functionEventsMap [ functionName ] ) { return Observable . concat ( Observable . of ( this . functionEventsMap [ functionName ] ) , this . processedEvents . filter ( e => e . functionName === functionName ) ) ; } else { return Observable . concat ( Observable . of ( { id : '' , name : '' , functionName : functionName , diagnostics : [ ] , } ) , this . processedEvents . filter ( e => e . functionName === functionName ) ) ; } } public readHostEvents ( createEmpty = true , log ? ) { this . tokenSubscription = this . userService . getStartupInfo ( ) . subscribe ( info => { const defaultInterval = 500 ; const url = `${ this . context . scmUrl }/api/logstream/application/functions/structured` ; if ( this . req ) { if ( this . timeouts ) { this . timeouts . forEach ( window . clearTimeout ) ; this . timeouts = [ ] ; } this . req . abort ( ) ; } if ( ArmUtil . isLinuxApp ( this . context . site ) ) { return ; } this . req = new XMLHttpRequest ( ) ; this . req . open ( 'GET' , url , true ) ; this . req . setRequestHeader ( 'Authorization' , `Bearer ${ info . token }` ) ; this . req . setRequestHeader ( 'FunctionsPortal' , '1' ) ; this . req . send ( null ) ; const callBack = ( ) => { const diff = this . req . responseText . length - this . currentPosition ; if ( diff > 0 ) { if ( this . req . responseText . length ) { const newText = this . req . responseText . substring ( this . currentPosition ) ; const lines = newText . split ( '\n' ) ; lines . filter ( l => l . startsWith ( '[' ) || l . startsWith ( '{' ) ) . map ( l => ( l . startsWith ( '[' ) ? l . substring ( '[Verbose] ' . length ) : l ) ) . filter ( l => l . startsWith ( '{' ) ) . map ( l => { try { return JSON . parse ( l ) ; } catch ( e ) { console . error ( e ) ; return null ; } } ) . forEach ( e => { if ( e !== null ) { this . events . next ( e ) ; } } ) ; this . currentPosition += newText . lastIndexOf ( '\n' ) ; } } if ( this . req . readyState === XMLHttpRequest . DONE ) { this . readHostEvents ( ) ; } else { this . timeouts . push ( window . setTimeout ( callBack , defaultInterval ) ) ; } } ; callBack ( ) ; } ) ; } }