import { Observable } from 'rxjs/Observable' ; import { of } from 'rxjs/observable/of' ; import { timer } from 'rxjs/observable/timer' ; import { _throw } from 'rxjs/observable/throw' ; import { SDKFetch , SDKFetchOptions } from './SDKFetch' ; import { appendQueryString , toQueryString } from './utils' ; interface PollingOptions { /**
     * 最大轮询数
     */ maxTimes ? : number ; /**
     * 轮询间隔
     */ interval ? : number ; /**
     * 最大轮询间隔
     */ maxInterval ? : number ; /**
     * 轮询间隔步长
     */ intervalSteps ? : number ; } export interface AsyncJobOptions < T > extends AsyncJobCallbacks < T > { timeoutSeconds ? : number ; readySid ? : string ; pollingOptions ? : PollingOptions ; } interface AsyncJobCallbacks < T > { onPending ? : ( ) => void ; onFulfilled ? : ( res : T ) => void ; onRejected ? : ( e : Error ) => void ; } type Options < T > = SDKFetchOptions & AsyncJobOptions < T > ; interface AsyncResult { readyKey : string ; result : null ; timeout : boolean ; } interface AsyncJobInfo < T > { isDone : boolean ; timeCost ? : number ; statusCode ? : number ; request : { headers : string ; url : string ; body ? : { } ; } ; response ? : { status : number ; headers : string ; body : T ; } ; } interface AsyncJobsRes < T > { result : Record < string , AsyncJobInfo < T > > ; } const DefaultPollInterval = 3000 ; const DefaultSteps = 1.2 ; const DefaultMaxPollInterval = 1000 * 10 ; const DefaultMaxPollTimes = 7 ; export class SDKAsyncJob { constructor ( private fetch ) { } get < T > ( path , query ? , options = { } ) { const { onPending , onFulfilled , onRejected , timeoutSeconds , readySid , pollingOptions , ... rest } = options ; const callbacks = { onPending , onFulfilled , onRejected } ; const q = this . normalizeQuery ( query , timeoutSeconds , readySid ) ; return this . fetch . get < T > ( path , q , rest ) . switchMap ( ( res ) => this . handleRes ( res , callbacks , pollingOptions ) ) ; } post < T > ( path , body ? , options = { } ) { const { onPending , onFulfilled , onRejected , timeoutSeconds , readySid , pollingOptions , ... rest } = options ; const callbacks = { onPending , onFulfilled , onRejected } ; const query = this . normalizeQuery ( void 0 , timeoutSeconds , readySid ) ; const url = this . appendQueryToUrl ( path , query ) ; return this . fetch . post < T > ( url , body , rest ) . switchMap ( ( res ) => this . handleRes ( res , callbacks , pollingOptions ) ) ; } put < T > ( path , body ? , options = { } ) { const { onPending , onFulfilled , onRejected , timeoutSeconds , readySid , pollingOptions , ... rest } = options ; const callbacks = { onPending , onFulfilled , onRejected } ; const query = this . normalizeQuery ( void 0 , timeoutSeconds , readySid ) ; const url = this . appendQueryToUrl ( path , query ) ; return this . fetch . put < T > ( url , body , rest ) . switchMap ( ( res ) => this . handleRes ( res , callbacks , pollingOptions ) ) ; } delete < T > ( path , body ? , options = { } ) { const { onPending , onFulfilled , onRejected , timeoutSeconds , readySid , pollingOptions , ... rest } = options ; const callbacks = { onPending , onFulfilled , onRejected } ; const query = this . normalizeQuery ( void 0 , timeoutSeconds , readySid ) ; const url = this . appendQueryToUrl ( path , query ) ; return this . fetch . delete < T > ( url , body , rest ) . switchMap ( ( res ) => this . handleRes ( res , callbacks , pollingOptions ) ) ; } private appendQueryToUrl ( url , q ) { return appendQueryString ( url , toQueryString ( q ) ) ; } private handleRes < T > ( res , callbacks , pollingOptions = { } ) { if ( ! this . isAsyncResult ( res ) ) { return of ( res ) ; } if ( callbacks . onPending ) { callbacks . onPending ( ) ; } const { maxTimes = DefaultMaxPollTimes , interval = DefaultPollInterval , maxInterval = DefaultMaxPollInterval , intervalSteps = DefaultSteps , } = pollingOptions ; return this . waitingForJobDone < T > ( res . readyKey , callbacks , { maxTimes , interval , maxInterval , intervalSteps , } ) ; } private waitingForJobDone < T > ( key , callbacks , pollingOptions ) { const polling$ = this . getAsyncJobResult < T > ( key ) . switchMap ( res => { if ( res === null ) { return this . polling < T > ( key , 0 , callbacks , pollingOptions ) ; } if ( callbacks . onFulfilled ) { callbacks . onFulfilled ( res ) ; } return of ( res ) ; } ) ; return polling$ ; } private polling < T > ( key , times , callbacks , pollingOptions ) { const { intervalSteps , maxInterval , maxTimes , interval , } = pollingOptions ; return timer ( interval ) . switchMap ( ( ) => this . getAsyncJobResult < T > ( key ) ) . switchMap ( res => { if ( res === null ) { if ( times > maxTimes ) { const e = new Error ( 'Async job polling failed' ) ; if ( callbacks . onRejected ) { callbacks . onRejected ( e ) ; } return _throw ( e ) ; } const nextTimes = times + 1 ; const nextOptions = { intervalSteps , maxInterval , maxTimes , interval : Math . floor ( Math . min ( interval * intervalSteps , maxInterval ) ) , } ; return this . polling < T > ( key , nextTimes , callbacks , nextOptions ) ; } if ( callbacks . onFulfilled ) { callbacks . onFulfilled ( res ) ; } return of ( res ) ; } ) ; } private getAsyncJobResult < T > ( key ) { return this . fetch . get < AsyncJobsRes < T > > ( 'async-jobs' , { keys : [ key ] } ) . map ( m => { const r = m . result [ key ] ; if ( r . response && r . response . status && r . response . status >= 400 ) { throw new Error ( `${ r . response . body || r . response . status }` ) ; } if ( r . isDone ) { if ( ! r . response ) { throw new Error ( 'response is undefined' ) ; } return r . response . body ; } return null ; } ) ; } /**
     * 判断资源是否需要异步等待
     */ private isAsyncResult ( res ) { if ( typeof res !== 'object' ) { return false ; } if ( 'timeout' in res && 'readyKey' in res ) { return true ; } return false ; } private normalizeQuery ( query ? , timeoutSeconds = 3 , readySid = '' ) { return { timeoutAsync : true , timeoutSeconds , readySid , ... query , } ; } }