Merge pull request #68 from aol/cyclops-integration

v0.99.4 release

build.gradle

@@ -28,13 +28,15 @@ apply plugin: 'jacoco'
 apply from: custom('jacoco-version')
 
 sourceCompatibility = 1.8
-version = '0.99.3'
+version = '0.99.4'
 jar {
     manifest {
         attributes 'Implementation-Title': 'Simple React', 'Implementation-Version': version
     }
 }
-
+shadowJar {
+	mergeServiceFiles()
+  }
 apply from: custom('jacoco-config')
 apply from: custom('jacoco-excludes')
 
@@ -54,14 +56,14 @@ dependencies {
 	compile 'org.jooq:jool:0.9.7'
 	compile 'org.pcollections:pcollections:2.1.2'
 	compile 'com.nurkiewicz.asyncretry:asyncretry:0.0.7'
-	compile 'uk.co.real-logic:Agrona:0.4.2'
+	compile 'uk.co.real-logic:Agrona:0.4.5'
 	compile group: 'org.slf4j', name: 'jcl-over-slf4j', version:'1.6.1'
 	compile 'org.reactivestreams:reactive-streams:1.0.0'
-	compile 'com.aol.cyclops:cyclops-sequence-api:6.0.2'
-	compile 'com.aol.cyclops:cyclops-streams:6.0.2'
-	compile 'com.aol.cyclops:cyclops-monad-api:6.0.2'
-	compile 'com.aol.cyclops:cyclops-functions:6.0.2'
-	compile 'com.aol.cyclops:cyclops-for-comprehensions:6.0.2'
+	compile 'com.aol.cyclops:cyclops-sequence-api:6.0.3'
+	compile 'com.aol.cyclops:cyclops-streams:6.0.3'
+	compile 'com.aol.cyclops:cyclops-monad-api:6.0.3'
+	compile 'com.aol.cyclops:cyclops-functions:6.0.3'
+	compile 'com.aol.cyclops:cyclops-for-comprehensions:6.0.3'
 	testCompile 'org.reactivestreams:reactive-streams-tck:1.0.0'
 	testCompile group: 'junit', name: 'junit', version: '4.12'
 	testCompile group: 'org.hamcrest', name: 'hamcrest-all', version:'1.3'

src/main/java/com/aol/simple/react/stream/traits/LazyFutureStream.java

@@ -1565,7 +1565,7 @@ default Tuple2<LazyFutureStream<U>, LazyFutureStream<U>> partitionFutureStream(
 	@Override
 	default LazyFutureStream<U> slice(long from, long to) {
 
-		return fromStream(fromStream(SequenceM.fromStream(toQueue().stream(getSubscription()))
+		return fromStream(SequenceM.fromStream(toQueue().stream(getSubscription())
 				.slice(from, to)));
 	}
 
@@ -1634,7 +1634,7 @@ default <T, R> LazyFutureStream<R> zip(Seq<T> other,
 	@Override
 	default <T> LazyFutureStream<T> scanLeft(T seed,
 			BiFunction<T, ? super U, T> function) {
-		return fromStream(fromStream(SequenceM.fromStream(toQueue().stream(getSubscription())).scanLeft(seed, function)));
+		return fromStream(SequenceM.fromStream(toQueue().stream(getSubscription())).scanLeft(seed, function));
 
 	}
 
@@ -1649,7 +1649,7 @@ default <T> LazyFutureStream<T> scanLeft(T seed,
 	@Override
 	default <R> LazyFutureStream<R> scanRight(R seed,
 			BiFunction<? super U, R, R> function) {
-		return fromStream(fromStream(SequenceM.fromStream(toQueue().stream(getSubscription())).scanRight(seed, function)));
+		return fromStream(SequenceM.fromStream(toQueue().stream(getSubscription())).scanRight(seed, function));
 
 	}
 

src/main/java/com/aol/simple/react/stream/traits/future/operators/OperationsOnFutures.java

@@ -403,7 +403,7 @@ default <S, U> LazyFutureStream<Tuple3<T, S, U>> zip3Lfs(
 	 *</pre>
 	 *
 	 */
-	default <T2, T3, T4> SequenceM<Tuple4<T, T2, T3, T4>> zip4(
+	default <T2, T3, T4> LazyFutureStream<Tuple4<T, T2, T3, T4>> zip4(
 			Stream<T2> second, Stream<T3> third, Stream<T4> fourth) {
 		return (LazyFutureStream) fromStreamOfFutures((Stream) this
 				.getLastActive()
@@ -427,7 +427,7 @@ default <T2, T3, T4> SequenceM<Tuple4<T, T2, T3, T4>> zip4(
 	 *}
 	 *</pre>
 	 */
-	default <T2, T3, T4> SequenceM<Tuple4<T, T2, T3, T4>> zip4Lfs(
+	default <T2, T3, T4> LazyFutureStream<Tuple4<T, T2, T3, T4>> zip4Lfs(
 			LazyFutureStream<T2> second, LazyFutureStream<T3> third, LazyFutureStream<T4> fourth) {
 		return (LazyFutureStream) fromStreamOfFutures((Stream) this
 				.getLastActive()
@@ -675,7 +675,7 @@ default LazyFutureStream<T> intersperse(CompletableFuture<T> value) {
 	 * 
 	 * @see org.jooq.lambda.Seq#shuffle()
 	 */
-	default SequenceM<T> shuffle() {
+	default LazyFutureStream<T> shuffle() {
 		return fromStreamOfFutures(this.getLastActive().injectFuturesSeq()
 				.shuffle());
 	}

src/test/java/com/aol/simple/react/base/BaseSeqTest.java

@@ -380,6 +380,7 @@ public void limitWhileTest(){
 
     @Test
     public void testScanLeftStringConcat() {
+
         assertThat(of("a", "b", "c").scanLeft("", String::concat).toList().size(),
         		is(4));
     }

src/test/java/com/aol/simple/react/examples/FilesExamplesTest.java

@@ -0,0 +1,29 @@
+package com.aol.simple.react.examples;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.stream.Stream;
+
+import org.jooq.lambda.Unchecked;
+import org.junit.Test;
+
+import com.aol.simple.react.stream.lazy.LazyReact;
+
+public class FilesExamplesTest {
+
+	@Test
+	public void test() throws IOException {
+
+		LazyReact react = new LazyReact(100,110);
+		react.from(Files.walk(Paths.get(".")))
+		//	 .map(Unchecked.function(Files::readAllBytes))
+			.map(d->{ throw new RuntimeException("hello");})
+			.map(Object::toString)
+			 .recover(e->"hello world")
+			// .flatMap(s->Stream.of(s.split("\n")))
+			 .forEach(System.out::println);
+
+	}
+
+}

src/test/java/com/aol/simple/react/lazy/AutoMemoizationTest.java

@@ -17,6 +17,7 @@ public class AutoMemoizationTest {
    AtomicInteger called = new AtomicInteger(0);
 	@Test
 	public void autoMemoize(){
+
 		called.set(0);
 		Map cache = new ConcurrentHashMap<>();
 		LazyReact react = new LazyReact().autoMemoizeOn((key,fn)-> cache.computeIfAbsent(key,fn));
@@ -31,6 +32,8 @@ public void autoMemoize(){
 		System.out.println(result);
 		assertThat(called.get(),equalTo(1));
 		assertThat(result.size(),equalTo(4));
+
+
 	  }
 	@Test
 	public void autoMemoizeOff(){

src/test/java/com/aol/simple/react/lazy/futures/OperationsOnFuturesTest.java

@@ -467,10 +467,14 @@ public void testToSet() {
 	    } 
 	    @Test
 	    public void testReduce() {
-	       CompletableFuture<Integer> sum = of(1, 2, 3).actOnFutures()
-	        							.reduce((cf1,cf2)-> cf1.thenCombine(cf2, (a,b)->a+b)).get();
+
+	       CompletableFuture<Integer> sum = LazyFutureStream.of(1, 2, 3)
+	    		   											.actOnFutures()
+	    		   											.reduce((cf1,cf2)-> cf1.thenCombine(cf2, (a,b)->a+b)).get();
 
 	        assertThat(sum.join(),equalTo(6));
+
+
 	    }
 	    @Test
 	    public void testReduceIdentity() {

src/test/java/com/aol/simple/react/lazy/sequenceM/WindowingTest.java

@@ -31,55 +31,58 @@ public void setup(){
 
 	@Test
 	public void windowWhile(){
-		assertThat(SequenceM.of(1,2,3,4,5,6)
+		assertThat(LazyFutureStream.of(1,2,3,4,5,6)
 				.windowWhile(i->i%3!=0)
 				.toList().size(),equalTo(2));
-		assertThat(SequenceM.of(1,2,3,4,5,6)
+		assertThat(LazyFutureStream.of(1,2,3,4,5,6)
 				.windowWhile(i->i%3!=0)
 				.toList().get(0).sequenceM().toList(),equalTo(Arrays.asList(1,2,3)));
 	}
 	@Test
 	public void windowUntil(){
-		assertThat(SequenceM.of(1,2,3,4,5,6)
+
+
+
+		assertThat(LazyFutureStream.of(1,2,3,4,5,6)
 				.windowUntil(i->i%3==0)
 				.toList().size(),equalTo(2));
-		assertThat(SequenceM.of(1,2,3,4,5,6)
+		assertThat(LazyFutureStream.of(1,2,3,4,5,6)
 				.windowUntil(i->i%3==0)
 				.toList().get(0).sequenceM().toList(),equalTo(Arrays.asList(1,2,3)));
 	}
 	@Test
 	public void windowUntilEmpty(){
-		assertThat(SequenceM.<Integer>of()
+		assertThat(LazyFutureStream.<Integer>of()
 				.windowUntil(i->i%3==0)
 				.toList().size(),equalTo(0));
 	}
 	@Test
 	public void windowStatefullyWhile(){
 
-		assertThat(SequenceM.of(1,2,3,4,5,6)
+		assertThat(LazyFutureStream.of(1,2,3,4,5,6)
 				.windowStatefullyWhile((s,i)->s.sequenceM().toList().contains(4) ? true : false)
 				.toList().size(),equalTo(5));
 
 	}
 	@Test
 	public void windowStatefullyWhileEmpty(){
 
-		assertThat(SequenceM.of()
+		assertThat(LazyFutureStream.of()
 				.windowStatefullyWhile((s,i)->s.sequenceM().toList().contains(4) ? true : false)
 				.toList().size(),equalTo(0));
 
 	}
 	@Test
 	public void sliding() {
-		List<List<Integer>> list = SequenceM.of(1, 2, 3, 4, 5, 6).sliding(2).collect(Collectors.toList());
+		List<List<Integer>> list = LazyFutureStream.of(1, 2, 3, 4, 5, 6).sliding(2).collect(Collectors.toList());
 
 		assertThat(list.get(0), hasItems(1, 2));
 		assertThat(list.get(1), hasItems(2, 3));
 	}
 
 	@Test
 	public void slidingIncrement() {
-		List<List<Integer>> list = SequenceM.of(1, 2, 3, 4, 5, 6).sliding(3, 2).collect(Collectors.toList());
+		List<List<Integer>> list = LazyFutureStream.of(1, 2, 3, 4, 5, 6).sliding(3, 2).collect(Collectors.toList());
 
 		assertThat(list.get(0), hasItems(1, 2, 3));
 		assertThat(list.get(1), hasItems(3, 4, 5));
@@ -88,7 +91,7 @@ public void slidingIncrement() {
 	@Test
 	public void grouped() {
 
-		List<List<Integer>> list = SequenceM.of(1, 2, 3, 4, 5, 6).grouped(3).collect(Collectors.toList());
+		List<List<Integer>> list = LazyFutureStream.of(1, 2, 3, 4, 5, 6).grouped(3).collect(Collectors.toList());
 
 		assertThat(list.get(0), hasItems(1, 2, 3));
 		assertThat(list.get(1), hasItems(4, 5, 6));
@@ -99,15 +102,15 @@ public void grouped() {
 	public void sliding2() {
 
 
-		List<List<Integer>> sliding = SequenceM.of(1, 2, 3, 4, 5).sliding(2).toList();
+		List<List<Integer>> sliding = LazyFutureStream.of(1, 2, 3, 4, 5).sliding(2).toList();
 
 		assertThat(sliding, contains(asList(1, 2), asList(2, 3), asList(3, 4), asList(4, 5)));
 	}
 
 	@Test
 	public void slidingOverlap() {
 
-		List<List<Integer>> sliding = SequenceM.of(1, 2, 3, 4, 5).sliding(3,2).toList();
+		List<List<Integer>> sliding = LazyFutureStream.of(1, 2, 3, 4, 5).sliding(3,2).toList();
 
 		assertThat(sliding, contains(asList(1, 2, 3), asList(3, 4, 5)));
 	}
@@ -116,14 +119,14 @@ public void slidingOverlap() {
 	public void slidingEmpty() {
 
 
-		assertThat(SequenceM.of().sliding(1).toList().size(),equalTo(0));
+		assertThat(LazyFutureStream.of().sliding(1).toList().size(),equalTo(0));
 	}
 
 	@Test
 	public void slidingWithSmallWindowAtEnd() {
 
 
-		List<List<Integer>> sliding = SequenceM.of(1, 2, 3, 4, 5).sliding(2,2).toList();
+		List<List<Integer>> sliding = LazyFutureStream.of(1, 2, 3, 4, 5).sliding(2,2).toList();
 
 		assertThat(sliding, contains(asList(1, 2), asList(3, 4), asList(5)));
 	}
@@ -147,25 +150,25 @@ public void grouped0() throws Exception {
 
 	@Test
 	public void groupedShorter() throws Exception {
-		final Streamable<Integer> fixed = Streamable.fromStream(of(5, 7, 9));
-		assertThat(fixed.sequenceM().grouped(4).get(0).v1,equalTo(Arrays.asList(5,7,9)));
-		assertThat(fixed.sequenceM().grouped(4).count(),equalTo(1l));
+
+		assertThat(of(5, 7, 9).grouped(4).get(0).v1,equalTo(Arrays.asList(5,7,9)));
+		assertThat(of(5, 7, 9).grouped(4).count(),equalTo(1l));
 
 
 	}
 
 	@Test
 	public void groupedEqualSize() throws Exception {
-		final Streamable<Integer> fixed = Streamable.fromStream(of(5, 7, 9));
-		assertThat(fixed.sequenceM().grouped(3).elementAt(0).get(),equalTo(Arrays.asList(5,7,9)));
-		assertThat(fixed.sequenceM().grouped(3).count(),equalTo(1l));
+
+		assertThat(of(5, 7, 9).grouped(3).elementAt(0).get(),equalTo(Arrays.asList(5,7,9)));
+		assertThat(of(5, 7, 9).grouped(3).count(),equalTo(1l));
 	}
 
 	@Test
 	public void multipleGrouped() throws Exception {
 		final Streamable<Integer> fixed = Streamable.fromStream(of(5, 7, 9,10));
-		assertThat(fixed.sequenceM().grouped(3).elementAt(0).get(),equalTo(Arrays.asList(5,7,9)));
-		assertThat(fixed.sequenceM().grouped(3).count(),equalTo(2l));
+		assertThat(of(5, 7, 9,10).grouped(3).elementAt(0).get(),equalTo(Arrays.asList(5,7,9)));
+		assertThat(of(5, 7, 9,10).grouped(3).count(),equalTo(2l));
 
 	}
 

src/test/java/com/aol/simple/react/lazy/sequenceM/hotstream/HotStreamTest.java

@@ -24,6 +24,7 @@ public class HotStreamTest {
 	public void hotStream() throws InterruptedException{
 		value= null;
 		CountDownLatch latch = new CountDownLatch(1);
+
 		LazyFutureStream.of(1,2,3)
 				.peek(v->value=v)
 				.peek(v->latch.countDown())

src/test/java/com/aol/simple/react/lazy/sequenceM/reactivestreams/ReactiveStreamsTest.java

@@ -14,10 +14,14 @@ public class ReactiveStreamsTest {
 
 	@Test
 	public void publishAndSubscribe(){
+
 		CyclopsSubscriber<Integer> sub = SequenceM.subscriber();
+
 		SequenceM.of(1,2,3).subscribe(sub);
+
 		assertThat(sub.sequenceM().toList(),equalTo(
 				Arrays.asList(1,2,3)));
+
 	}
 	@Test
 	public void publishAndSubscribeEmpty(){