import { get , trimEnd , debounce } from 'lodash' ; import { BehaviorSubject , throwError , timer , defer , from , Observable , NEVER } from 'rxjs' ; import { catchError , finalize } from 'rxjs/operators' ; import { CoreStart , CoreSetup , ToastsSetup } from 'opensearch-dashboards/public' ; import { getCombinedSignal , AbortError , IOpenSearchDashboardsSearchRequest , IOpenSearchDashboardsSearchResponse , ISearchOptions , OPENSEARCH_SEARCH_STRATEGY , } from '../../common' ; import { SearchUsageCollector } from './collectors' ; import { SearchTimeoutError , PainlessError , isPainlessError , TimeoutErrorMode } from './errors' ; import { toMountPoint } from '../../../opensearch_dashboards_react/public' ; export interface SearchInterceptorDeps { http : CoreSetup [ 'http' ] ; uiSettings : CoreSetup [ 'uiSettings' ] ; startServices : Promise < [ CoreStart , any , unknown ] > ; toasts : ToastsSetup ; usageCollector ? : SearchUsageCollector ; } export class SearchInterceptor { internal protected abortController = new AbortController ( ) ; internal protected pendingCount$ = new BehaviorSubject ( 0 ) ; internal protected application ! : CoreStart [ 'application' ] ; constructor ( protected readonly deps ) { this . deps . http . addLoadingCountSource ( this . pendingCount$ ) ; this . deps . startServices . then ( ( [ coreStart ] ) => { this . application = coreStart . application ; } ) ; } protected getTimeoutMode ( ) { return TimeoutErrorMode . UPGRADE ; } protected handleSearchError ( e , request , timeoutSignal , appAbortSignal ? ) { if ( timeoutSignal . aborted || get ( e , 'body.message' ) === 'Request timed out' ) { const err = new SearchTimeoutError ( e , this . getTimeoutMode ( ) ) ; this . showTimeoutError ( err ) ; return err ; } else if ( appAbortSignal ?. aborted ) { return e ; } else if ( isPainlessError ( e ) ) { return new PainlessError ( e , request ) ; } else { return e ; } } internal protected runSearch ( request , signal , strategy ? ) { const { id , ... searchRequest } = request ; const path = trimEnd ( `/internal/search/${ strategy || OPENSEARCH_SEARCH_STRATEGY }/${ id || '' }` , '/' ) ; const body = JSON . stringify ( searchRequest ) ; return from ( this . deps . http . fetch ( { method : 'POST' , path , body , signal , } ) ) ; } internal protected setupAbortSignal ( { abortSignal , timeout , } ) { const timeoutController = new AbortController ( ) ; const { signal : timeoutSignal } = timeoutController ; const timeout$ = timeout ? timer ( timeout ) : NEVER ; const subscription = timeout$ . subscribe ( ( ) => { timeoutController . abort ( ) ; } ) ; const signals = [ this . abortController . signal , timeoutSignal , ... ( abortSignal ? [ abortSignal ] : [ ] ) , ] ; const combinedSignal = getCombinedSignal ( signals ) ; const cleanup = ( ) => { subscription . unsubscribe ( ) ; } ; combinedSignal . addEventListener ( 'abort' , cleanup ) ; return { combinedSignal , timeoutSignal , cleanup , } ; } internal private showTimeoutError = debounce ( ( e ) => { this . deps . toasts . addDanger ( { title : 'Timed out' , text : toMountPoint ( e . getErrorMessage ( this . application ) ) , } ) ; } , 30000 , { leading : true , trailing : false } ) ; param request options returns public search ( request , options ? ) { return defer ( ( ) => { if ( options ?. abortSignal ?. aborted ) { return throwError ( new AbortError ( ) ) ; } const { timeoutSignal , combinedSignal , cleanup } = this . setupAbortSignal ( { abortSignal : options ?. abortSignal , } ) ; this . pendingCount$ . next ( this . pendingCount$ . getValue ( ) + 1 ) ; return this . runSearch ( request , combinedSignal , options ?. strategy ) . pipe ( catchError ( ( e ) => { return throwError ( this . handleSearchError ( e , request , timeoutSignal , options ?. abortSignal ) ) ; } ) , finalize ( ( ) => { this . pendingCount$ . next ( this . pendingCount$ . getValue ( ) - 1 ) ; cleanup ( ) ; } ) ) ; } ) ; } public showError ( e ) { if ( e instanceof AbortError ) return ; if ( e instanceof SearchTimeoutError ) { return ; } if ( e instanceof PainlessError ) { this . deps . toasts . addDanger ( { title : 'Search Error' , text : toMountPoint ( e . getErrorMessage ( this . application ) ) , } ) ; return ; } this . deps . toasts . addError ( e , { title : 'Search Error' , } ) ; } } export type ISearchInterceptor = PublicMethodsOf < SearchInterceptor > ;