// Path: src/main/java/reactor/ColdFlux.java
// public class ColdFlux<T> extends Flux<T> implements TestablePublisher<T> {
// 
//     private final TestablePublisher<T> publisher;
// 
//     protected ColdFlux(TestablePublisher<T> publisher) {
//         this.publisher = publisher;
//     }
// 
//     @Override
//     public void subscribe(Subscriber<? super T> s) {
//         publisher.subscribe(s);
//     }
// 
//     @Override
//     public List<SubscriptionLog> getSubscriptions() {
//         return publisher.getSubscriptions();
//     }
// 
//     @Override
//     public List<Recorded<T>> getMessages() {
//         return publisher.getMessages();
//     }
// 
//     public static <T> ColdFlux<T> create(Scheduler scheduler, Recorded<T>... notifications) {
//         return create(scheduler, Arrays.asList(notifications));
//     }
// 
//     public static <T> ColdFlux<T> create(final Scheduler scheduler, List<Recorded<T>> notifications) {
// 
//         ColdPublisher<T> coldPublisher = new ColdPublisher<>(new SchedulerFactory() {
//             @Override
//             public org.reactivestreams.Scheduler create() {
//                 return new SchedulerAdapter(scheduler);
//             }
//         }, notifications);
// 
//         return new ColdFlux<>(coldPublisher);
//     }
// 
// }
// 
// Path: src/main/java/reactor/HotFlux.java
// public class HotFlux<T> extends Flux<T> implements TestablePublisher<T> {
// 
//     private final TestablePublisher<T> publisher;
// 
//     protected HotFlux(TestablePublisher<T> publisher) {
//         this.publisher = publisher;
//     }
// 
//     @Override
//     public void subscribe(Subscriber<? super T> s) {
//         publisher.subscribe(s);
//     }
// 
//     @Override
//     public List<SubscriptionLog> getSubscriptions() {
//         return publisher.getSubscriptions();
//     }
// 
//     @Override
//     public List<Recorded<T>> getMessages() {
//         return publisher.getMessages();
//     }
// 
//     public static <T> HotFlux<T> create(Scheduler scheduler, Recorded<T>... notifications) {
//         return create(scheduler, Arrays.asList(notifications));
//     }
// 
//     public static <T> HotFlux<T> create(Scheduler scheduler, List<Recorded<T>> notifications) {
//         HotPublisher<T> hotPublisher = new HotPublisher<>(new SchedulerAdapter(scheduler), notifications);
//         return new HotFlux<>(hotPublisher);
//     }
// 
// }
// 
// Path: src/main/java/reactor/MarbleScheduler.java
// public class MarbleScheduler extends VirtualTimeScheduler {
//     private final MarbleSchedulerState state;
//     private final long frameTimeFactor;
// 
//     public MarbleScheduler() {
//         this(10);
//     }
// 
//     public MarbleScheduler(long frameTimeFactor) {
//         this.frameTimeFactor = frameTimeFactor;
//         state = new MarbleSchedulerState(frameTimeFactor, new MarbleSchedulerState.ISchedule() {
//             @Override
//             public long now() {
//                 return MarbleScheduler.this.now(TimeUnit.MILLISECONDS);
//             }
// 
//             @Override
//             public void schedule(Runnable runnable, long time) {
//                 MarbleScheduler.this.schedule(runnable, time, TimeUnit.MILLISECONDS);
//             }
//         }, getClass());
//     }
// 
// 
//     public <T> ColdFlux<T> createColdFlux(String marbles, Map<String, T> values) {
//         List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
//         return ColdFlux.create(this, notifications);
//     }
// 
//     public <T> ColdFlux<T> createColdFlux(String marbles) {
//         return createColdFlux(marbles, null);
//     }
// 
//     public <T> HotFlux<T> createHotFlux(String marbles, Map<String, T> values) {
//         List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
//         return HotFlux.create(this, notifications);
//     }
// 
//     public <T> HotFlux<T> createHotFlux(String marbles) {
//         return createHotFlux(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() {
//         advanceTimeTo(Instant.ofEpochMilli(Long.MAX_VALUE));
//         state.flush();
//     }
// 
//     public <T> ISetupTest expectFlux(Flux<T> flux) {
//         return expectFlux(flux, null);
//     }
// 
//     public <T> ISetupTest expectFlux(Flux<T> flux, String unsubscriptionMarbles) {
//         return state.expectPublisher(flux, unsubscriptionMarbles);
//     }
// 
//     public ISetupSubscriptionsTest expectSubscriptions(List<SubscriptionLog> subscriptions) {
//         return state.expectSubscriptions(subscriptions);
//     }
// }
// 
// Path: src/main/java/io/reactivex/marble/MapHelper.java
// public static <V> Map<String, V> of(String k, V v) {
//     return Collections.singletonMap(k,v);
// }


import io.reactivex.subscribers.TestSubscriber;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.ColdFlux;
import reactor.HotFlux;
import reactor.MarbleScheduler;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
import static io.reactivex.marble.MapHelper.of;
package org.reactivestreams;




public class MarbleSchedulerTest {

    private MarbleScheduler scheduler;

    @Before
    public void setupScheduler() {
        scheduler = new MarbleScheduler();
    }

    @After
    public void flushScheduler() {
        if (scheduler != null) {
            scheduler.flush();
        }
    }

    @Test
    public void should_create_a_cold_observable() {
// Path: src/main/java/reactor/ColdFlux.java
// public class ColdFlux<T> extends Flux<T> implements TestablePublisher<T> {
// 
//     private final TestablePublisher<T> publisher;
// 
//     protected ColdFlux(TestablePublisher<T> publisher) {
//         this.publisher = publisher;
//     }
// 
//     @Override
//     public void subscribe(Subscriber<? super T> s) {
//         publisher.subscribe(s);
//     }
// 
//     @Override
//     public List<SubscriptionLog> getSubscriptions() {
//         return publisher.getSubscriptions();
//     }
// 
//     @Override
//     public List<Recorded<T>> getMessages() {
//         return publisher.getMessages();
//     }
// 
//     public static <T> ColdFlux<T> create(Scheduler scheduler, Recorded<T>... notifications) {
//         return create(scheduler, Arrays.asList(notifications));
//     }
// 
//     public static <T> ColdFlux<T> create(final Scheduler scheduler, List<Recorded<T>> notifications) {
// 
//         ColdPublisher<T> coldPublisher = new ColdPublisher<>(new SchedulerFactory() {
//             @Override
//             public org.reactivestreams.Scheduler create() {
//                 return new SchedulerAdapter(scheduler);
//             }
//         }, notifications);
// 
//         return new ColdFlux<>(coldPublisher);
//     }
// 
// }
// 
// Path: src/main/java/reactor/HotFlux.java
// public class HotFlux<T> extends Flux<T> implements TestablePublisher<T> {
// 
//     private final TestablePublisher<T> publisher;
// 
//     protected HotFlux(TestablePublisher<T> publisher) {
//         this.publisher = publisher;
//     }
// 
//     @Override
//     public void subscribe(Subscriber<? super T> s) {
//         publisher.subscribe(s);
//     }
// 
//     @Override
//     public List<SubscriptionLog> getSubscriptions() {
//         return publisher.getSubscriptions();
//     }
// 
//     @Override
//     public List<Recorded<T>> getMessages() {
//         return publisher.getMessages();
//     }
// 
//     public static <T> HotFlux<T> create(Scheduler scheduler, Recorded<T>... notifications) {
//         return create(scheduler, Arrays.asList(notifications));
//     }
// 
//     public static <T> HotFlux<T> create(Scheduler scheduler, List<Recorded<T>> notifications) {
//         HotPublisher<T> hotPublisher = new HotPublisher<>(new SchedulerAdapter(scheduler), notifications);
//         return new HotFlux<>(hotPublisher);
//     }
// 
// }
// 
// Path: src/main/java/reactor/MarbleScheduler.java
// public class MarbleScheduler extends VirtualTimeScheduler {
//     private final MarbleSchedulerState state;
//     private final long frameTimeFactor;
// 
//     public MarbleScheduler() {
//         this(10);
//     }
// 
//     public MarbleScheduler(long frameTimeFactor) {
//         this.frameTimeFactor = frameTimeFactor;
//         state = new MarbleSchedulerState(frameTimeFactor, new MarbleSchedulerState.ISchedule() {
//             @Override
//             public long now() {
//                 return MarbleScheduler.this.now(TimeUnit.MILLISECONDS);
//             }
// 
//             @Override
//             public void schedule(Runnable runnable, long time) {
//                 MarbleScheduler.this.schedule(runnable, time, TimeUnit.MILLISECONDS);
//             }
//         }, getClass());
//     }
// 
// 
//     public <T> ColdFlux<T> createColdFlux(String marbles, Map<String, T> values) {
//         List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
//         return ColdFlux.create(this, notifications);
//     }
// 
//     public <T> ColdFlux<T> createColdFlux(String marbles) {
//         return createColdFlux(marbles, null);
//     }
// 
//     public <T> HotFlux<T> createHotFlux(String marbles, Map<String, T> values) {
//         List<Recorded<T>> notifications = Parser.parseMarbles(marbles, values, null, frameTimeFactor);
//         return HotFlux.create(this, notifications);
//     }
// 
//     public <T> HotFlux<T> createHotFlux(String marbles) {
//         return createHotFlux(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() {
//         advanceTimeTo(Instant.ofEpochMilli(Long.MAX_VALUE));
//         state.flush();
//     }
// 
//     public <T> ISetupTest expectFlux(Flux<T> flux) {
//         return expectFlux(flux, null);
//     }
// 
//     public <T> ISetupTest expectFlux(Flux<T> flux, String unsubscriptionMarbles) {
//         return state.expectPublisher(flux, unsubscriptionMarbles);
//     }
// 
//     public ISetupSubscriptionsTest expectSubscriptions(List<SubscriptionLog> subscriptions) {
//         return state.expectSubscriptions(subscriptions);
//     }
// }
// 
// Path: src/main/java/io/reactivex/marble/MapHelper.java
// public static <V> Map<String, V> of(String k, V v) {
//     return Collections.singletonMap(k,v);
// }


// Path: src/test/java/org/reactivestreams/MarbleSchedulerTest.java
import io.reactivex.subscribers.TestSubscriber;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.ColdFlux;
import reactor.HotFlux;
import reactor.MarbleScheduler;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
import static io.reactivex.marble.MapHelper.of;

package org.reactivestreams;




public class MarbleSchedulerTest {

    private MarbleScheduler scheduler;

    @Before
    public void setupScheduler() {
        scheduler = new MarbleScheduler();
    }

    @After
    public void flushScheduler() {
        if (scheduler != null) {
            scheduler.flush();
        }
    }

    @Test
    public void should_create_a_cold_observable() {
        ColdFlux<String> source = scheduler.createColdFlux("--a---b--|", of("a", "A", "b", "B"));