Feature/stream docs (#742)

* adding documentation for the new streaming methods.

* Adding tests for the misisng query methods

* Removing the implicit guava adapter in favour of promise interface inference.

* Removing guava adapter references.

* Fixing streams test.

* Adding more tests for the newly added fun bits.

* Renaming and repurposing test

build.sbt

@@ -315,6 +315,7 @@ lazy val phantomSbtPlugin = (project in file("phantom-sbt"))
     sbtPlugin := true,
     publishArtifact := !Publishing.publishingToMaven && { scalaVersion.value.startsWith("2.10") },
     libraryDependencies ++= Seq(
+      "com.datastax.cassandra" % "cassandra-driver-core" % Versions.datastax,
       "org.cassandraunit" % "cassandra-unit"  % Versions.cassandraUnit excludeAll (
         ExclusionRule("org.slf4j", "slf4j-log4j12"),
         ExclusionRule("org.slf4j", "slf4j-jdk14")

build/run_tests.sh

@@ -1,5 +1,5 @@
 #!/usr/bin/env bash
-if [ "${TRAVIS_SCALA_VERSION}" == "2.12.2" ] && [ "${TRAVIS_JDK_VERSION}" == "oraclejdk8" ];
+if [ "${TRAVIS_SCALA_VERSION}" == "2.12.3" ] && [ "${TRAVIS_JDK_VERSION}" == "oraclejdk8" ];
 then
     echo "Running tests with coverage and report submission"
     sbt ++$TRAVIS_SCALA_VERSION coverage test coverageReport coverageAggregate coveralls

phantom-dsl/src/main/scala/com/outworkers/phantom/ScalaQueryContext.scala

@@ -51,8 +51,7 @@ object ScalaPromiseInterface extends PromiseInterface[Promise, Future] {
 
 class ScalaQueryContext extends QueryContext[Promise, Future, Duration](10.seconds)(
   ScalaFutureImplicits.monadInstance,
-  ScalaPromiseInterface,
-  ScalaGuavaAdapter
+  ScalaPromiseInterface
 ) {
 
   override def await[T](f: Future[T], timeout: Duration): T = Await.result(f, timeout)

phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/DeleteQuery.scala

@@ -92,7 +92,6 @@ case class DeleteQuery[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     monad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedBlock[Rev]] = {
     val flatten = new PreparedFlattener(qb)

phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/InsertQuery.scala

@@ -129,7 +129,6 @@ case class InsertQuery[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     fMonad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedBlock[Rev]] = {
     val flatten = new PreparedFlattener(qb)
@@ -245,7 +244,6 @@ class InsertJsonQuery[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     fMonad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedBlock[Rev]] = {
     val flatten = new PreparedFlattener(qb)

phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/UpdateQuery.scala

@@ -76,7 +76,6 @@ case class UpdateQuery[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     fMonad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedBlock[Rev]] = {
     val flatten = new PreparedFlattener(qb)
@@ -391,7 +390,6 @@ sealed case class ConditionalQuery[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     fMonad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedBlock[Rev]] = {
     val flatten = new PreparedFlattener(qb)

phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/prepared/PreparedBuilder.scala

@@ -69,11 +69,10 @@ class PreparedFlattener(qb: CQLQuery)(
   def async[P[_], F[_]]()(
     implicit executor: ExecutionContextExecutor,
     monad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedStatement] = {
     new ExactlyOncePromise[P, F, PreparedStatement](
-      adapter.fromGuava(session.prepareAsync(qb.queryString))
+      interface.adapter.fromGuava(session.prepareAsync(qb.queryString))
     ).future
   }
 }

phantom-dsl/src/main/scala/com/outworkers/phantom/dsl/package.scala

@@ -28,7 +28,7 @@ package object dsl extends ScalaQueryContext with DefaultImports {
 
   implicit class ExecuteQueries[M[X] <: TraversableOnce[X]](val qc: QueryCollection[M]) extends AnyVal {
     def executable(): ExecutableStatements[Future, M] = {
-      new ExecutableStatements[Future, M](qc)
+      new ExecutableStatements[Future, M](qc)(futureMonad, promiseInterface.adapter)
     }
 
     def future()(implicit session: Session,
@@ -47,7 +47,7 @@ package object dsl extends ScalaQueryContext with DefaultImports {
   def cql(
     str: CQLQuery,
     options: QueryOptions
-  ): QueryInterface[Future] = new QueryInterface[Future]() {
+  ): QueryInterface[Future] = new QueryInterface[Future]()(promiseInterface.adapter) {
     override def executableQuery: ExecutableCqlQuery = ExecutableCqlQuery(str, options)
   }
 

phantom-dsl/src/main/scala/com/outworkers/phantom/ops/QueryContext.scala

@@ -34,10 +34,11 @@ abstract class QueryContext[P[_], F[_], Timeout](
   defaultTimeout: Timeout
 )(
   implicit fMonad: FutureMonad[F],
-  val promiseInterface: PromiseInterface[P, F],
-  val adapter: GuavaAdapter[F]
+  val promiseInterface: PromiseInterface[P, F]
 ) { outer =>
 
+  implicit val adapter: GuavaAdapter[F] = promiseInterface.adapter
+
   def executeStatements[M[X] <: TraversableOnce[X]](
     col: QueryCollection[M]
   ): ExecutableStatements[F, M] = {
@@ -48,7 +49,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
 
   implicit class BatchOps[Status <: ConsistencyBound](val query: BatchQuery[Status]) {
     def future()(implicit session: Session, ctx: ExecutionContextExecutor): F[ResultSet] = {
-      adapter.executeBatch(query.makeBatch())
+      promiseInterface.adapter.executeBatch(query.makeBatch())
     }
   }
 
@@ -60,7 +61,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
     def future()(
       implicit session: Session,
       ctx: ExecutionContextExecutor
-    ): F[ResultSet] = adapter.fromGuava(query.executableQuery)
+    ): F[ResultSet] = promiseInterface.adapter.fromGuava(query.executableQuery)
   }
 
   implicit class RootSelectBlockOps[
@@ -98,7 +99,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
     PS <: HList
   ](
     override val query: SelectQuery[Table, Record, Limit, Order, Status, Chain, PS]
-  ) extends SelectQueryOps(query) {
+  ) extends SelectQueryOps(query)(promiseInterface.adapter, fMonad) {
     override def executableQuery: ExecutableCqlQuery = query.executableQuery
   }
 
@@ -137,7 +138,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
       implicit session: Session,
       ec: ExecutionContextExecutor
     ): F[ResultSet] = {
-      adapter.fromGuava(query.options(query.st))
+      promiseInterface.adapter.fromGuava(query.options(query.st))
     }
 
     /**
@@ -156,7 +157,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
     override def future(modifyStatement: Statement => Statement)(
       implicit session: Session,
       executor: ExecutionContextExecutor
-    ): F[ResultSet] = adapter.fromGuava(modifyStatement(query.options(query.st)))
+    ): F[ResultSet] = promiseInterface.adapter.fromGuava(modifyStatement(query.options(query.st)))
   }
 
   implicit class ExecutablePreparedSelect[
@@ -183,7 +184,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
       implicit session: Session,
       ec: ExecutionContextExecutor
     ): F[ResultSet] = {
-      adapter.fromGuava(query.options(query.st))
+      promiseInterface.adapter.fromGuava(query.options(query.st))
     }
 
     /**
@@ -267,7 +268,7 @@ abstract class QueryContext[P[_], F[_], Timeout](
       sg: SingleGeneric.Aux[V1, Repr, HL, Out],
       ctx: ExecutionContextExecutor,
       ev: Out ==:== Repr
-    ): F[ResultSet] = adapter.fromGuava(table.store(input).executableQuery)
+    ): F[ResultSet] = promiseInterface.adapter.fromGuava(table.store(input).executableQuery)
 
     def storeRecords[M[X] <: TraversableOnce[X], V1, Repr <: HList, HL <: HList, Out <: HList](inputs: M[V1])(
       implicit keySpace: KeySpace,

phantom-dsl/src/main/scala/com/outworkers/phantom/ops/SelectQueryOps.scala

@@ -88,7 +88,6 @@ class SelectQueryOps[
     ev: PS =:!= HNil,
     rev: Reverse.Aux[PS, Rev],
     fMonad: FutureMonad[F],
-    adapter: GuavaAdapter[F],
     interface: PromiseInterface[P, F]
   ): F[PreparedSelectBlock[Table, Record, Limit, Rev]] = {
     val flatten = new PreparedFlattener(executableQuery.qb)

phantom-finagle/src/main/scala/com/outworkers/phantom/finagle/execution/TwitterQueryContext.scala

@@ -55,8 +55,7 @@ object TwitterPromiseInterface extends PromiseInterface[Promise, Future] {
 
 class TwitterQueryContext extends QueryContext[Promise, Future, Duration](10.seconds)(
   TwitterFutureImplicits.monadInstance,
-  TwitterPromiseInterface,
-  TwitterGuavaAdapter
+  TwitterPromiseInterface
 ) {
 
   override def await[T](f: Future[T], timeout: Duration): T = Await.result(f, timeout)

phantom-streams/src/main/scala/com/outworkers/phantom/streams/package.scala

@@ -225,11 +225,26 @@ package object streams {
       }
     }
 
+    /**
+      * Creates a Reactive Streams publisher from a root select block.
+      * This will create a reactive streams publisher containing all the records found by this [[SelectQuery]].
+      * @param session The database session in which to execute this.
+      * @param ctx The execution context to use.
+      * @return A streams publisher interface clients can subscribe to.
+      */
     def publisher()(
       implicit session: Session,
       ctx: ExecutionContextExecutor
     ): Publisher[R] = enumeratorToPublisher(query.fetchEnumerator())
 
+    /**
+      * Creates a Reactive Streams publisher from a root select block but also
+      * allows passing through a query modifier.
+      * This will create a reactive streams publisher containing all the records found by this [[SelectQuery]].
+      * @param session The database session in which to execute this.
+      * @param ctx The execution context to use.
+      * @return A streams publisher interface clients can subscribe to.
+      */
     def publisher(modifier: Statement => Statement)(
       implicit session: Session,
       ctx: ExecutionContextExecutor
@@ -283,12 +298,31 @@ package object streams {
       }
     }
 
+    /**
+      * Creates a Reactive Streams publisher from a root select block.
+      * Because this is a [[RootSelectBlock]], the default execution profile
+      * of this method will be to select all records in a table and stream them.
+      * @param session The database session in which to execute this.
+      * @param keySpace The keyspace in which to execute the query.
+      * @param ctx The execution context to use.
+      * @return A streams publisher interface clients can subscribe to.
+      */
     def publisher()(
       implicit session: Session,
       keySpace: KeySpace,
       ctx: ExecutionContextExecutor
     ): Publisher[R] = enumeratorToPublisher(block.fetchEnumerator())
 
+    /**
+      * Creates a Reactive Streams publisher from a root select block but also
+      * allows passing through a query modifier.
+      * Because this is a [[RootSelectBlock]], the default execution profile
+      * of this method will be to select all records in a table and stream them.
+      * @param session The database session in which to execute this.
+      * @param keySpace The keyspace in which to execute the query.
+      * @param ctx The execution context to use.
+      * @return A streams publisher interface clients can subscribe to.
+      */
     def publisher(modifier: Statement => Statement)(
       implicit session: Session,
       keySpace: KeySpace,