import { TopologicalSorter } from '../fact/sorter' ; import { Feed , Handler , Observable , Subscription } from '../feed/feed' ; import { WebClient } from '../http/web-client' ; import { Query } from '../query/query' ; import { FactEnvelope , FactRecord , FactReference , factReferenceEquals } from '../storage' ; import { flatten } from '../util/fn' ; import { serializeLoad , serializeQuery , serializeSave } from './serialize' ; class TransientForkSubscription implements Subscription { constructor ( private inner , private loaded ) { } async load ( ) { await this . inner . load ( ) ; await this . loaded ; } dispose ( ) { return this . inner . dispose ( ) ; } } class TransientForkObservable implements Observable { constructor ( private inner , private loaded ) { } subscribe ( added , removed ) { return new TransientForkSubscription ( this . inner . subscribe ( added , removed ) , this . loaded ) ; } } export class TransientFork implements Feed { constructor ( private feed , private client ) { } async save ( envelopes ) { const response = await this . client . save ( serializeSave ( envelopes ) ) ; const saved = await this . feed . save ( envelopes ) ; return saved ; } async query ( start , query ) { if ( query . isDeterministic ( ) ) { const results = await this . feed . query ( start , query ) ; return results ; } else { const response = await this . client . query ( serializeQuery ( start , query ) ) ; return response . results ; } } exists ( fact ) { throw new Error ( "Exists method not implemented on Fork." ) ; } async load ( references ) { const known = await this . feed . load ( references ) ; const remaining = references . filter ( reference => ! known . some ( factReferenceEquals ( reference ) ) ) ; if ( remaining . length === 0 ) { return known ; } else { const records = await this . loadRecords ( remaining ) ; return records . concat ( known ) ; } } from ( fact , query ) { const observable = this . feed . from ( fact , query ) ; const loaded = this . initiateQuery ( fact , query ) ; return new TransientForkObservable ( observable , loaded ) ; } private async initiateQuery ( start , query ) { const queryResponse = await this . client . query ( serializeQuery ( start , query ) ) ; const paths = queryResponse . results ; if ( paths . length > 0 ) { const references = distinct ( flatten ( paths , p => p ) ) ; await this . load ( references ) ; } } private async loadRecords ( references ) { const sorter = new TopologicalSorter < FactRecord > ( ) ; let records = [ ] ; for ( let start = 0 ; start < references . length ; start += 300 ) { const chunk = references . slice ( start , start + 300 ) ; const response = await this . client . load ( serializeLoad ( chunk ) ) ; const facts = sorter . sort ( response . facts , ( p , f ) => f ) ; const envelopes = facts . map ( fact => { return < FactEnvelope > { fact : fact , signatures : [ ] } ; } ) ; await this . feed . save ( envelopes ) ; records = records . concat ( facts ) ; } return records ; } } function distinct ( references ) { const result = [ ] ; references . forEach ( reference => { if ( ! result . some ( r => r . hash === reference . hash && r . type === reference . type ) ) { result . push ( reference ) ; } } ) ; return result ; }