import * as mongodb from "mongodb" ; import * as async from "async" ; import { onlyOnce } from "./core/callback" ; import { EntityMapping } from "./mapping/entityMapping" ; import { ResultCallback } from "./core/callback" ; import { InternalSession } from "./session" ; import { ChangeTrackingType } from "./mapping/mappingModel" ; import { IdentityGenerator } from "./config/configuration" ; import { Batch } from "./batch" ; import { Callback } from "./core/callback" ; import { Command } from "./core/command" ; import { QueryDefinition } from "./query/queryDefinition" ; import { QueryKind } from "./query/queryKind" ; import { IteratorCallback } from "./core/callback" ; import { QueryDocument } from "./query/queryBuilder" ; import { CriteriaBuilder } from "./query/criteriaBuilder" ; import { UpdateDocumentBuilder } from "./query/updateDocumentBuilder" ; import { ReadContext } from "./mapping/readContext" ; import { Observer } from "./observer" ; import { OrderDocument } from "./query/orderDocument" ; import { WriteContext } from "./mapping/writeContext" ; import { PersistenceError , EntityNotFoundError } from "./persistenceError" ; import { getDuration } from "./core/timerUtil" ; import { Property } from "./mapping/property" ; import { CollationOptions } from "./mapping/collationOptions" ; interface FindOneQuery { criteria : any ; fields ? : any ; fetchPaths ? : string [ ] ; isLazy ? : boolean ; } interface FindAllQuery extends FindOneQuery { orderDocument ? : OrderDocument [ ] ; limitCount ? : number ; collationOptions ? : CollationOptions ; skipCount ? : number ; batchSizeValue ? : number ; } interface FindEachQuery extends FindAllQuery { iterator : IteratorCallback < any > ; } hidden export interface Persister { changeTracking : ChangeTrackingType ; identity : IdentityGenerator ; areDocumentsEqual ( context , entity , originalDocument ) : boolean ; dirtyCheck ( batch , entity , originalDocument ) : Object ; addInsert ( batch , entity ) : Object ; addRemove ( batch , entity ) : void ; fetch ( entity , path , callback ) : void ; fetchPropertyValue ( entity , property , callback ) : void ; refresh ( entity , callback ) : void ; watch ( value , observer ) : void ; executeQuery ( query , callback ) : void ; findOneById ( id , callback ) : void ; findInverseOf ( entity , path , callback ) : void ; findOneInverseOf ( entity , path , callback ) : void ; } export interface DirtyCheckContext { error ? : Error ; } hidden export class PersisterImpl implements Persister { changeTracking : ChangeTrackingType ; identity : IdentityGenerator ; private _versioned : boolean ; private _findQueue : FindQueue ; private _mapping : EntityMapping ; private _collection : mongodb . Collection ; private _session : InternalSession ; private _criteriaBuilder : CriteriaBuilder ; private _updateDocumentBuilder : UpdateDocumentBuilder ; private _traceEnabled : boolean ; private _defaultFields : QueryDocument ; constructor ( session , mapping , collection ) { this . _session = session ; this . _mapping = mapping ; this . _collection = collection ; var inheritanceRoot = ( < EntityMapping > mapping . inheritanceRoot ) ; this . changeTracking = inheritanceRoot . changeTracking ; this . identity = inheritanceRoot . identity ; this . _versioned = inheritanceRoot . versioned ; this . _traceEnabled = this . _session . factory . logger != null ; this . _defaultFields = mapping . getDefaultFields ( ) ; } dirtyCheck ( batch , entity , originalDocument ) { var document = this . _dirtyCheck ( batch , entity , originalDocument ) ; if ( ! document ) { return null ; } else { if ( this . _versioned ) { var version = this . _mapping . getDocumentVersion ( originalDocument ) ; this . _mapping . setDocumentVersion ( document , ( version || 0 ) + 1 ) ; } this . _getCommand ( batch ) . addReplace ( document , version ) ; return document ; } } areDocumentsEqual ( context , entity , originalDocument ) { return this . _dirtyCheck ( context , entity , originalDocument ) == null ; } private _dirtyCheck ( context , entity , originalDocument ) { var writeContext = new WriteContext ( ) ; var document = this . _mapping . write ( writeContext , entity ) ; if ( writeContext . hasErrors ) { context . error = new PersistenceError ( `Error serializing document:\n${ writeContext . getErrorMessage ( ) }` ) ; return null ; } if ( this . _mapping . areDocumentsEqual ( originalDocument , document ) ) { return null ; } else { return document ; } } addInsert ( batch , entity ) { var context = new WriteContext ( ) ; var document = this . _mapping . write ( context , entity ) ; if ( context . hasErrors ) { batch . error = new PersistenceError ( `Error serializing document:\n${ context . getErrorMessage ( ) }` ) ; return null ; } if ( this . _versioned ) { this . _mapping . setDocumentVersion ( document , 1 ) ; } this . _getCommand ( batch ) . addInsert ( document ) ; return document ; } addRemove ( batch , entity ) { this . _getCommand ( batch ) . addRemove ( entity [ "_id" ] ) ; } param entity param callback refresh ( entity , callback ) { this . findOneById ( entity [ "_id" ] , ( err , document ) => { if ( err ) return callback ( err ) ; this . _refreshFromDocument ( entity , document , callback ) ; } ) ; } watch ( value , observer ) { this . _mapping . watchEntity ( value , observer ) ; } private _refreshFromDocument ( entity , document , callback ) { var context = new ReadContext ( this . _session ) ; this . _mapping . refresh ( context , entity , document ) ; if ( context . hasErrors ) { return callback ( new PersistenceError ( "Error deserializing document:\n" + context . getErrorMessage ( ) ) ) ; } callback ( null , document ) ; } fetch ( entity , path , callback ) { if ( typeof path !== "string" ) { return callback ( new PersistenceError ( "Path must be a string." ) ) ; } this . _mapping . fetch ( this . _session , undefined , entity , path . split ( "." ) , 0 , callback ) ; } param entity param property param callback fetchPropertyValue ( entity , property , callback ) { var fields = { _id : 0 } ; property . setFieldValue ( fields , 1 ) ; var version ; if ( this . _versioned ) { version = this . _session . getVersion ( entity ) ; this . _mapping . setDocumentVersion ( fields , 1 ) ; } var criteria = { _id : entity [ "_id" ] } ; var handleCallback = callback ; if ( this . _traceEnabled ) { handleCallback = this . _createTraceableCallback ( { kind : QueryKind [ QueryKind . FindOne ] , criteria , fields } , callback ) ; } this . _collection . findOne ( criteria , fields , ( err , document ) => { if ( err ) return callback ( err ) ; if ( this . _versioned ) { if ( this . _mapping . getDocumentVersion ( document ) != version ) { return callback ( new Error ( "Cannot fetch property value because document version has changed." ) ) ; } } var readContext = new ReadContext ( this . _session ) ; readContext . path = property . name ; var propertyValue = property . mapping . read ( readContext , property . getFieldValue ( document ) ) ; if ( readContext . hasErrors ) { return callback ( new PersistenceError ( "Error deserializing property value: " + readContext . getErrorMessage ( ) ) ) ; } handleCallback ( null , propertyValue ) ; } ) ; } findInverseOf ( entity , path , callback ) { var criteria = this . _prepareInverseQuery ( entity , path , callback ) ; if ( criteria ) { this . findAll ( criteria , callback ) ; } } findOneInverseOf ( entity , path , callback ) { var criteria = this . _prepareInverseQuery ( entity , path , callback ) ; if ( criteria ) { this . findOne ( criteria , callback ) ; } } param entity param path param callback _prepareInverseQuery ( entity , path , callback ) { var property = this . _mapping . getProperty ( path ) ; if ( property === undefined ) { callback ( new PersistenceError ( "Missing property '" + path + "'." ) ) ; return null ; } var id = ( < any > entity ) . _id ; if ( id === undefined ) { callback ( new PersistenceError ( "Missing identifier on entity that is the inverse side of a relationship." ) ) ; return null ; } var query = { } ; property . setFieldValue ( query , id ) ; return query ; } findOneById ( id , callback ) { var entity = this . _session . getObject ( id ) ; if ( entity !== undefined ) { process . nextTick ( ( ) => callback ( null , entity ) ) ; return ; } ( this . _findQueue || ( this . _findQueue = new FindQueue ( this ) ) ) . add ( id , callback ) ; } findOne ( criteria , callback ) { var query = { criteria , fields : this . _defaultFields } ; var handleCallback = callback ; if ( this . _traceEnabled ) { query . kind = QueryKind [ QueryKind . FindOne ] ; handleCallback = this . _createTraceableCallback ( query , callback ) ; } this . _findOne ( query , handleCallback ) ; } private _findOne ( query , callback ) { this . _collection . findOne ( query . criteria , { projection : query . fields } , ( err , document ) => { if ( err ) return callback ( err ) ; this . _loadOne ( document , query . isLazy , callback ) ; } ) ; } findAll ( criteria , callback ) { var query = { criteria , fields : this . _defaultFields } ; var handleCallback = callback ; if ( this . _traceEnabled ) { query . kind = QueryKind [ QueryKind . FindAll ] ; handleCallback = this . _createTraceableCallback ( query , callback ) ; } this . _findAll ( query , handleCallback ) ; } private _findAll ( query , callback ) { var cursor = this . _prepareFind ( query ) , self = this , entities = [ ] ; next ( ) ; function next ( err ? ) { if ( err ) return error ( err ) ; cursor . next ( ( err , item ) => { if ( err ) return error ( err ) ; if ( item == null ) { return callback ( null , entities ) ; } self . _loadOne ( item , query . isLazy , ( err , value ) => { if ( err ) return error ( err ) ; if ( value !== null ) { entities . push ( value ) ; } next ( ) ; } ) ; } ) ; } function error ( err ) { cursor . close ( null ) ; callback ( err ) ; callback = < Callback > function ( ) { } ; } } executeQuery ( query , callback ) { if ( query . criteria ) { query . criteria = ( this . _criteriaBuilder || ( this . _criteriaBuilder = new CriteriaBuilder ( this . _mapping ) ) ) . build ( query . criteria ) ; if ( this . _criteriaBuilder . error ) { return callback ( this . _criteriaBuilder . error ) ; } } if ( query . isLazy ) { query . fields = { _id : 1 } ; } else { query . fields = this . _defaultFields ; } if ( query . updateDocument ) { query . updateDocument = ( this . _updateDocumentBuilder || ( this . _updateDocumentBuilder = new UpdateDocumentBuilder ( this . _mapping ) ) ) . build ( query . updateDocument ) ; if ( this . _updateDocumentBuilder . error ) { return callback ( this . _updateDocumentBuilder . error ) ; } if ( this . _versioned ) { var fields = query . updateDocument [ "$inc" ] ; if ( ! fields ) { fields = query . updateDocument [ "$inc" ] = { } ; } this . _mapping . setDocumentVersion ( fields , 1 ) ; } } if ( query . sortValue ) { this . _prepareOrderDocument ( query . sortValue , ( err , preparedOrder ) => { if ( err ) return callback ( err ) ; query . orderDocument = preparedOrder ; this . _executeQuery ( query , callback ) ; } ) ; } else { this . _executeQuery ( query , callback ) ; } } private _executeQuery ( query , callback ) { var handleCallback = callback ; if ( this . _traceEnabled && query . kind != QueryKind . FindOneById ) { handleCallback = this . _createTraceableCallback ( query . toObject ( ) , callback ) ; } switch ( query . kind ) { case QueryKind . FindOne : this . _findOne ( query , this . _fetchOne ( query , handleCallback ) ) ; break ; case QueryKind . FindOneById : this . findOneById ( query . id , this . _fetchOne ( query , handleCallback ) ) ; break ; case QueryKind . FindAll : this . _findAll ( query , this . _fetchAll ( query , handleCallback ) ) ; break ; case QueryKind . FindEach : this . _findEach ( query , handleCallback ) ; break ; case QueryKind . FindEachSeries : this . _findEachSeries ( query , handleCallback ) ; break ; case QueryKind . FindCursor : this . _findCursor ( query , handleCallback ) ; break ; case QueryKind . FindOneAndRemove : case QueryKind . FindOneAndUpdate : case QueryKind . FindOneAndUpsert : this . _findOneAndModify ( query , this . _fetchOne ( query , handleCallback ) ) ; break ; case QueryKind . RemoveOne : this . _removeOne ( query , handleCallback ) ; break ; case QueryKind . RemoveAll : this . _removeAll ( query , handleCallback ) ; break ; case QueryKind . UpdateOne : case QueryKind . Upsert : this . _updateOne ( query , handleCallback ) ; break ; case QueryKind . UpdateAll : this . _updateAll ( query , handleCallback ) ; break ; case QueryKind . Distinct : this . _distinct ( query , handleCallback ) ; break ; case QueryKind . Count : this . _count ( query , handleCallback ) ; break ; default : handleCallback ( new PersistenceError ( `Unknown query type '${ query . kind }'.` ) ) ; } } private _createTraceableCallback ( query , callback ) { var start = process . hrtime ( ) ; return ( err ? , result ? ) => { if ( err ) return callback ( err ) ; this . _session . factory . logger . trace ( { collection : ( < EntityMapping > this . _mapping . inheritanceRoot ) . collectionName , query , duration : getDuration ( start ) } , "[Hydrate] Executed query." ) ; callback ( null , result ) ; } ; } private _prepareOrderDocument ( sorting , callback ) { var order = [ ] ; for ( var i = 0 ; i < sorting . length ; i ++ ) { var value = sorting [ i ] ; var context = this . _mapping . resolve ( value [ 0 ] ) ; if ( context . error ) { return callback ( context . error ) ; } order . push ( [ context . resolvedPath , value [ 1 ] ] ) ; } callback ( null , order ) ; } private _findEach ( query , callback ) { var cursor = this . _prepareFind ( query ) ; var iterator = this . _fetchIterator ( query ) ; var completed = 0 , started = 0 , finished = false , self = this ; replenish ( ) ; function replenish ( ) { cursor . next ( ( err , item ) => { if ( err ) return error ( err ) ; if ( item == null ) { if ( completed >= started ) { callback ( ) ; } finished = true ; return ; } process ( err , item ) ; while ( ( < any > cursor ) . bufferedCount ( ) > 0 ) { cursor . next ( process ) ; } } ) ; } function process ( err , item ) { if ( err ) return error ( err ) ; started ++ ; self . _loadOne ( item , query . isLazy , ( err , value ) => { if ( err ) return error ( err ) ; if ( value === null ) { done ( null ) ; } else { iterator ( value , onlyOnce ( done ) ) ; } } ) ; } function done ( err ) { if ( err ) return error ( err ) ; completed ++ ; if ( ( < any > cursor ) . bufferedCount ( ) == 0 && completed >= started ) { if ( finished ) return callback ( ) ; replenish ( ) ; } } function error ( err ) { cursor . close ( null ) ; callback ( err ) ; callback = < Callback > function ( ) { } ; } } private _findEachSeries ( query , callback ) { var cursor = this . _prepareFind ( query ) , iterator = this . _fetchIterator ( query ) , self = this ; ( function next ( err ? ) { if ( err ) return error ( err ) ; cursor . next ( ( err , item ) => { if ( err ) return error ( err ) ; if ( item == null ) { return callback ( ) ; } self . _loadOne ( item , query . isLazy , ( err , value ) => { if ( err ) return error ( err ) ; if ( value === null ) { next ( ) ; } else { iterator ( value , next ) ; } } ) ; } ) ; } ) ( ) ; function error ( err ) { cursor . close ( null ) ; callback ( err ) ; callback = < Callback > function ( ) { } ; } } private _findCursor ( query , callback ) { callback ( null , new CursorImpl ( this . _prepareFind ( query ) , ( document , callback ) => { this . _loadOne ( document , query . isLazy , ( err , entity ) => { if ( err ) return callback ( err ) ; if ( ! query . fetchPaths ) { callback ( null , entity ) ; return ; } this . _session . fetchInternal ( entity , query . fetchPaths , callback ) ; } ) ; } ) ) ; } private _prepareFind ( query ) { var cursor = this . _collection . find ( query . criteria ) ; if ( query . fields ) { cursor . project ( query . fields ) ; } if ( query . orderDocument !== undefined ) { cursor . sort ( query . orderDocument ) ; } if ( query . skipCount !== undefined ) { cursor . skip ( query . skipCount ) ; } if ( query . limitCount !== undefined ) { cursor . limit ( query . limitCount ) ; } if ( query . collationOptions !== undefined ) { cursor . collation ( query . collationOptions ) ; } if ( query . batchSizeValue !== undefined ) { cursor . batchSize ( query . batchSizeValue ) ; } return cursor ; } private _findOneAndModify ( query , callback ) { var options = { returnOriginal : ! query . wantsUpdated , upsert : query . kind == QueryKind . FindOneAndUpsert , sort : < any > query . orderDocument } ; var self = this ; if ( query . kind == QueryKind . FindOneAndRemove ) { this . _collection . findOneAndDelete ( query . criteria , options , handleCallback ) ; } else { this . _collection . findOneAndUpdate ( query . criteria , query . updateDocument , options , handleCallback ) ; } function handleCallback ( err , response ) { if ( err ) return callback ( err ) ; var document = response . value ; if ( ! document ) return callback ( null ) ; var entity = self . _session . getObject ( document [ "_id" ] ) ; if ( entity !== undefined ) { var alreadyLoaded = true ; finish ( ) ; } else { self . _loadOne ( document , query . isLazy , ( err , value ) => { if ( err ) return callback ( err ) ; entity = value ; finish ( ) ; } ) ; } function finish ( ) { if ( entity ) { if ( query . kind == QueryKind . FindOneAndRemove ) { self . _session . notifyRemoved ( entity ) ; } else if ( ! options . returnOriginal && alreadyLoaded ) { self . _refreshFromDocument ( entity , document , ( err ) => { if ( err ) return callback ( err ) ; callback ( null , entity ) ; } ) ; return ; } } callback ( null , entity ) ; } } } private _removeOne ( query , callback ) { this . _collection . deleteOne ( query . criteria , ( err , response ) => { if ( err ) return callback ( err ) ; callback ( null , response . deletedCount ) ; } ) ; } private _removeAll ( query , callback ) { this . _collection . deleteMany ( query . criteria , ( err , response ) => { if ( err ) return callback ( err ) ; callback ( null , response . deletedCount ) ; } ) ; } private _updateOne ( query , callback ) { this . _collection . updateOne ( query . criteria , query . updateDocument , { upsert : query . kind == QueryKind . Upsert } , ( err , response ) => { if ( err ) return callback ( err ) ; callback ( null , response . modifiedCount ) ; } ) ; } private _updateAll ( query , callback ) { this . _collection . updateMany ( query . criteria , query . updateDocument , ( err , response ) => { if ( err ) return callback ( err ) ; callback ( null , response . modifiedCount ) ; } ) ; } private _distinct ( query , callback ) { var context = this . _mapping . resolve ( query . key ) ; if ( context . error ) { return callback ( context . error ) ; } this . _collection . distinct ( context . resolvedPath , query . criteria , undefined , ( err , results ) => { if ( err ) return callback ( err ) ; var readContext = new ReadContext ( this . _session ) ; for ( var i = 0 , l = results . length ; i < l ; i ++ ) { readContext . path = context . resolvedPath ; results [ i ] = context . resolvedMapping . read ( readContext , results [ i ] ) ; if ( readContext . hasErrors ) { return callback ( new PersistenceError ( "Error deserializing distinct values for: " + readContext . getErrorMessage ( ) ) ) ; } } callback ( null , results ) ; } ) ; } private _count ( query , callback ) { var options = { limit : query . limitCount , skip : < any > query . skipCount } ; this . _collection . count ( query . criteria , options , ( err , result ) => { if ( err ) return callback ( err ) ; callback ( null , result ) ; } ) ; } private _fetchOne ( query , callback ) { if ( ! query . fetchPaths ) { return callback ; } return ( err , entity ) => { if ( err ) return callback ( err ) ; if ( ! entity ) return callback ( null ) ; this . _session . fetchInternal ( entity , query . fetchPaths , callback ) ; } ; } private _fetchAll ( query , callback ) { if ( ! query . fetchPaths ) { return callback ; } return ( err , entities ) => { if ( err ) return callback ( err ) ; async . each ( entities , ( entity , done ) => this . _session . fetchInternal ( entity , query . fetchPaths , done ) , err => { if ( err ) return callback ( err ) ; callback ( null , entities ) ; } ) ; } ; } private _fetchIterator ( query ) { if ( ! query . fetchPaths ) { return query . iterator ; } return ( entity , done ) => { if ( ! entity ) return done ( ) ; this . _session . fetchInternal ( entity , query . fetchPaths , ( err , result ) => { if ( err ) return done ( err ) ; query . iterator ( result , done ) ; } ) ; } ; } private _loadOne ( document , reference , callback ) { var entity ; if ( ! document ) { entity = null ; } else if ( reference ) { entity = this . _session . getReferenceInternal ( this . _mapping , document [ "_id" ] ) ; } else { entity = this . _session . getObject ( document [ "_id" ] ) ; if ( entity === undefined ) { var context = new ReadContext ( this . _session ) ; entity = this . _mapping . read ( context , document ) ; if ( context . hasErrors ) { return callback ( new PersistenceError ( "Error deserializing document:\n" + context . getErrorMessage ( ) ) ) ; } this . _session . registerManaged ( this , entity , document ) ; if ( context . fetches ) { this . _session . fetchInternal ( entity , context . fetches , callback ) ; return ; } } } process . nextTick ( ( ) => callback ( null , entity ) ) ; } private _getCommand ( batch ) { var id = this . _mapping . inheritanceRoot . id ; var command = < BulkOperationCommand > batch . getCommand ( id ) ; if ( ! command ) { command = new BulkOperationCommand ( batch , this . _collection , this . _mapping ) ; batch . addCommand ( id , command ) ; } return command ; } } class BulkOperationCommand implements Command { collectionName : string ; operation : mongodb . UnorderedBulkOperation ; inserted : number ; updated : number ; removed : number ; priority : number ; private _mapping : EntityMapping ; private _batch : Batch ; constructor ( batch , collection , mapping ) { this . _mapping = mapping ; this . _batch = batch ; this . priority = ( < EntityMapping > mapping . inheritanceRoot ) . flushPriority ; this . collectionName = collection . collectionName ; this . operation = collection . initializeUnorderedBulkOp ( ) ; this . inserted = this . updated = this . removed = 0 ; } addInsert ( document ) { this . inserted ++ ; try { this . operation . insert ( document ) ; } catch ( err ) { this . _batch . error = new PersistenceError ( "Error adding insert to batch: " + err . message , err ) ; } } addReplace ( document , version ) { var query = { _id : document [ "_id" ] } ; if ( version != null ) { this . _mapping . setDocumentVersion ( query , version ) ; } this . updated ++ ; try { this . operation . find ( query ) . replaceOne ( document ) ; } catch ( err ) { this . _batch . error = new PersistenceError ( "Error adding replace to batch: " + err . message , err ) ; } } addRemove ( id ) { var query = { _id : id } ; this . removed ++ ; try { this . operation . find ( query ) . removeOne ( ) ; } catch ( err ) { this . _batch . error = new PersistenceError ( "Error adding remove to batch: " + err . message , err ) ; } } execute ( callback ) { this . operation . execute ( ( err , result ) => { if ( err ) return callback ( err ) ; if ( ( result . nInserted || 0 ) != this . inserted ) { return callback ( new PersistenceError ( "Flush failed for collection '" + this . collectionName + "'. Expected to insert " + this . inserted + " documents but only inserted " + ( result . nInserted || 0 ) + "." ) ) ; } if ( ( result . nModified || 0 ) != this . updated ) { return callback ( new PersistenceError ( "Flush failed for collection '" + this . collectionName + "'. Expected to update " + this . updated + " documents but only updated " + ( result . nModified || 0 ) + "." ) ) ; } if ( ( result . nRemoved || 0 ) != this . removed ) { return callback ( new PersistenceError ( "Flush failed for collection '" + this . collectionName + "'. Expected to remove " + this . removed + " documents but only removed " + ( result . nRemoved || 0 ) + "." ) ) ; } callback ( ) ; } ) ; } } class FindQueue { private _persister : PersisterImpl ; private _ids : any [ ] ; private _callbacksMap : Map < string , ResultCallback < any > [ ] > ; constructor ( persister ) { this . _persister = persister ; } add ( id , callback ) { if ( ! this . _ids ) { this . _ids = [ ] ; this . _callbacksMap = new Map ( ) ; process . nextTick ( ( ) => this . _process ( ) ) ; } var normalizedId = this . _normalizeIdentifier ( id ) ; if ( normalizedId == null ) { callback ( new PersistenceError ( `'${ id }' is not a valid identifier.` ) ) ; return ; } var key = normalizedId . toString ( ) ; var callbacks = this . _callbacksMap . get ( key ) ; if ( callbacks === undefined ) { callbacks = [ ] ; this . _ids . push ( normalizedId ) ; this . _callbacksMap . set ( key , callbacks ) ; } callbacks . push ( callback ) ; } private _normalizeIdentifier ( id ) { if ( typeof id === "string" ) { return this . _persister . identity . fromString ( id ) ; } else if ( this . _persister . identity . validate ( id ) ) { return id ; } return null ; } private _process ( ) { var callbacksMap = this . _callbacksMap , ids = this . _ids ; this . _ids = this . _callbacksMap = undefined ; if ( ids . length == 1 ) { var id = ids [ 0 ] ; this . _persister . findOne ( { _id : id } , ( err , entity ) => { if ( err ) return handleCallback ( id , err ) ; if ( ! entity ) { return handleCallback ( id , new EntityNotFoundError ( "Unable to find document with identifier '" + id . toString ( ) + "'." ) ) ; } handleCallback ( id , null , entity ) ; } ) ; return ; } this . _persister . findAll ( { _id : { $in : ids } } , ( err , entities ) => { if ( ! err ) { for ( var i = 0 , l = entities . length ; i < l ; i ++ ) { var entity = entities [ i ] ; handleCallback ( entity [ "_id" ] , null , entity ) ; } } callbacksMap . forEach ( ( callbacks , id ) => callCallbacks ( callbacks , err || new EntityNotFoundError ( "Unable to find document with identifier '" + id + "'." ) ) ) ; } ) ; function handleCallback ( id , err , entity ? ) { var key = id . toString ( ) , callbacks = callbacksMap . get ( key ) ; callbacksMap . delete ( key ) ; callCallbacks ( callbacks , err , entity ) ; } function callCallbacks ( callbacks , err , entity ? ) { for ( var i = 0 ; i < callbacks . length ; i ++ ) { callbacks [ i ] ( err , entity ) ; } } } } hidden export interface Cursor < T > { param callback next ( callback ) : void ; param callback close ( callback ? ) : void ; } hidden type EntityLoaded = ( document : Object , callback : ResultCallback < any > ) => void ; hidden class CursorImpl < T > { private _cursor : mongodb . Cursor ; private _loader : EntityLoaded ; constructor ( cursor , loader ) { this . _cursor = cursor ; this . _loader = loader ; } next ( callback ) { this . _cursor . next ( ( err , item ) => { if ( err ) { this . close ( ) ; callback ( err ) ; return ; } if ( item == null ) { callback ( null , null ) ; return ; } this . _loader ( item , ( err , result ) => { if ( err ) { this . close ( ) ; callback ( err ) ; return ; } callback ( null , result ) ; } ) ; } ) ; } close ( callback ? ) { this . _cursor . close ( callback ) ; } }