import { Readable } from 'stream' ; import { createConcatStream , createFilterStream , createMapStream , createPromiseFromStreams , } from '../../utils/streams' ; import { SavedObject } from '../types' ; import { createLimitStream } from './create_limit_stream' ; import { SavedObjectsImportError } from './types' ; import { getNonUniqueEntries } from './get_non_unique_entries' ; import { SavedObjectsErrorHelpers } from '..' ; interface CollectSavedObjectsOptions { readStream : Readable ; objectLimit : number ; filter ? : ( obj : SavedObject ) => boolean ; supportedTypes : string [ ] ; } export async function collectSavedObjects ( { readStream , objectLimit , filter , supportedTypes , } ) { const errors = [ ] ; const entries = [ ] ; const importIdMap = new Map < string , { id ? : string ; omitOriginId ? : boolean ; } > ( ) ; const collectedObjects = await createPromiseFromStreams ( [ readStream , createLimitStream ( objectLimit ) , createFilterStream < SavedObject < { title : string ; } > > ( ( obj ) => { entries . push ( { type : obj . type , id : obj . id } ) ; if ( supportedTypes . includes ( obj . type ) ) { return true ; } const { title } = obj . attributes ; errors . push ( { id : obj . id , type : obj . type , title , meta : { title } , error : { type : 'unsupported_type' , } , } ) ; return false ; } ) , createFilterStream < SavedObject > ( ( obj ) => ( filter ? filter ( obj ) : true ) ) , createMapStream ( ( obj ) => { importIdMap . set ( `${ obj . type }:${ obj . id }` , { } ) ; return Object . assign ( { migrationVersion : { } } , obj ) ; } ) , createConcatStream ( [ ] ) , ] ) ; const nonUniqueEntries = getNonUniqueEntries ( entries ) ; if ( nonUniqueEntries . length > 0 ) { throw SavedObjectsErrorHelpers . createBadRequestError ( `Non-unique import objects detected: [${ nonUniqueEntries . join ( ) }]` ) ; } return { errors , collectedObjects , importIdMap , } ; }