Feature/streams 2.12 (#693)

* Adding streams for 2.12 * Adding Enumerator implementation. * Removing obsolete deps * Adding nullValue aliases for codacy and porting over test. * Fixing test * Fixing more tests. * Fixing Scala 2.12 compilation * Updating the readme * Improving on Codacy feedback * More codacy feedback * Adding encrypted token. * Re-adding streaming tests * Re-enabling iteratee tests. * Fixing tests and adding dedicated tests for typed primitive derivation.
outworkers · May 26, 2017 · 33f84f7 · 33f84f7
1 parent 7075d84
commit 33f84f7
Show file tree

Hide file tree

Showing 23 changed files with 978 additions and 166 deletions.
diff --git a/.travis.yml b/.travis.yml
@@ -1,12 +1,9 @@
 language: scala
-
 sudo: required
 dist: trusty
-
 scala:
 - 2.10.6
 - 2.11.8
-
 env:
   global:
   - GH_REF: github.com/outworkers/phantom.git
@@ -15,40 +12,32 @@ env:
   - secure: hbuc+lReAC/tzOfmF3M70L4Qfb48S6fI/gqchctnBD7TcZ6i13EOLhsjgPkFwTuINYU7Bvwd54yjy9simu1Se34+DaGtxKPJ8NZIeWk0ciCGyFxNiz0QVkE+/Cc0kZLD7hJXb0UpLsbMwpIQFlL6FHcB/2xGmMhgYvGHe7+64zM=
   - secure: ma0SLHvjzABno6RmwR/LSZQVC6rv6UcUt5kJAz1l4yJmASpdxkivoa8xWXahsjYP1O05IWuPFGHy/o0Vu7fxykgqN5gOg01wVQ0irjguWR04nNLUQTrOUgKPXR3w22USAbPpzxxqQrGc0ZS/2QFpiu9fvLNsBTkZcosA3oLpyiY=
   - secure: hihHuKLYf8YemlopCITLM1QvvgJH7ITTTsLRvLxKOGfzzBsOLd1m6LKpDOrmWdoKtcigyON+J3AP24DBd4zO95Y8r+Cwq8I4IVfZol3qwhJxvYMSB4Tyje3ssSUGqr/LBUBdLAe8H5pBeoeAECt/iSjDcYYRLKVexWRaJJBafXY=
-
+  - secure: "EyY0mxgk+rA0Vc0O/rb6pkcN1H5jXfPLkG6lJzH9PhljzLRc/5PltOJ3HMTCe1lgNQBa4kdVRqGAEkMcaFLWWLiCWJeN1lTTvBeFDBGtF9Abyc0Fr1Z0eOjiQofUwQoEyUagfsd5f8ypikruGubxvfh4bQndJdDpqMAi9z/u8ks="
 notifications:
   slack:
   - websudos:P9QNXx1ZGFnDHp3v3jUqtB8k
   email:
   - dev@outworkers.com
-
 branches:
   only:
   - master
   - develop
-
 jdk:
 - oraclejdk7
 - oraclejdk8
-
 matrix:
   include:
   - scala: 2.12.1
     jdk: oraclejdk8
-
 addons:
   apt:
     packages:
-      - oracle-java8-installer
-
+    - oracle-java8-installer
 before_install: unset SBT_OPTS JVM_OPTS
-
-
 before_script:
-  - mkdir -p $HOME/.sbt/launchers/0.13.15/
-  - curl -L -o $HOME/.sbt/launchers/0.13.15/sbt-launch.jar https://repo.typesafe.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/0.13.15/sbt-launch.jar
-  - travis_retry sbt "plz $TRAVIS_SCALA_VERSION update"
-
+- mkdir -p $HOME/.sbt/launchers/0.13.15/
+- curl -L -o $HOME/.sbt/launchers/0.13.15/sbt-launch.jar https://repo.typesafe.com/typesafe/ivy-releases/org.scala-sbt/sbt-launch/0.13.15/sbt-launch.jar
+- travis_retry sbt "plz $TRAVIS_SCALA_VERSION update"
 install: ./build/install_cassandra.sh
 script: ./build/run_tests.sh
-after_success: ./build/publish_develop.sh
+after_success: ./build/publish_develop.sh
diff --git a/README.md b/README.md
@@ -29,14 +29,14 @@ This is a table of the available modules for the various Scala versions. Not all
 
 | Module name           | Scala 2.10.x        | Scala 2.11.x      | Scala 2.12.0      |
 | ------------          | ------------------- | ------------------| ----------------- |
-| phantom-connectors    | <span>yes</span>    | <span>yes</span> | <span>yes</span>  |
-| phantom-dsl           | <span>yes</span>    | <span>yes</span> | <span>yes</span>  |
-| phantom-jdk8          | <span>yes</span>    | <span>yes</span> | <span>yes</span>  |
-| phantom-sbt           | <span>yes</span>    | <span>no</span>  | <span>no</span>   |
+| phantom-connectors    | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
+| phantom-dsl           | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
+| phantom-jdk8          | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
+| phantom-sbt           | <span>yes</span>    | <span>no</span>  | <span>no</span>    |
 | phantom-example       | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
 | phantom-thrift        | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
 | phantom-finagle       | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
-| phantom-streams       | <span>yes</span>    | <span>yes</span> | <span>no</span>   |
+| phantom-streams       | <span>yes</span>    | <span>yes</span> | <span>yes</span>   |
 
 #### Phantom Pro subscription edition
 

diff --git a/build.sbt b/build.sbt
@@ -83,28 +83,9 @@ lazy val Versions = new {
       case _ => "2.4.8"
     }
   }
-
-  val playStreams: String => ModuleID = s => {
-    val v = play(s)
-    CrossVersion.partialVersion(s) match {
-      case Some((_, minor)) if minor == 12=> {
-        "com.typesafe.play" %% "play-reactive-streams" % v
-      }
-      case Some((_, minor)) if minor == 11 && Publishing.isJdk8 => {
-        "com.typesafe.play" %% "play-streams" % v
-      }
-      case Some((_, minor)) if minor >= 11  && !Publishing.isJdk8 => {
-        "com.typesafe.play" %% "play-streams-experimental" % "2.4.8"
-      }
-      case _ => "com.typesafe.play" %% "play-streams-experimental" % v
-    }
-  }
 }
-val defaultConcurrency = 4
 
-val PerformanceTest = config("perf").extend(Test)
-
-lazy val performanceFilter: String => Boolean = _.endsWith("PerformanceTest")
+val defaultConcurrency = 4
 
 scalacOptions in ThisBuild ++= Seq(
   "-language:experimental.macros",
@@ -121,7 +102,7 @@ scalacOptions in ThisBuild ++= Seq(
 
 val sharedSettings: Seq[Def.Setting[_]] = Defaults.coreDefaultSettings ++ Seq(
   organization := "com.outworkers",
-  scalaVersion := "2.11.8",
+  scalaVersion := "2.12.1",
   credentials ++= Publishing.defaultCredentials,
   resolvers ++= Seq(
     "Twitter Repository" at "http://maven.twttr.com",
@@ -142,10 +123,7 @@ val sharedSettings: Seq[Def.Setting[_]] = Defaults.coreDefaultSettings ++ Seq(
   ),
   envVars := Map("SCALACTIC_FILL_FILE_PATHNAMES" -> "yes"),
   gitTagName in ThisBuild := s"version=${scalaVersion.value}",
-  testFrameworks in PerformanceTest := Seq(new TestFramework("org.scalameter.ScalaMeterFramework")),
-  testOptions in Test := Seq(Tests.Filter(x => !performanceFilter(x))),
-  testOptions in PerformanceTest := Seq(Tests.Filter(performanceFilter)),
-  fork in PerformanceTest := false,
+  testFrameworks in Test ++= Seq(new TestFramework("org.scalameter.ScalaMeterFramework")),
   parallelExecution in ThisBuild := false
 ) ++ VersionManagement.newSettings ++
   GitProject.gitSettings ++
@@ -163,11 +141,7 @@ lazy val baseProjectList: Seq[ProjectReference] = Seq(
 lazy val fullProjectList = baseProjectList ++ Publishing.addOnCondition(Publishing.isJdk8, phantomJdk8)
 
 lazy val phantom = (project in file("."))
-  .configs(
-    PerformanceTest
-  ).settings(
-    inConfig(PerformanceTest)(Defaults.testTasks): _*
-  ).settings(
+  .settings(
     sharedSettings ++ Publishing.noPublishSettings
   ).settings(
     name := "phantom",
@@ -177,37 +151,33 @@ lazy val phantom = (project in file("."))
     fullProjectList: _*
   ).enablePlugins(CrossPerProjectPlugin)
 
-lazy val phantomDsl = (project in file("phantom-dsl")).configs(
-  PerformanceTest
-).settings(
-  inConfig(PerformanceTest)(Defaults.testTasks): _*
-).settings(
-  sharedSettings: _*
-).settings(
-  name := "phantom-dsl",
-  moduleName := "phantom-dsl",
-  concurrentRestrictions in Test := Seq(
-    Tags.limit(Tags.ForkedTestGroup, defaultConcurrency)
-  ),
-  crossScalaVersions := Seq("2.10.6", "2.11.8", "2.12.1"),
-  libraryDependencies ++= Seq(
-    "org.typelevel" %% "macro-compat" % Versions.macrocompat,
-    "org.scala-lang" % "scala-compiler" % scalaVersion.value % "provided",
-    compilerPlugin("org.scalamacros" % "paradise" % Versions.macroParadise cross CrossVersion.full),
-    "com.chuusai"                  %% "shapeless"                         % Versions.shapeless,
-    "joda-time"                    %  "joda-time"                         % Versions.joda,
-    "org.joda"                     %  "joda-convert"                      % Versions.jodaConvert,
-    "com.datastax.cassandra"       %  "cassandra-driver-core"             % Versions.datastax,
-    "org.json4s"                   %% "json4s-native"                     % Versions.json4s % Test,
-    "org.scalamock"                %% "scalamock-scalatest-support"       % Versions.scalamock % Test,
-    "org.scalacheck"               %% "scalacheck"                        % Versions.scalacheck % Test,
-    "com.outworkers"               %% "util-samplers"                     % Versions.util % Test,
-    "com.storm-enroute"            %% "scalameter"                        % Versions.scalameter % Test,
-    "ch.qos.logback"               % "logback-classic"                    % Versions.logback % Test
-  )
-).dependsOn(
-  phantomConnectors
-).enablePlugins(CrossPerProjectPlugin)
+lazy val phantomDsl = (project in file("phantom-dsl"))
+  .settings(sharedSettings: _*)
+  .settings(
+    name := "phantom-dsl",
+    moduleName := "phantom-dsl",
+    concurrentRestrictions in Test := Seq(
+      Tags.limit(Tags.ForkedTestGroup, defaultConcurrency)
+    ),
+    crossScalaVersions := Seq("2.10.6", "2.11.8", "2.12.1"),
+    libraryDependencies ++= Seq(
+      "org.typelevel" %% "macro-compat" % Versions.macrocompat,
+      "org.scala-lang" % "scala-compiler" % scalaVersion.value % "provided",
+      compilerPlugin("org.scalamacros" % "paradise" % Versions.macroParadise cross CrossVersion.full),
+      "com.chuusai"                  %% "shapeless"                         % Versions.shapeless,
+      "joda-time"                    %  "joda-time"                         % Versions.joda,
+      "org.joda"                     %  "joda-convert"                      % Versions.jodaConvert,
+      "com.datastax.cassandra"       %  "cassandra-driver-core"             % Versions.datastax,
+      "org.json4s"                   %% "json4s-native"                     % Versions.json4s % Test,
+      "org.scalamock"                %% "scalamock-scalatest-support"       % Versions.scalamock % Test,
+      "org.scalacheck"               %% "scalacheck"                        % Versions.scalacheck % Test,
+      "com.outworkers"               %% "util-samplers"                     % Versions.util % Test,
+      "com.storm-enroute"            %% "scalameter"                        % Versions.scalameter % Test,
+      "ch.qos.logback"               % "logback-classic"                    % Versions.logback % Test
+    )
+  ).dependsOn(
+    phantomConnectors
+  ).enablePlugins(CrossPerProjectPlugin)
 
 lazy val phantomJdk8 = (project in file("phantom-jdk8"))
   .settings(
@@ -225,7 +195,6 @@ lazy val phantomJdk8 = (project in file("phantom-jdk8"))
   ).enablePlugins(CrossPerProjectPlugin)
 
 lazy val phantomConnectors = (project in file("phantom-connectors"))
-  .configs(PerformanceTest)
   .settings(
     sharedSettings: _*
   ).settings(
@@ -238,7 +207,8 @@ lazy val phantomConnectors = (project in file("phantom-connectors"))
   ).enablePlugins(CrossPerProjectPlugin)
 
 lazy val phantomFinagle = (project in file("phantom-finagle"))
-  .configs(PerformanceTest).settings(
+  .settings(sharedSettings: _*)
+  .settings(
     name := "phantom-finagle",
     moduleName := "phantom-finagle",
     crossScalaVersions := Seq("2.10.6", "2.11.8", "2.12.1"),
@@ -248,8 +218,6 @@ lazy val phantomFinagle = (project in file("phantom-finagle"))
       "com.outworkers"               %% "util-testing-twitter"              % Versions.util % Test,
       "com.storm-enroute"            %% "scalameter"                        % Versions.scalameter % Test
     )
-  ).settings(
-    inConfig(PerformanceTest)(Defaults.testTasks) ++ sharedSettings: _*
   ).dependsOn(
     phantomDsl % "compile->compile;test->test"
   ).enablePlugins(CrossPerProjectPlugin)
@@ -259,7 +227,7 @@ lazy val phantomThrift = (project in file("phantom-thrift"))
     name := "phantom-thrift",
     moduleName := "phantom-thrift",
     crossScalaVersions := Seq("2.10.6", "2.11.8", "2.12.1"),
-    addCompilerPlugin("org.scalamacros" % "paradise" % "2.1.0" cross CrossVersion.full),
+    addCompilerPlugin("org.scalamacros" % "paradise" % Versions.macroParadise cross CrossVersion.full),
     libraryDependencies ++= Seq(
       "org.typelevel" %% "macro-compat" % Versions.macrocompat,
       "org.scala-lang" % "scala-compiler" % scalaVersion.value % "provided",
@@ -298,10 +266,9 @@ lazy val phantomStreams = (project in file("phantom-streams"))
   .settings(
     name := "phantom-streams",
     moduleName := "phantom-streams",
-    crossScalaVersions := Seq("2.10.6", "2.11.8"),
+    crossScalaVersions := Seq("2.10.6", "2.11.8", "2.12.1"),
     libraryDependencies ++= Seq(
       "com.typesafe" % "config" % Versions.typesafeConfig force(),
-      Versions.playStreams(scalaVersion.value),
       "com.typesafe.play"   %% "play-iteratees" % Versions.play(scalaVersion.value) exclude ("com.typesafe", "config"),
       "org.reactivestreams" % "reactive-streams"            % Versions.reactivestreams,
       "com.typesafe.akka"   %% s"akka-actor"                % Versions.akka(scalaVersion.value) exclude ("com.typesafe", "config"),

diff --git a/phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/CassandraOperations.scala b/phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/CassandraOperations.scala
@@ -15,13 +15,8 @@
  */
 package com.outworkers.phantom.builder.query
 
-import com.datastax.driver.core.{
-  PreparedStatement,
-  Session,
-  Statement,
-  ResultSet => DatastaxResultSet
-}
-import com.google.common.util.concurrent.{FutureCallback, Futures}
+import com.datastax.driver.core.{PreparedStatement, Session, Statement, ResultSet => DatastaxResultSet}
+import com.google.common.util.concurrent.{FutureCallback, Futures, ListenableFuture}
 import com.outworkers.phantom.{Manager, ResultSet}
 import com.outworkers.phantom.connectors.SessionAugmenterImplicits
 

diff --git a/phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/DeleteQuery.scala b/phantom-dsl/src/main/scala/com/outworkers/phantom/builder/query/DeleteQuery.scala
@@ -246,7 +246,9 @@ sealed class ConditionalDeleteQuery[
 
   override val qb: CQLQuery = (usingPart merge wherePart merge casPart) build init
 
-  final def and(clause: Table => CompareAndSetClause.Condition): ConditionalDeleteQuery[Table, Record, Limit, Order, Status, Chain, PS] = {
+  final def and(
+    clause: Table => CompareAndSetClause.Condition
+  ): ConditionalDeleteQuery[Table, Record, Limit, Order, Status, Chain, PS] = {
     new ConditionalDeleteQuery(
       table,
       init,

diff --git a/phantom-dsl/src/test/scala/com/outworkers/phantom/builder/primitives/DerivedField.scala b/phantom-dsl/src/test/scala/com/outworkers/phantom/builder/primitives/DerivedField.scala
@@ -15,6 +15,8 @@
  */
 package com.outworkers.phantom.builder.primitives
 
+import com.outworkers.util.samplers.{Sample, Samples}
+
 case class DerivedField(value: String)
 
 object DerivedField {
@@ -34,3 +36,21 @@ object DerivedTupleField {
   }
 }
 
+sealed trait HashStatus
+trait Serialized extends HashStatus
+trait Unserialized extends HashStatus
+
+sealed case class Username[T <: HashStatus](name: String)
+
+object Username {
+
+  implicit val usernamePrimitive: Primitive[Username[Serialized]] = {
+    Primitive.derive[Username[Serialized], String](_.name)(Username.apply)
+  }
+
+  implicit val usernameSampler: Sample[Username[Serialized]] = {
+    Samples.derive[String, Username[Serialized]](Username.apply)
+  }
+
+  def apply(st: String, trigger: Int = 0): Username[Serialized] = new Username[Serialized](st)
+}
diff --git a/...sl/src/test/scala/com/outworkers/phantom/builder/primitives/PrimitiveRoundtripTests.scala b/...sl/src/test/scala/com/outworkers/phantom/builder/primitives/PrimitiveRoundtripTests.scala
@@ -248,4 +248,7 @@ class PrimitiveRoundtripTests extends FlatSpec
     sroundtrip[DerivedField]
   }
 
+  it should "serialize and deserialize a derived Primitive with a type" in {
+    sroundtrip[Username[Serialized]]
+  }
 }
diff --git a/...finagle/src/test/scala/com/outworkers/phantom/finagle/SpoolBenchmarkPerformanceTest.scala b/...finagle/src/test/scala/com/outworkers/phantom/finagle/SpoolBenchmarkPerformanceTest.scala
@@ -33,18 +33,14 @@ class SpoolBenchmarkPerformanceTest extends Bench.LocalTime with TestDatabase.co
   implicit object JodaTimeSampler extends Sample[DateTime] {
     override def sample: DateTime = DateTime.now(DateTimeZone.UTC)
   }
-
+  val sampleSize = 100
 
   val fs: IndexedSeq[Future[Unit]] = for {
     step <- 1 to 3
-    rows = Iterator.fill(10000)(gen[JodaRow])
+    rows = Iterator.fill(sampleSize)(gen[JodaRow])
 
     batch = rows.foldLeft(Batch.unlogged)((b, row) => {
-      val statement = TestDatabase.primitivesJoda.insert
-        .value(_.pkey, row.pkey)
-        .value(_.intColumn, row.intColumn)
-        .value(_.timestamp, row.timestamp)
-      b.add(statement)
+      b.add(TestDatabase.primitivesJoda.store(row))
     })
     w = batch.future()
     f = w map (_ => println(s"step $step has succeed") )

diff --git a/phantom-streams/src/main/scala/com/outworkers/phantom/streams/lib/EnumeratorPublisher.scala b/phantom-streams/src/main/scala/com/outworkers/phantom/streams/lib/EnumeratorPublisher.scala
@@ -0,0 +1,25 @@
+/*
+ * Copyright 2013 - 2017 Outworkers Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.outworkers.phantom.streams.lib
+
+import play.api.libs.iteratee.Enumerator
+
+/**
+  * Adapts an Enumerator to a Publisher.
+  */
+private[streams] final class EnumeratorPublisher[T](
+  val enum: Enumerator[T],
+  val emptyElement: Option[T] = None) extends RelaxedPublisher[T] with EnumeratorSubscriptionFactory[T]