scaladsl for CassandraSession, #1

* moved CassandraSessionImpl to akka-persistence-cassandra,
  since we already had a similar class there and some users
  of akka-persistence-cassandra have requested it
* akka-persistence-cassandra provides both scaladsl and javadsl
  and in lagom we can have thin api wrappers (delegating)

build.sbt

@@ -15,7 +15,7 @@ import de.heikoseeberger.sbtheader.HeaderPattern
 
 val PlayVersion = "2.5.4"
 val AkkaVersion = "2.4.11"
-val AkkaPersistenceCassandraVersion = "0.17"
+val AkkaPersistenceCassandraVersion = "0.19-SNAPSHOT"
 val ScalaTestVersion = "2.2.4"
 val JacksonVersion = "2.7.8"
 val CassandraAllVersion = "3.0.9"

persistence-cassandra/javadsl/src/main/resources/reference.conf

@@ -81,6 +81,12 @@ lagom.persistence.read-side {
     # Maximum size of result set
     max-result-size = 50001
 
+    # Number of retries before giving up connecting for the initial connection to the Cassandra cluster
+    connect-retries = 3
+
+    # Delay between connection retries, for the initial connection to the Cassandra cluster
+    connect-retry-delay = 1s
+
     # Max delay of the ExponentialReconnectionPolicy that is used when reconnecting
     # to the Cassandra cluster
     reconnect-max-delay = 30s
@@ -120,6 +126,25 @@ lagom.persistence.read-side {
     # Set the protocol version explicitly, should only be used for compatibility testing.
     # Supported values: 3, 4
     protocol-version = ""
+
+    # Options to configure low-level socket options for the connections to Cassandra hosts
+    # See: https://datastax.github.io/java-driver/manual/socket_options
+    socket {
+
+      # how long the driver waits to establish a new connection to a Cassandra node before giving up
+      connection-timeout-millis = 5000
+
+      # the per-host read timeout in milliseconds. Should be higher than the timeout settings used on the Cassandra side
+      read-timeout-millis = 12000
+
+      # a hint to the size of the underlying buffers for outgoing network I/O. Set to zero to
+      # use the default from the underlying Netty transport (Java NIO or native epoll)
+      send-buffer-size = 0
+
+      # a hint to the size of the underlying buffers for incoming network I/O. Set to zero to
+      # use the default from the underlying Netty transport (Java NIO or native epoll)
+      receive-buffer-size = 0
+    }
 
   }
 

persistence-cassandra/javadsl/src/main/scala/com/lightbend/lagom/javadsl/persistence/cassandra/CassandraPersistenceModule.scala

@@ -14,7 +14,7 @@ import com.lightbend.lagom.javadsl.persistence.PersistentEntityRegistry
 class CassandraPersistenceModule extends AbstractModule {
   override def configure(): Unit = {
     binder.bind(classOf[PersistentEntityRegistry]).to(classOf[CassandraPersistentEntityRegistry])
-    binder.bind(classOf[CassandraSession]).to(classOf[CassandraSessionImpl])
+    binder.bind(classOf[CassandraSession])
     binder.bind(classOf[CassandraReadSide]).to(classOf[CassandraReadSideImpl])
     binder.bind(classOf[CassandraConfig]).toProvider(classOf[CassandraConfigProvider])
     binder.bind(classOf[CassandraOffsetStore])