// Path: src/main/java/org/reactivestreams/ExpectSubscriptionsException.java
// public class ExpectSubscriptionsException extends RuntimeException {
// 
//     public ExpectSubscriptionsException(String message, String caller) {
//         super(message + "\n\n from assertion at " + caller + "\n\n----------------------\n");
//     }
// 
// }


import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.annotations.NonNull;
import io.reactivex.schedulers.TestScheduler;
import org.reactivestreams.*;
import org.reactivestreams.ExpectSubscriptionsException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

    public <T> ColdObservable<T> createColdObservable(String marbles) {
        return createColdObservable(marbles, null);
    }

    public <T> HotObservable<T> createHotObservable(String marbles, Map<String, T> values) {
        List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
        return HotObservable.create(this, notifications);
    }

    public <T> HotObservable<T> createHotObservable(String marbles) {
        return createHotObservable(marbles, null);
    }


    public long createTime(String marbles) {
        int endIndex = marbles.indexOf("|");
        if (endIndex == -1) {
            throw new RuntimeException("Marble diagram for time should have a completion marker '|'");
        }

        return endIndex * frameTimeFactor;
    }

    public void flush() {
        testScheduler.advanceTimeTo(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
        try {
            state.flush();
        } catch (ExpectPublisherException ex) {
            throw new ExpectFlowableException(ex.getMessage());
// Path: src/main/java/org/reactivestreams/ExpectSubscriptionsException.java
// public class ExpectSubscriptionsException extends RuntimeException {
// 
//     public ExpectSubscriptionsException(String message, String caller) {
//         super(message + "\n\n from assertion at " + caller + "\n\n----------------------\n");
//     }
// 
// }


// Path: src/main/java/io/reactivex/marble/MarbleScheduler.java
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.annotations.NonNull;
import io.reactivex.schedulers.TestScheduler;
import org.reactivestreams.*;
import org.reactivestreams.ExpectSubscriptionsException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;


    public <T> ColdObservable<T> createColdObservable(String marbles) {
        return createColdObservable(marbles, null);
    }

    public <T> HotObservable<T> createHotObservable(String marbles, Map<String, T> values) {
        List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
        return HotObservable.create(this, notifications);
    }

    public <T> HotObservable<T> createHotObservable(String marbles) {
        return createHotObservable(marbles, null);
    }


    public long createTime(String marbles) {
        int endIndex = marbles.indexOf("|");
        if (endIndex == -1) {
            throw new RuntimeException("Marble diagram for time should have a completion marker '|'");
        }

        return endIndex * frameTimeFactor;
    }

    public void flush() {
        testScheduler.advanceTimeTo(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
        try {
            state.flush();
        } catch (ExpectPublisherException ex) {
            throw new ExpectFlowableException(ex.getMessage());
        } catch (ExpectSubscriptionsException ex) {