import * as async from "async" ; import { EventEmitter } from "events" ; import { Callback } from "./core/callback" ; import { ChangeTrackingType } from "./mapping/mappingModel" ; import { Constructor } from "./index" ; import { MappingModel } from "./mapping/mappingModel" ; import { ResultCallback } from "./core/callback" ; import { InternalSessionFactory , SessionFactory } from "./sessionFactory" ; import { TaskQueue } from "./taskQueue" ; import { DirtyCheckContext , Persister } from "./persister" ; import { Batch } from "./batch" ; import { Reference } from "./reference" ; import { Table } from "./core/table" ; import { EntityMapping } from "./mapping/entityMapping" ; import { QueryBuilder , QueryBuilderImpl , FindOneQuery } from "./query/queryBuilder" ; import { QueryDefinition } from "./query/queryDefinition" ; import { Observer } from "./observer" ; import { PersistenceError } from "./persistenceError" ; import { WriteContext } from "./mapping/writeContext" ; import { getDuration } from "./core/timerUtil" ; /**
 * The state of an object.
 */ const enum ObjectState { /**
     * Managed entities have a persistent identity and are associated with a session.
     */ Managed , /**
     * Detached entities have a persistent identity and are not currently associated with a session.
     */ Detached , /**
     * Removed entities have a persistent identity, are associated with a session, and are scheduled for deletion
     * from the database. Once the entity is deleted from the database, the entity is no longer managed and is
     * considered a new entity.
     */ Removed } const enum ObjectFlags { None = 0 , ReadOnly = 0x00000001 , Dirty = 0x00000002 , Observing = 0x00000004 } const enum ScheduledOperation { None = 0 , Insert , Update , Delete , DirtyCheck } const enum Action { Save = 0x00000001 , Remove = 0x00000002 , Detach = 0x00000004 , Flush = 0x00000008 , Clear = 0x00000010 , Refresh = 0x00000020 , Merge = 0x00000040 , Fetch = 0x00000080 , FindQuery = 0x00000100 , ModifyQuery = 0x00000200 , Wait = 0x00000400 , Close = 0x00000800 , ReadOnly = Fetch | FindQuery , All = Save | Remove | Detach | Flush | Clear | Refresh | Merge | Fetch | FindQuery | ModifyQuery | Wait | Close } interface ObjectLinks { state : ObjectState ; flags : ObjectFlags ; object ? : any ; originalDocument ? : any ; persister ? : Persister ; scheduledOperation ? : ScheduledOperation ; next ? : ObjectLinks ; prev ? : ObjectLinks ; index ? : number ; observer ? : Observer ; } /**
 * Static ObjectLinks returned from getObjectLinks for detached objects. There is no need to create a new ObjectLinks
 * object for each detached object.
 */ var cachedDetachedLinks = { state : ObjectState . Detached , flags : ObjectFlags . None } ; /**
 *  The primary interface used to interact with the database. The Session is a
 *  [Unit of Work](http://martinfowler.com/eaaCatalog/unitOfWork.html) that keeps track of changes until they are
 *  synchronized with the database by calling [[flush]] or [[close]].
 */ export interface Session { /**
     * The [SessionFactory] that created this [Session].
     */ factory : SessionFactory ; param obj param callback save ( obj , callback ? ) : void ; param obj param callback remove ( obj , callback ? ) : void ; param obj param callback detach ( obj , callback ? ) : void ; param obj param callback refresh ( obj , callback ) : void ; param callback flush ( callback ? ) : void ; param callback clear ( callback ? ) : void ; param ctr param id param callback find < T > ( ctr , id , callback ? ) : FindOneQuery < T > ; param ctr param id param callback exists < T > ( ctr , id , callback ) : void ; param ctr param id getReference < T > ( ctr , id ) : T ; param obj param callback fetch < T > ( obj , callback ? ) : void ; param obj param path param callback fetch < T > ( obj , path , callback ? ) : void ; query < T > ( ctr ) : QueryBuilder < T > ; param callback wait ( callback ? ) : void ; param callback close ( callback ? ) : void ; param obj contains ( obj ) : boolean ; param obj param callback toDocument ( obj , callback ? ) : Object ; param obj param callback isDirty ( obj , callback ? ) : boolean ; param obj getVersion ( obj ) : number ; /**
     * Gets an array of all entities managed by the session.
     */ getEntities ( ) : Object [ ] ; param event param listener on ( event , listener ) : EventEmitter ; } hidden export interface InternalSession extends Session { factory : InternalSessionFactory ; getObject ( id ) : any ; registerManaged ( persister , entity , document ) : void ; notifyRemoved ( entity ) : void ; getPersister ( mapping ) : Persister ; getReferenceInternal ( mapping , id ) : any ; fetchInternal ( entity , paths , callback ) : void ; executeQuery ( query , callback ) : void ; } hidden export class SessionImpl extends EventEmitter implements InternalSession { /**
     * Cached Persisters by Mapping
     */ private _persisters : Table < Persister > = [ ] ; /**
     * Hash table containing all objects by id associated with the session.
     */ private _objectLinksById : Map < string , ObjectLinks > = new Map ( ) ; /**
     * List of all ObjectLinks associated with the session.
     */ private _objectLinks : ObjectLinks [ ] = [ ] ; /**
     * Scheduled operations linked list head.
     */ private _scheduleHead : ObjectLinks ; /**
     * Scheduled operations linked list tail.
     */ private _scheduleTail : ObjectLinks ; /**
     * Task queue coordinates asynchronous actions.
     */ private _queue : TaskQueue ; private _traceEnabled : boolean ; constructor ( public factory ) { super ( ) ; this . _createTaskQueue ( ) ; this . _traceEnabled = this . factory . logger != null ; } save ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } this . _queue . add ( Action . Save , Action . All & ~ ( Action . Save | Action . ReadOnly ) , obj , callback ) ; } remove ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } this . _queue . add ( Action . Remove , Action . All & ~ Action . Remove , obj , callback ) ; } refresh ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } this . _queue . add ( Action . Refresh , Action . All & ~ Action . Refresh , obj , callback ) ; } detach ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } this . _queue . add ( Action . Detach , Action . All & ~ Action . Detach , obj , callback ) ; } clear ( callback ? ) { if ( this . _queue . invalid ) { this . _createTaskQueue ( ) ; } this . _queue . add ( Action . Clear , Action . All , undefined , callback ) ; } flush ( callback ? ) { this . _queue . add ( Action . Flush , Action . All , undefined , callback ) ; } wait ( callback ? ) { this . _queue . add ( Action . Wait , Action . All , undefined , callback ) ; } find < T > ( ctr , id , callback ? ) { return this . query ( ctr ) . findOneById ( id , callback ) ; } exists < T > ( ctr , id , callback ) { var obj = this . getObject ( id ) ; if ( obj ) { process . nextTick ( ( ) => callback ( null , true ) ) ; return ; } if ( obj === null ) { process . nextTick ( ( ) => callback ( null , false ) ) ; return ; } this . query ( ctr ) . count ( { _id : id } , ( err , count ) => { if ( err ) return callback ( err ) ; callback ( null , count >= 1 ) ; } ) ; } fetch < T > ( obj , pathsOrCallback , callback ? ) { var paths ; if ( typeof pathsOrCallback === "function" ) { callback = pathsOrCallback ; } else if ( typeof pathsOrCallback === "string" ) { paths = [ pathsOrCallback ] ; } else { paths = pathsOrCallback ; } this . _queue . add ( Action . Fetch , Action . All & ~ Action . ReadOnly , [ obj , paths ] , callback ) ; } close ( callback ? ) { this . _queue . add ( Action . Close , Action . All , undefined , callback ) ; } executeQuery ( query , callback ) { if ( query . readOnly ) { this . _queue . add ( Action . FindQuery , Action . All & ~ Action . ReadOnly , query , callback ) ; } else { this . _queue . add ( Action . ModifyQuery , Action . All , query , callback ) ; } } param obj contains ( obj ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } var id = obj [ "id" ] ; if ( id ) { var links = this . _objectLinksById . get ( id ) ; if ( links && links . state == ObjectState . Managed ) { return true ; } } return false ; } /**
     * Gets an array of all entities managed by the session.
     */ getEntities ( ) { var ret = new Array ( this . _objectLinksById . size ) , i = 0 ; this . _objectLinksById . forEach ( ( value ) => { ret [ i ++ ] = value . object ; } ) ; return ret ; } param ctr param id param callback getReference < T > ( ctr , id ) { var mapping = this . factory . getMappingForConstructor ( ctr ) ; if ( mapping ) { var identityGenerator = ( < EntityMapping > mapping . inheritanceRoot ) . identity ; if ( typeof id === "string" ) { id = identityGenerator . fromString ( id ) ; } else { if ( ! identityGenerator . validate ( id ) ) { id = null ; } } } return this . getReferenceInternal ( mapping , id ) ; } param obj param callback toDocument ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } var mapping = this . factory . getMappingForObject ( obj ) ; if ( ! mapping ) { if ( callback ) { callback ( new PersistenceError ( "Object type is not mapped as an entity." ) ) ; } return null ; } var context = new WriteContext ( ) ; var document = mapping . write ( context , obj ) ; if ( context . hasErrors ) { if ( callback ) { callback ( new PersistenceError ( `Error serializing document:\n${ context . getErrorMessage ( ) }` ) ) ; } return null ; } if ( callback ) { callback ( null , document ) ; } return document ; } isDirty ( obj , callback ? ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } var links = this . _getObjectLinks ( obj ) ; if ( ! links || links . state == ObjectState . Detached ) { if ( callback ) { callback ( new PersistenceError ( "Object is not managed." ) ) ; } return null ; } if ( ! links . originalDocument ) { if ( callback ) { process . nextTick ( ( ) => callback ( null , false ) ) ; } return false ; } var context = { } , areEqual = links . persister . areDocumentsEqual ( context , obj , links . originalDocument ) ; if ( context . error ) { if ( callback ) { callback ( context . error ) ; } return null ; } if ( callback ) { process . nextTick ( ( ) => callback ( null , ! areEqual ) ) ; } return ! areEqual ; } param obj getVersion ( obj ) { if ( typeof obj !== "object" ) { throw new Error ( "'obj' argument should be an object." ) ; } var links = this . _getObjectLinks ( obj ) ; if ( ! links || links . state == ObjectState . Detached ) { return null ; } if ( ! links . originalDocument ) { return 0 ; } var mapping = this . factory . getMappingForObject ( obj ) ; if ( ! mapping ) { return null ; } return mapping . getDocumentVersion ( links . originalDocument ) ; } getReferenceInternal ( mapping , id ) { return this . getObject ( id ) || new Reference ( mapping , id ) ; } param id getObject ( id ) { if ( id ) { var links = this . _objectLinksById . get ( id . toString ( ) ) ; if ( links ) { switch ( links . state ) { case ObjectState . Removed : return null ; case ObjectState . Managed : return links . object ; } } } } registerManaged ( persister , entity , document ) { var links = this . _linkObject ( entity , persister ) ; links . originalDocument = document ; this . _trackChanges ( links ) ; } private _trackChanges ( links ) { if ( ( links . flags & ObjectFlags . Dirty ) || links . persister . changeTracking == ChangeTrackingType . DeferredImplicit ) { this . _makeDirty ( links ) ; return ; } } private _stopWatching ( links ) { if ( links . observer ) { links . observer . destroy ( ) ; links . observer = undefined ; } } getPersister ( mapping ) { return this . _persisters [ mapping . id ] || ( this . _persisters [ mapping . id ] = this . factory . createPersister ( this , mapping ) ) ; } query < T > ( ctr ) { return new QueryBuilderImpl < T > ( this , ctr ) ; } param action param arg param callback private _execute ( action , arg , callback ) { switch ( action ) { case Action . Save : this . _save ( arg , callback ) ; break ; case Action . Remove : this . _remove ( arg , callback ) ; break ; case Action . Detach : this . _detach ( arg , callback ) ; break ; case Action . Refresh : this . _refresh ( arg , callback ) ; break ; case Action . Clear : this . _clear ( callback ) ; break ; case Action . Flush : this . _flush ( callback ) ; break ; case Action . Wait : if ( callback ) { process . nextTick ( ( ) => callback ( null ) ) ; } break ; case Action . Fetch : this . fetchInternal ( arg [ 0 ] , arg [ 1 ] , callback ) ; break ; case Action . FindQuery : case Action . ModifyQuery : ( < QueryDefinition > arg ) . executeInternal ( callback ) ; break ; case Action . Close : this . _close ( callback ) ; break ; } } private _save ( obj , callback ) { if ( Reference . isReference ( obj ) ) { return callback ( new PersistenceError ( "Reference passed to save" ) ) ; } this . _findReferencedEntities ( obj , MappingModel . PropertyFlags . CascadeSave , ( err , entities ) => { if ( err ) return callback ( err ) ; this . _saveEntities ( entities , callback ) ; } ) ; } private _saveEntities ( entities , callback ) { for ( var i = 0 , l = entities . length ; i < l ; i ++ ) { var obj = entities [ i ] ; var links = this . _getObjectLinks ( obj ) ; if ( ! links ) { var mapping = this . factory . getMappingForObject ( obj ) ; if ( ! mapping ) { callback ( new PersistenceError ( "Object type is not mapped as an entity." ) ) ; return ; } var persister = this . getPersister ( mapping ) ; var id = obj [ "_id" ] = persister . identity . generate ( ) ; obj [ "id" ] = id . toString ( ) ; links = this . _linkObject ( obj , persister ) ; this . _scheduleOperation ( links , ScheduledOperation . Insert ) ; } else { switch ( links . state ) { case ObjectState . Managed : if ( links . persister . changeTracking == ChangeTrackingType . DeferredExplicit ) { this . _makeDirty ( links ) ; } break ; case ObjectState . Detached : callback ( new PersistenceError ( "Cannot save a detached object." ) ) ; return ; case ObjectState . Removed : if ( links . scheduledOperation == ScheduledOperation . Delete ) { this . _clearScheduledOperation ( links ) ; this . _trackChanges ( links ) ; } else { this . _scheduleOperation ( links , ScheduledOperation . Insert ) ; } links . state = ObjectState . Managed ; break ; } } } callback ( ) ; } private _makeDirty ( links ) { links . flags |= ObjectFlags . Dirty ; if ( ! links . scheduledOperation ) { this . _scheduleOperation ( links , ScheduledOperation . DirtyCheck ) ; } } private _clearDirty ( links ) { links . flags &= ~ ObjectFlags . Dirty ; if ( links . scheduledOperation == ScheduledOperation . DirtyCheck ) { this . _clearScheduledOperation ( links ) ; } } param entity notifyRemoved ( entity ) { if ( ! this . _removeEntity ( entity , false ) ) { throw new PersistenceError ( "Cannot remove a detached object." ) ; } } private _remove ( obj , callback ) { Reference . fetch ( this , obj , ( err , entity ) => { if ( err ) return callback ( err ) ; this . _findReferencedEntities ( entity , MappingModel . PropertyFlags . CascadeRemove | MappingModel . PropertyFlags . Dereference , ( err , entities ) => { if ( err ) return callback ( err ) ; this . _removeEntities ( entities , callback ) ; } ) ; } ) ; } private _removeEntities ( entities , callback ) { for ( var i = entities . length - 1 ; i >= 0 ; i -- ) { var obj = entities [ i ] ; if ( ! this . _removeEntity ( obj , true ) ) { callback ( new PersistenceError ( "Cannot remove a detached object." ) ) ; return ; } } callback ( ) ; } private _removeEntity ( obj , scheduleDelete ) { var links = this . _getObjectLinks ( obj ) ; if ( links ) { switch ( links . state ) { case ObjectState . Managed : links . state = ObjectState . Removed ; if ( links . scheduledOperation == ScheduledOperation . Insert ) { this . _unlinkObject ( links ) ; this . _clearScheduledOperation ( links ) ; } else { if ( scheduleDelete ) { this . _scheduleOperation ( links , ScheduledOperation . Delete ) ; } } break ; case ObjectState . Detached : return false ; } } return true ; } private _detach ( obj , callback ) { if ( Reference . isReference ( obj ) ) { return callback ( new PersistenceError ( "Reference passed to detach" ) ) ; } this . _findReferencedEntities ( obj , MappingModel . PropertyFlags . CascadeDetach , ( err , entities ) => { if ( err ) return callback ( err ) ; this . _detachEntities ( entities , callback ) ; } ) ; } private _detachEntities ( entities , callback ) { for ( var i = 0 , l = entities . length ; i < l ; i ++ ) { var links = this . _getObjectLinks ( entities [ i ] ) ; if ( links && links . state == ObjectState . Managed ) { this . _unlinkObject ( links ) ; this . _clearScheduledOperation ( links ) ; } } callback ( ) ; } private _refresh ( obj , callback ) { if ( Reference . isReference ( obj ) ) { return callback ( new PersistenceError ( "Reference passed to refresh" ) ) ; } this . _findReferencedEntities ( obj , MappingModel . PropertyFlags . CascadeRefresh , ( err , entities ) => { if ( err ) return callback ( err ) ; this . _refreshEntities ( entities , callback ) ; } ) ; } private _refreshEntities ( entities , callback ) { async . each ( entities , ( entity , done ) => { var links = this . _getObjectLinks ( entity ) ; if ( ! links || links . state != ObjectState . Managed ) { return done ( new PersistenceError ( "Object is not managed." ) ) ; } this . _stopWatching ( links ) ; links . persister . refresh ( links . object , ( err , document ) => { if ( err ) return done ( err ) ; links . originalDocument = document ; this . _clearDirty ( links ) ; this . _trackChanges ( links ) ; done ( ) ; } ) ; } , callback ) ; } private _flush ( callback ) { setImmediate ( ( ) => { var head = this . _scheduleHead ; this . _scheduleHead = this . _scheduleTail = null ; if ( this . _traceEnabled ) { var start = process . hrtime ( ) ; } var batch = new Batch ( ) ; this . _buildBatch ( batch , head , ( err ) => { if ( err ) return callback ( err ) ; batch . execute ( ( err ) => { if ( err ) return callback ( err ) ; if ( start ) { var duration = getDuration ( start ) ; } this . _batchCompleted ( head , duration , callback ) ; } ) ; } ) ; } ) ; } param batch param links param callback private _buildBatch ( batch , links , callback ) { var count = 0 ; while ( links && count < 1000 ) { if ( links . scheduledOperation == ScheduledOperation . DirtyCheck ) { var document = links . persister . dirtyCheck ( batch , links . object , links . originalDocument ) ; if ( document ) { links . scheduledOperation = ScheduledOperation . Update ; links . originalDocument = document ; } } else if ( links . scheduledOperation == ScheduledOperation . Insert ) { var document = links . persister . addInsert ( batch , links . object ) ; if ( document ) { links . originalDocument = document ; } } else if ( links . scheduledOperation == ScheduledOperation . Delete ) { links . persister . addRemove ( batch , links . object ) ; } if ( batch . error ) { return callback ( batch . error ) ; } links = links . next ; count ++ ; } if ( links ) { process . nextTick ( ( ) => this . _buildBatch ( batch , links , callback ) ) ; } else { callback ( ) ; } } param head param duration param callback private _batchCompleted ( head , duration , callback ) { if ( this . _traceEnabled ) { this . _logFlushStats ( head , duration ) ; } var links = head ; while ( links ) { var next = links . next ; links . scheduledOperation = ScheduledOperation . None ; links . flags = ObjectFlags . None ; links . prev = links . next = null ; switch ( links . state ) { case ObjectState . Removed : this . _unlinkObject ( links ) ; this . _clearIdentifier ( links ) ; break ; case ObjectState . Managed : this . _trackChanges ( links ) ; break ; default : return callback ( new PersistenceError ( "Unexpected object state in flush." ) ) ; } links = next ; } callback ( ) ; } private _logFlushStats ( head , duration ) { var inserted = 0 , updated = 0 , removed = 0 , dirtyChecked = 0 ; var links = head ; while ( links ) { switch ( links . scheduledOperation ) { case ScheduledOperation . Update : updated ++ ; break ; case ScheduledOperation . Delete : inserted ++ ; break ; case ScheduledOperation . Insert : removed ++ ; break ; case ScheduledOperation . DirtyCheck : dirtyChecked ++ ; break ; } links = links . next ; } this . factory . logger . trace ( { duration , operations : { inserted , updated , removed , dirtyChecked : dirtyChecked + updated } } , `[Hydrate] Completed flush.` ) ; } private _close ( callback ) { this . _queue . close ( ) ; this . _flush ( ( err ) => { if ( err ) return callback ( err ) ; this . _clear ( callback ) ; } ) ; } fetchInternal ( obj , paths , callback ) { if ( obj == null ) { return callback ( null , obj ) ; } if ( Array . isArray ( obj ) ) { async . map ( obj , ( item , done ) => this . fetchInternal ( item , paths , done ) , callback ) ; return ; } Reference . fetch ( this , obj , ( err , entity ) => { if ( err ) return callback ( err ) ; this . _fetchPaths ( entity , paths , callback ) ; } ) ; } private _fetchPaths ( obj , paths , callback ) { var links = this . _getObjectLinks ( obj ) ; if ( ! links || links . state == ObjectState . Detached ) { return callback ( new PersistenceError ( "Object is not associated with the session." ) ) ; } if ( ! paths || paths . length === 0 ) { process . nextTick ( ( ) => callback ( null , obj ) ) ; } else { var persister = links . persister ; async . each ( paths , ( path , done ) => { persister . fetch ( obj , path , done ) ; } , ( err ) => { if ( err ) return callback ( err ) ; callback ( null , obj ) ; } ) ; } } param callback private _clear ( callback ) { for ( var i = 0 ; i < this . _objectLinks . length ; i ++ ) { var links = this . _objectLinks [ i ] ; if ( links ) { this . _cleanupUnlinkedObject ( links ) ; } } this . _objectLinksById = new Map ( ) ; this . _objectLinks = [ ] ; this . _scheduleHead = this . _scheduleTail = null ; process . nextTick ( callback ) ; } private _getObjectLinks ( obj ) { var id = obj [ "id" ] ; if ( id ) { var links = this . _objectLinksById . get ( id ) ; if ( ! links || links . object !== obj ) { return cachedDetachedLinks ; } return links ; } } private _linkObject ( obj , persister ) { var id = obj [ "id" ] ; if ( id == null ) { throw new PersistenceError ( "Object is missing identifier." ) ; } if ( this . _objectLinksById . has ( id ) ) { throw new PersistenceError ( "Session already contains an entity with identifier '" + id + "'." ) ; } var links = { state : ObjectState . Managed , flags : ObjectFlags . None , object : obj , persister : persister , index : this . _objectLinks . length } ; this . _objectLinksById . set ( id , links ) ; this . _objectLinks . push ( links ) ; return links ; } private _unlinkObject ( links ) { this . _objectLinksById . delete ( links . object [ "id" ] ) ; this . _objectLinks [ links . index ] = undefined ; this . _cleanupUnlinkedObject ( links ) ; } private _cleanupUnlinkedObject ( links ) { this . _stopWatching ( links ) ; if ( links . scheduledOperation == ScheduledOperation . Insert ) { this . _clearIdentifier ( links ) ; } } private _scheduleOperation ( links , operation ) { if ( ! links . scheduledOperation ) { if ( this . _scheduleHead ) { this . _scheduleTail . next = links ; links . prev = this . _scheduleTail ; this . _scheduleTail = links ; } else { this . _scheduleHead = this . _scheduleTail = links ; } } links . scheduledOperation = operation ; } private _clearScheduledOperation ( links ) { if ( ! links . scheduledOperation ) { return ; } if ( links . prev ) { links . prev . next = links . next ; } else { this . _scheduleHead = links . next ; } if ( links . next ) { links . next . prev = links . prev ; } else { this . _scheduleTail = links . prev ; } links . scheduledOperation = ScheduledOperation . None ; } private _clearIdentifier ( links ) { links . object [ "_id" ] = links . object [ "id" ] = undefined ; } private _findReferencedEntities ( obj , flags , callback ) { var mapping = this . factory . getMappingForObject ( obj ) ; if ( ! mapping ) { callback ( new PersistenceError ( "Object type is not mapped as an entity." ) ) ; return ; } var entities = [ ] , embedded = [ ] ; this . _walk ( mapping , obj , flags , entities , embedded , err => { if ( err ) return callback ( err ) ; process . nextTick ( ( ) => callback ( null , entities ) ) ; } ) ; } private _walk ( mapping , entity , flags , entities , embedded , callback ) { var references = [ ] ; mapping . walk ( this , entity , flags | MappingModel . PropertyFlags . WalkEntities , entities , embedded , references ) ; async . each ( references , ( reference , done ) => { reference . fetch ( this , ( err , entity ) => { if ( err ) return done ( err ) ; this . _walk ( reference . mapping , entity , flags , entities , embedded , done ) ; } ) ; } , callback ) ; } private _createTaskQueue ( ) { if ( this . _queue ) { this . _queue . removeAllListeners ( ) ; } this . _queue = new TaskQueue ( ( action , args , callback ) => this . _execute ( action , args , callback ) ) ; this . _queue . on ( 'error' , ( err ) => { this . emit ( 'error' , err ) ; } ) ; } }