import type { StreamState , Stream } from '../stream' ; import { take , filter } from 'rxjs/operators' ; import { BehaviorSubject } from 'rxjs' ; import { RunningStateLike } from '../running-state-like' ; import { StreamID } from '@ceramicnetwork/streamid' ; class FakeRunningState extends BehaviorSubject < StreamState > implements RunningStateLike { readonly id : StreamID ; readonly state : StreamState ; constructor ( value ) { super ( value ) ; this . state = this . value ; this . id = new StreamID ( this . state . type , this . state . log [ 0 ] . cid ) ; } } export class TestUtils { param stream static registerChangeListener ( stream ) { return stream . pipe ( take ( 1 ) ) . toPromise ( ) ; } param stream param timeout param predicate param onFailure static async waitForState ( stream , timeout , predicate , onFailure ) { if ( predicate ( stream . state ) ) return ; const timeoutPromise = new Promise ( ( resolve ) => setTimeout ( resolve , timeout ) ) ; const completionPromise = stream . pipe ( filter ( ( state ) => predicate ( state ) ) ) . toPromise ( ) ; await Promise . race ( [ timeoutPromise , completionPromise ] ) ; if ( ! predicate ( stream . state ) ) { onFailure ( ) ; } } static runningState ( state ) { return new FakeRunningState ( state ) ; } static async delay ( ms ) { return new Promise < void > ( ( resolve ) => { setTimeout ( ( ) => resolve ( ) , ms ) ; } ) ; } }