// Path: src/main/java/org/reactivejournal/examples/helloworld/BytesToWordsProcessor.java
// public class BytesToWordsProcessor {
//     private static final Logger LOG = LoggerFactory.getLogger(BytesToWordsProcessor.class.getName());
//     public Flowable<String> process(Flowable<Byte> observableInput){
// 
//         PublishProcessor<String> publishProcessor = PublishProcessor.create();
// 
//         StringBuilder sb = new StringBuilder();
//         observableInput.subscribe(b->{
//                 if(b==32){ //send out a new word on a space
//                     publishProcessor.onNext(sb.toString());
//                     sb.setLength(0);
//                 }else{
//                     sb.append((char)b.byteValue());
//                 }
//             },
//             e->LOG.error("Error in BytesToWordsProcessor [{}]", e),
//             publishProcessor::onComplete
//         );
// 
//         return publishProcessor;
//     }
// }
// 
// Path: src/main/java/org/reactivejournal/examples/helloworld/HelloWorldApp_JournalPlayThrough.java
// public class HelloWorldApp_JournalPlayThrough {
//     private static final Logger LOG = LoggerFactory.getLogger(HelloWorldApp_JournalPlayThrough.class.getName());
// 
//     public final static String FILE_NAME = System.getProperty("filename", "/tmp/Demo");
//     public final static int INTERVAL_MS = 10;
//     public final static String INPUT_FILTER = "input";
//     public final static String OUTPUT_FILTER = "output";
//     public static final Flowable observableInput =
//             Flowable.fromArray(new Byte[]{72,101,108,108,111,32,87,111,114,108,100,32}).map(
//                     i->{
//                         DSUtil.sleep(INTERVAL_MS);
//                         return i;
//                     });
// 
//     public static void main(String[] args) throws IOException {
// 
//         //Create the reactiveRecorder and delete any previous content by clearing the cache
//         ReactiveJournal reactiveJournal = new ReactiveJournal(FILE_NAME);
//         reactiveJournal.clearCache();
// 
//         //Pass the input stream into the reactiveRecorder which will subscribe to it and record all events.
//         //The subscription will not be activated on a new thread which will allow this program to continue.
//         ReactiveRecorder reactiveRecorder = reactiveJournal.createReactiveRecorder();
//         reactiveRecorder.recordAsync(observableInput, INPUT_FILTER);
// 
//         BytesToWordsProcessor bytesToWords = new BytesToWordsProcessor();
// 
//         //Retrieve a stream of
//         RxJavaPlayer rxPlayer = new RxJavaPlayer(reactiveJournal);
//         PlayOptions options = new PlayOptions().filter(INPUT_FILTER).playFromNow(true);
//         ConnectableFlowable recordedObservable = rxPlayer.play(options).publish();
//         //Pass the input Byte stream into the BytesToWordsProcessor class which subscribes to the stream and returns
//         //a stream of words.
//         Flowable<String> flowableOutput = bytesToWords.process(recordedObservable);
// 
//         //Pass the output stream (of words) into the reactiveRecorder which will subscribe to it and record all events.
//         reactiveRecorder.record(flowableOutput, OUTPUT_FILTER);
//         flowableOutput.subscribe(s -> LOG.info("HelloWorldHot->" + s),
//                 throwable -> LOG.error("", throwable),
//                 ()->LOG.info("HelloWorldHot Complete"));
//         //Only start the recording now because we want to make sure that the BytesToWordsProcessor and the reactiveRecorder
//         //are both setup up to receive subscriptions.
//         recordedObservable.connect();
//         //Sometimes useful to see the recording written to a file
//         reactiveJournal.writeToFile("/tmp/Demo/demo.txt",true);
//     }
// }
// 
// Path: src/main/java/org/reactivejournal/impl/rxjava/RxJavaPlayer.java
// public class RxJavaPlayer<T> {
//     private ReactivePlayer reactivePlayer;
// 
//     public RxJavaPlayer(ReactiveJournal reactiveJournal){
//         reactivePlayer = reactiveJournal.createReactivePlayer();
//     }
// 
//     public Flowable<T> play(PlayOptions options){
//         return Flowable.fromPublisher(reactivePlayer.play(options));
//     }
// }


import io.reactivex.Flowable;
import io.reactivex.flowables.ConnectableFlowable;
import org.junit.Assert;
import org.junit.Test;
import org.reactivejournal.examples.helloworld.BytesToWordsProcessor;
import org.reactivejournal.examples.helloworld.HelloWorldApp_JournalPlayThrough;
import org.reactivejournal.impl.rxjava.RxJavaPlayer;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.stream.Collectors;
package org.reactivejournal.impl;



/**
 *  Test for RxRecorder
 */
public class ReactiveRecorderTest {
    private final String tmpDir = System.getProperty("java.io.tmpdir");

    /*
     * Test the record and play functionality as well as the writeToFile.
     *
     * Record to a journal, play from it into the BytesProcessor and then write the data to a file.
     *
     * Compare the data in the file with a file produced from a control run.
     */
    @Test
    public void recorderTest() throws IOException{
        //Flowable used to create the control run.
        Flowable<Byte> observableInput = HelloWorldApp_JournalPlayThrough.observableInput;

        //Create the rxRecorder and delete any previous content by clearing the cache
        ReactiveJournal reactiveJournal = new ReactiveJournal(tmpDir +"/playTest");
        reactiveJournal.clearCache();

        //Pass the input stream into the rxRecorder which will subscribe to it and record all events.
        //The subscription will happen on a new thread which will allow this program to continue.
        ReactiveRecorder reactiveRecorder = reactiveJournal.createReactiveRecorder();
        reactiveRecorder.recordAsync(observableInput, "input");

        BytesToWordsProcessor bytesToWords = new BytesToWordsProcessor();

// Path: src/main/java/org/reactivejournal/examples/helloworld/BytesToWordsProcessor.java
// public class BytesToWordsProcessor {
//     private static final Logger LOG = LoggerFactory.getLogger(BytesToWordsProcessor.class.getName());
//     public Flowable<String> process(Flowable<Byte> observableInput){
// 
//         PublishProcessor<String> publishProcessor = PublishProcessor.create();
// 
//         StringBuilder sb = new StringBuilder();
//         observableInput.subscribe(b->{
//                 if(b==32){ //send out a new word on a space
//                     publishProcessor.onNext(sb.toString());
//                     sb.setLength(0);
//                 }else{
//                     sb.append((char)b.byteValue());
//                 }
//             },
//             e->LOG.error("Error in BytesToWordsProcessor [{}]", e),
//             publishProcessor::onComplete
//         );
// 
//         return publishProcessor;
//     }
// }
// 
// Path: src/main/java/org/reactivejournal/examples/helloworld/HelloWorldApp_JournalPlayThrough.java
// public class HelloWorldApp_JournalPlayThrough {
//     private static final Logger LOG = LoggerFactory.getLogger(HelloWorldApp_JournalPlayThrough.class.getName());
// 
//     public final static String FILE_NAME = System.getProperty("filename", "/tmp/Demo");
//     public final static int INTERVAL_MS = 10;
//     public final static String INPUT_FILTER = "input";
//     public final static String OUTPUT_FILTER = "output";
//     public static final Flowable observableInput =
//             Flowable.fromArray(new Byte[]{72,101,108,108,111,32,87,111,114,108,100,32}).map(
//                     i->{
//                         DSUtil.sleep(INTERVAL_MS);
//                         return i;
//                     });
// 
//     public static void main(String[] args) throws IOException {
// 
//         //Create the reactiveRecorder and delete any previous content by clearing the cache
//         ReactiveJournal reactiveJournal = new ReactiveJournal(FILE_NAME);
//         reactiveJournal.clearCache();
// 
//         //Pass the input stream into the reactiveRecorder which will subscribe to it and record all events.
//         //The subscription will not be activated on a new thread which will allow this program to continue.
//         ReactiveRecorder reactiveRecorder = reactiveJournal.createReactiveRecorder();
//         reactiveRecorder.recordAsync(observableInput, INPUT_FILTER);
// 
//         BytesToWordsProcessor bytesToWords = new BytesToWordsProcessor();
// 
//         //Retrieve a stream of
//         RxJavaPlayer rxPlayer = new RxJavaPlayer(reactiveJournal);
//         PlayOptions options = new PlayOptions().filter(INPUT_FILTER).playFromNow(true);
//         ConnectableFlowable recordedObservable = rxPlayer.play(options).publish();
//         //Pass the input Byte stream into the BytesToWordsProcessor class which subscribes to the stream and returns
//         //a stream of words.
//         Flowable<String> flowableOutput = bytesToWords.process(recordedObservable);
// 
//         //Pass the output stream (of words) into the reactiveRecorder which will subscribe to it and record all events.
//         reactiveRecorder.record(flowableOutput, OUTPUT_FILTER);
//         flowableOutput.subscribe(s -> LOG.info("HelloWorldHot->" + s),
//                 throwable -> LOG.error("", throwable),
//                 ()->LOG.info("HelloWorldHot Complete"));
//         //Only start the recording now because we want to make sure that the BytesToWordsProcessor and the reactiveRecorder
//         //are both setup up to receive subscriptions.
//         recordedObservable.connect();
//         //Sometimes useful to see the recording written to a file
//         reactiveJournal.writeToFile("/tmp/Demo/demo.txt",true);
//     }
// }
// 
// Path: src/main/java/org/reactivejournal/impl/rxjava/RxJavaPlayer.java
// public class RxJavaPlayer<T> {
//     private ReactivePlayer reactivePlayer;
// 
//     public RxJavaPlayer(ReactiveJournal reactiveJournal){
//         reactivePlayer = reactiveJournal.createReactivePlayer();
//     }
// 
//     public Flowable<T> play(PlayOptions options){
//         return Flowable.fromPublisher(reactivePlayer.play(options));
//     }
// }


// Path: src/test/java/org/reactivejournal/impl/ReactiveRecorderTest.java
import io.reactivex.Flowable;
import io.reactivex.flowables.ConnectableFlowable;
import org.junit.Assert;
import org.junit.Test;
import org.reactivejournal.examples.helloworld.BytesToWordsProcessor;
import org.reactivejournal.examples.helloworld.HelloWorldApp_JournalPlayThrough;
import org.reactivejournal.impl.rxjava.RxJavaPlayer;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.stream.Collectors;

package org.reactivejournal.impl;



/**
 *  Test for RxRecorder
 */
public class ReactiveRecorderTest {
    private final String tmpDir = System.getProperty("java.io.tmpdir");

    /*
     * Test the record and play functionality as well as the writeToFile.
     *
     * Record to a journal, play from it into the BytesProcessor and then write the data to a file.
     *
     * Compare the data in the file with a file produced from a control run.
     */
    @Test
    public void recorderTest() throws IOException{
        //Flowable used to create the control run.
        Flowable<Byte> observableInput = HelloWorldApp_JournalPlayThrough.observableInput;

        //Create the rxRecorder and delete any previous content by clearing the cache
        ReactiveJournal reactiveJournal = new ReactiveJournal(tmpDir +"/playTest");
        reactiveJournal.clearCache();

        //Pass the input stream into the rxRecorder which will subscribe to it and record all events.
        //The subscription will happen on a new thread which will allow this program to continue.
        ReactiveRecorder reactiveRecorder = reactiveJournal.createReactiveRecorder();
        reactiveRecorder.recordAsync(observableInput, "input");

        BytesToWordsProcessor bytesToWords = new BytesToWordsProcessor();

        RxJavaPlayer rxPlayer = new RxJavaPlayer(reactiveJournal);