From 071f6466aea72c037077874fe8f33c7b7be27d29 Mon Sep 17 00:00:00 2001
From: Albert Shift <a@albertshift.com>
Date: Tue, 23 Jun 2015 08:34:20 -0700
Subject: [PATCH] use Scala Option and Stream in Scala future results

---
 .../operation/AbstractOptionalOperation.java  | 46 ++++++++++++++-----
 .../operation/AbstractStreamOperation.java    | 37 +++++++++++----
 2 files changed, 62 insertions(+), 21 deletions(-)

diff --git a/src/main/java/com/noorq/casser/core/operation/AbstractOptionalOperation.java b/src/main/java/com/noorq/casser/core/operation/AbstractOptionalOperation.java
index 103a126..442577c 100644
--- a/src/main/java/com/noorq/casser/core/operation/AbstractOptionalOperation.java
+++ b/src/main/java/com/noorq/casser/core/operation/AbstractOptionalOperation.java
@@ -17,6 +17,9 @@ package com.noorq.casser.core.operation;
 
 import java.util.Optional;
 
+import scala.None;
+import scala.Option;
+import scala.Some;
 import scala.concurrent.Future;
 
 import com.datastax.driver.core.PreparedStatement;
@@ -82,25 +85,46 @@ public abstract class AbstractOptionalOperation<E, O extends AbstractOptionalOpe
 		
 		return future;
 	}
-	
-	public Future<Optional<E>> future() {
-		return Scala.asFuture(async());
+
+	public ListenableFuture<Option<E>> asyncForScala() {
+		
+		ResultSetFuture resultSetFuture = sessionOps.executeAsync(options(buildStatement()), showValues);
+
+		ListenableFuture<Option<E>> future = Futures.transform(resultSetFuture, new Function<ResultSet, Option<E>>() {
+
+			@Override
+			public Option<E> apply(ResultSet resultSet) {
+				Optional<E> optional = transform(resultSet);
+				if (optional.isPresent()) {
+					return new Some<E>(optional.get());
+				}
+				else {
+					return Option.empty();
+				}
+			}
+
+		}, sessionOps.getExecutor());
+		
+		return future;
+	}
+	public Future<Option<E>> future() {
+		return Scala.asFuture(asyncForScala());
 	}
 
-	public <A> Future<Fun.Tuple2<Optional<E>, A>> future(A a) {
-		return Scala.asFuture(async(), a);
+	public <A> Future<Fun.Tuple2<Option<E>, A>> future(A a) {
+		return Scala.asFuture(asyncForScala(), a);
 	}
 
-	public <A, B> Future<Fun.Tuple3<Optional<E>, A, B>> future(A a, B b) {
-		return Scala.asFuture(async(), a, b);
+	public <A, B> Future<Fun.Tuple3<Option<E>, A, B>> future(A a, B b) {
+		return Scala.asFuture(asyncForScala(), a, b);
 	}
 
-	public <A, B, C> Future<Fun.Tuple4<Optional<E>, A, B, C>> future(A a, B b, C c) {
-		return Scala.asFuture(async(), a, b, c);
+	public <A, B, C> Future<Fun.Tuple4<Option<E>, A, B, C>> future(A a, B b, C c) {
+		return Scala.asFuture(asyncForScala(), a, b, c);
 	}
 
-	public <A, B, C, D> Future<Fun.Tuple5<Optional<E>, A, B, C, D>> future(A a, B b, C c, D d) {
-		return Scala.asFuture(async(), a, b, c, d);
+	public <A, B, C, D> Future<Fun.Tuple5<Option<E>, A, B, C, D>> future(A a, B b, C c, D d) {
+		return Scala.asFuture(asyncForScala(), a, b, c, d);
 	}
 	
 }
diff --git a/src/main/java/com/noorq/casser/core/operation/AbstractStreamOperation.java b/src/main/java/com/noorq/casser/core/operation/AbstractStreamOperation.java
index 5516c57..ec5fbbc 100644
--- a/src/main/java/com/noorq/casser/core/operation/AbstractStreamOperation.java
+++ b/src/main/java/com/noorq/casser/core/operation/AbstractStreamOperation.java
@@ -83,24 +83,41 @@ public abstract class AbstractStreamOperation<E, O extends AbstractStreamOperati
 		return future;
 	}
 	
-	public Future<Stream<E>> future() {
-		return Scala.asFuture(async());
+	public ListenableFuture<scala.collection.immutable.Stream<E>> asyncForScala() {
+		
+		ResultSetFuture resultSetFuture = sessionOps.executeAsync(options(buildStatement()), showValues);
+
+		ListenableFuture<scala.collection.immutable.Stream<E>> future = Futures.transform(resultSetFuture, new Function<ResultSet, scala.collection.immutable.Stream<E>>() {
+
+			@Override
+			public scala.collection.immutable.Stream<E> apply(ResultSet resultSet) {
+				Stream<E> stream = transform(resultSet);
+				return scala.collection.JavaConversions.asScalaIterator(stream.iterator()).toStream();
+			}
+
+		}, sessionOps.getExecutor());
+		
+		return future;
+	}
+	
+	public Future<scala.collection.immutable.Stream<E>> future() {
+		return Scala.asFuture(asyncForScala());
 	}
 
-	public <A> Future<Fun.Tuple2<Stream<E>, A>> future(A a) {
-		return Scala.asFuture(async(), a);
+	public <A> Future<Fun.Tuple2<scala.collection.immutable.Stream<E>, A>> future(A a) {
+		return Scala.asFuture(asyncForScala(), a);
 	}
 
-	public <A, B> Future<Fun.Tuple3<Stream<E>, A, B>> future(A a, B b) {
-		return Scala.asFuture(async(), a, b);
+	public <A, B> Future<Fun.Tuple3<scala.collection.immutable.Stream<E>, A, B>> future(A a, B b) {
+		return Scala.asFuture(asyncForScala(), a, b);
 	}
 
-	public <A, B, C> Future<Fun.Tuple4<Stream<E>, A, B, C>> future(A a, B b, C c) {
-		return Scala.asFuture(async(), a, b, c);
+	public <A, B, C> Future<Fun.Tuple4<scala.collection.immutable.Stream<E>, A, B, C>> future(A a, B b, C c) {
+		return Scala.asFuture(asyncForScala(), a, b, c);
 	}
 
-	public <A, B, C, D> Future<Fun.Tuple5<Stream<E>, A, B, C, D>> future(A a, B b, C c, D d) {
-		return Scala.asFuture(async(), a, b, c, d);
+	public <A, B, C, D> Future<Fun.Tuple5<scala.collection.immutable.Stream<E>, A, B, C, D>> future(A a, B b, C c, D d) {
+		return Scala.asFuture(asyncForScala(), a, b, c, d);
 	}
 	
 }