import { Observable } from 'rxjs' ; import { first } from 'rxjs/operators' ; import { ApiResponse } from '@elastic/elasticsearch' ; import { SearchResponse } from 'elasticsearch' ; import { IUiSettingsClient , IScopedClusterClient , SharedGlobalConfig } from 'src/core/server' ; import { MsearchRequestBody , MsearchResponse } from '../../../common/search/search_source' ; import { shimHitsTotal } from './shim_hits_total' ; import { getShardTimeout , getDefaultSearchParams , toSnakeCase , shimAbortSignal } from '..' ; internal export function convertRequestBody ( requestBody , { timeout } ) { return requestBody . searches . reduce ( ( req , curr ) => { const header = JSON . stringify ( { ignore_unavailable : true , ... curr . header , } ) ; const body = JSON . stringify ( { timeout , ... curr . body , } ) ; return `${ req }${ header }\n${ body }\n` ; } , '' ) ; } interface CallMsearchDependencies { opensearchClient : IScopedClusterClient ; globalConfig$ : Observable < SharedGlobalConfig > ; uiSettings : IUiSettingsClient ; } internal export function getCallMsearch ( dependencies ) { return async ( params ) : Promise < MsearchResponse > => { const { opensearchClient , globalConfig$ , uiSettings } = dependencies ; const config = await globalConfig$ . pipe ( first ( ) ) . toPromise ( ) ; const timeout = getShardTimeout ( config ) ; const { trackTotalHits , ... defaultParams } = await getDefaultSearchParams ( uiSettings ) ; const body = convertRequestBody ( params . body , timeout ) ; const promise = shimAbortSignal ( opensearchClient . asCurrentUser . msearch ( { body , } , { querystring : toSnakeCase ( defaultParams ) , } ) , params . signal ) ; const response = ( await promise ) as ApiResponse < { responses : Array < SearchResponse < any > > ; } > ; return { body : { ... response , body : { responses : response . body . responses ?. map ( ( r ) => shimHitsTotal ( r ) ) , } , } , } ; } ; }