add ReactiveLoop

Author: sohan09

elasta-composer/src/main/java/elasta/composer/utils/ReactiveLoop.java

@@ -0,0 +1,71 @@
+package elasta.composer.utils;
+
+import elasta.commons.Utils;
+import elasta.core.intfs.CallableUnckd;
+import io.reactivex.Observable;
+import io.reactivex.Single;
+import io.reactivex.functions.Function;
+import io.reactivex.functions.Predicate;
+import io.reactivex.subjects.PublishSubject;
+
+import java.util.Objects;
+
+/**
+ * Created by sohan on 2017-09-02.
+ */
+@FunctionalInterface
+public interface ReactiveLoop<T> {
+    Observable<T> start();
+
+    static <T> ReactiveLoop<T> create(CreateParams<T> params) {
+        return () -> {
+            try {
+                final PublishSubject<T> subject = PublishSubject.create();
+
+                params.initialState.call()
+                    .doOnEvent((t, throwable) -> handle(t, throwable, subject))
+                    .subscribe();
+
+                return subject
+                    .doOnNext(t -> {
+
+                        if (Utils.not(params.hasNext.test(t))) {
+                            subject.onComplete();
+                            return;
+                        }
+
+                        params.nextState.apply(t)
+                            .doOnEvent((newState, throwable) -> handle(newState, throwable, subject))
+                            .subscribe();
+
+                    });
+            } catch (Throwable ex) {
+                return Observable.error(ex);
+            }
+        };
+    }
+
+    static <T> void handle(T state, Throwable throwable, PublishSubject<T> subject) {
+        if (throwable != null) {
+            subject.onError(throwable);
+        } else {
+            subject.onNext(state);
+        }
+    }
+
+
+    final class CreateParams<T> {
+        private final CallableUnckd<Single<T>> initialState;
+        private final Predicate<T> hasNext;
+        private final Function<T, Single<T>> nextState;
+
+        public CreateParams(CallableUnckd<Single<T>> initialState, Predicate<T> hasNext, Function<T, Single<T>> nextState) {
+            Objects.requireNonNull(initialState);
+            Objects.requireNonNull(hasNext);
+            Objects.requireNonNull(nextState);
+            this.initialState = initialState;
+            this.hasNext = hasNext;
+            this.nextState = nextState;
+        }
+    }
+}

proxy-config.json

@@ -0,0 +1,8 @@
+{
+  "targetScheme": "http",
+  "targetHost": "localhost",
+  "targetPort": 52054,
+  "targetPrefix": "",
+  "uri": "/*",
+  "port": 153
+}

settings.gradle

@@ -16,4 +16,5 @@ include 'elasta-eventbus'
 include 'elasta-authorization'
 include 'tracker'
 include 'tracker-server'
+include 'simple-proxy'
 

simple-proxy/build.gradle

@@ -0,0 +1,55 @@
+group 'com.github.codefacts'
+version '1.0-SNAPSHOT'
+
+buildscript {
+    repositories {
+        jcenter()
+    }
+    dependencies {
+        classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.1'
+    }
+}
+
+apply plugin: 'com.github.johnrengelman.shadow'
+apply plugin: 'java'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+
+    compile project(':commons')
+    compile project(':elasta-core')
+    compile project(':vertx-utils')
+    compile project(':elasta-module')
+    compile project(':elasta-eventbus')
+
+    compile group: 'io.vertx', name: 'vertx-core', version: '3.4.2'
+    compile group: 'io.vertx', name: 'vertx-web', version: '3.4.2'
+
+    compile group: 'io.reactivex.rxjava2', name: 'rxjava', version: '2.1.2'
+
+    compile group: 'com.koushikdutta.async', name: 'androidasync', version: '2.1.7'
+
+    compile group: 'org.apache.logging.log4j', name: 'log4j-slf4j-impl', version: '2.8.2'
+    compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.8.2'
+    compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.8.2'
+
+    compileOnly "org.projectlombok:lombok:1.16.14"
+
+    testCompile group: 'junit', name: 'junit', version: '4.11'
+    testCompileOnly "org.projectlombok:lombok:1.16.14"
+
+    testCompile group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.8.1'
+
+    compile group: 'org.apache.commons', name: 'commons-csv', version: '1.4'
+}
+
+jar {
+    manifest {
+        attributes 'Main-Class': 'simple.proxy.Proxy'
+    }
+}

simple-proxy/src/main/java/simple/proxy/InvalidConfigurationException.java

@@ -0,0 +1,12 @@
+package simple.proxy;
+
+import java.io.IOException;
+
+/**
+ * Created by sohan on 2017-08-05.
+ */
+final public class InvalidConfigurationException extends RuntimeException {
+    public InvalidConfigurationException(IOException e) {
+        super(e);
+    }
+}

simple-proxy/src/main/java/simple/proxy/Nana.java

@@ -0,0 +1,43 @@
+package simple.proxy;
+
+import io.reactivex.Observable;
+
+import java.rmi.UnexpectedException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created by sohan on 2017-08-08.
+ */
+public interface Nana {
+    static void main(String[] args) throws InterruptedException {
+        Observable
+            .create(e -> {
+                try {
+
+                    System.out.println("### onNext: value and isDisposed" + e.isDisposed());
+                    e.onNext("value");
+
+//                    Observable.timer(5, TimeUnit.SECONDS)
+//                        .subscribe(aLong -> {
+//                            System.out.println("## 22: " + e.isDisposed());
+//                            e.onNext("22");
+//                            if (!e.isDisposed()) {
+//                                e.onError(new UnexpectedException("error"));
+//                            }
+//                        });
+
+                } catch (Exception ex) {
+                    System.out.println(ex.toString());
+//                    ex.printStackTrace();
+                }
+            })
+            .doOnNext(o -> System.out.println("#### got11: " + o))
+            .flatMap(o -> Observable.timer(200, TimeUnit.MILLISECONDS).map(aLong -> o)
+                .doOnNext(o1 -> System.out.println("###from timer")))
+            .doOnNext(o -> System.out.println("#### got22: " + o))
+            .timeout(0, TimeUnit.MILLISECONDS)
+            .subscribe(o -> {
+            }, throwable -> System.out.println("##### " + throwable));
+        Thread.sleep(1000000);
+    }
+}