Cassandra sink #56

stephennancekivell · 2016-11-14T10:20:00Z

Hey, should we provide a Cassandra sink like this.

It seems pretty simple, but I think this is what a user would want.

What do you think ?

+def apply[T](parallelism: Int, statement: PreparedStatement, statementBinder: (T, PreparedStatement) => BoundStatement)(implicit session: Session, ex: ExecutionContext): Sink[T, Future[Done]] =
+    Flow[T].mapAsyncUnordered(parallelism)(t ⇒
+      session.executeAsync(statementBinder(t, statement)).toFuture()
+    ).toMat(Sink.ignore)(Keep.right)
+}

patriknw

I like this. Please add javadsl also.

patriknw · 2016-11-14T13:57:33Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/GuavaFutureOpts.scala

+object GuavaFutureOptsImplicits {
+  implicit def toGuavaFutureOpts[A](guavaFut: ListenableFuture[A]): GuavaFutureOpts[A] =
+    new GuavaFutureOpts[A](guavaFut)
+}


use implicit class ... extends AnyVal instead, like this: https://github.com/akka/akka-persistence-cassandra/blob/master/src/main/scala/akka/persistence/cassandra/package.scala#L22

patriknw · 2016-11-14T13:59:03Z

cassandra/src/test/scala/akka/stream/alpakka/cassandra/scaladsl/CassandraSourceSpec.scala

 import com.datastax.driver.core.{ Cluster, SimpleStatement }
 import org.scalatest._
 import org.scalatest.concurrent.ScalaFutures

 import scala.concurrent._
 import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global


Avoid this EC, also in tests, unless there is a good reason. Use import system.dispatcher inside the test instead.

patriknw · 2016-11-14T14:01:24Z

cassandra/src/test/scala/akka/stream/alpakka/cassandra/scaladsl/CassandraSourceSpec.scala

 import com.datastax.driver.core.{ Cluster, SimpleStatement }
 import org.scalatest._
 import org.scalatest.concurrent.ScalaFutures

 import scala.concurrent._
 import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
+
+import scala.collection.convert.decorateAsScala._


I think it's the same but I prefer scala.collection.JavaConverters._

stephennancekivell · 2016-11-16T06:19:24Z

Hi @patriknw, I've implemented your feedback, thanks for that.

Im not sure how to do the java dsl at the moment and it might be a few weeks before I have time to look at it. I agree it will be good for this project.

So, I dont know if you want to merge this in without java or not. Up to you.

Cheers,

patriknw · 2016-11-16T17:44:15Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/scaladsl/CassandraSink.scala

+import akka.stream.alpakka.cassandra.cassandra._
+
+object CassandraSink {
+  def apply[T](parallelism: Int, statement: PreparedStatement, statementBinder: (T, PreparedStatement) => BoundStatement)(implicit session: Session, ex: ExecutionContext): Sink[T, Future[Done]] =


some hints for the javadsl

statementBinder: BiFunction[T, PreparedStatement, BoundStatement]

delegate to scaladsl.CassandraSink.apply

mapMaterializedValue and scala.compat.java8.FutureConverters._ to convert the Future to CompletionStage

Sink.asJava to convert to akka.stream.javadsl.Sink

patriknw · 2016-11-16T17:47:39Z

@stephennancekivell We would like to see it completed with javadsl and some small usage example in the docs before merging.

stephennancekivell · 2016-11-17T02:36:21Z

Awesome, Thanks for the tips. Will do.

On Thu, 17 Nov 2016, 06:47 Patrik Nordwall notifications@github.com wrote:

@stephennancekivell https://github.com/stephennancekivell We would like
to see it completed with javadsl and some small usage example in the docs
https://github.com/akka/alpakka/blob/master/docs/src/main/paradox/cassandra.md
before merging.

—
You are receiving this because you were mentioned.

Reply to this email directly, view it on GitHub
#56 (comment), or mute
the thread
https://github.com/notifications/unsubscribe-auth/AAsX5W0QGxKy0HdZfEscYAO63UCdMPcZks5q-0G8gaJpZM4KxJ_H
.

stephennancekivell · 2016-11-22T07:06:30Z

Hi @patriknw I've completed the Java DSL and the logging. Do you or anyone else have any feedback?

johanandren

LGTM, some minor nitpicks though

johanandren · 2016-11-22T09:35:08Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/javadsl/CassandraSink.scala

+                session: Session): Sink[T, CompletionStage[Done]] = {
+    val sink = ScalaCSink.apply[T](parallelism, statement, (t, p) => statementBinder.apply(t, p))(
+      session,
+      scala.concurrent.ExecutionContext.global)


Take the EC as a parameter to create instead, just like the Scala api does (if done like this the user has no control).

johanandren · 2016-11-22T09:35:51Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/package.scala

+
+import scala.concurrent.{ Future, Promise }
+
+package object cassandra {


private[cassandra]

Hi Johan, where do you want me to put that. I cant get it to work. Its a compiler error on the object and cant be used on the implicit class. ???

Ah, you cannot put access restrictions on the package object itself, so then it needs to go on the implicit decorator, if kept public the risk is that library consumers start using it and then it cannot be moved/changed in the future, and this library is not here to provide guice-conversions.

Good point. Thanks.

I had two issues,

I had to change the object name from cassandara, I think the private[cassandra] declaration was referencing the object instead of the package, which stopped it working.

Then I had to add a implicit def to convert to the opts. It seems like the implicit class wasnt getting picked up with the private.

I that right,

johanandren · 2016-11-22T09:36:02Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/package.scala

+import scala.concurrent.{ Future, Promise }
+
+package object cassandra {
+  implicit class GuavaFutureOpts[A](val guavaFut: ListenableFuture[A]) extends AnyVal {


johanandren · 2016-11-22T09:38:54Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/CassandraSourceStage.scala

 import scala.util.{ Failure, Success, Try }

+import akka.stream.alpakka.cassandra.cassandra._
+
 class CassandraSourceStage(futStmt: Future[Statement], session: Session) extends GraphStage[SourceShape[Row]] {


final class (please boyscout this for us)

johanandren · 2016-11-22T11:00:19Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/package.scala

@@ -7,8 +7,8 @@ import com.google.common.util.concurrent.{ FutureCallback, Futures, ListenableFu

 import scala.concurrent.{ Future, Promise }

-package object cassandra {
-  implicit final class GuavaFutureOpts[A](val guavaFut: ListenableFuture[A]) extends AnyVal {
+package object GuavaFutureOpts {


With this name it is no longer a package object, just a normal object, so remove that and make it package private for cassandra with private[cassandra] which will cover the class inside it as well. You will have to import the implicit class into the scope where it is used though, so that will be an import GuavaFutureOpts._ in the classes you call asScala

Right, thanks. However i still need the implicit def the implicit class alone doesnt seem to get picked up...

johanandren · 2016-11-22T11:14:22Z

cassandra/src/main/scala/akka/stream/alpakka/cassandra/GuavaFutureOpts.scala

@@ -20,6 +20,6 @@ package object GuavaFutureOpts {
    }
  }

-  private[cassandra] implicit def toGuavaFutureOpts[A](guavaFut: ListenableFuture[A]): GuavaFutureOpts[A] =
+  implicit def toGuavaFutureOpts[A](guavaFut: ListenableFuture[A]): GuavaFutureOpts[A] =


if you make the class implicit you do not need this either

stephennancekivell · 2016-11-22T11:23:46Z

@johanandren I've just pushed a commit with the changes like you suggested. However now Its a compiler error. So what do I do here. It seemed ok to me with the def. Is there a better way.

https://travis-ci.org/akka/alpakka/jobs/177938167#L773

Note the previous build failed, because of Amqp TimeoutException.

johanandren · 2016-11-22T13:02:13Z

If you look at the docs for implicit classes (http://docs.scala-lang.org/overviews/core/implicit-classes.html) you can see that an implicit class cannot have the same name as something else that is already in scope, therefore GuavaFutureOpts inside the object GuavaFutureOpts which is in the same package as where it is used will not work.

Rename the object or the implicit class and that will solve the problem.

patriknw

LGTM

stephennancekivell · 2016-11-23T07:29:56Z

Thanks @patriknw

Thanks @johanandren I hadn't come across that rule before.

stephennancekivell · 2016-11-30T05:43:35Z

hey @patriknw @johanandren,

Can this be merged in ?

Thanks

patriknw · 2016-11-30T17:52:58Z

Refs #57

patriknw · 2016-11-30T17:53:16Z

Excellent contribution @stephennancekivell !

stephennancekivell added 2 commits November 14, 2016 21:07

add simple cassandra sink

9bc59d4

fix async

da6f91d

stephennancekivell mentioned this pull request Nov 14, 2016

Cassandra Sink #57

Closed

patriknw reviewed Nov 14, 2016

View reviewed changes

PR feedback.

29314ab

patriknw reviewed Nov 16, 2016

View reviewed changes

stephennancekivell added 4 commits November 21, 2016 19:24

added java dsl for cassandra sink

3e5c17c

add cassandra sink docs

2af79f8

Merge branch 'master' into cassandra-sink

197ecc2

apply scalafmt after merge

56a1021

johanandren approved these changes Nov 22, 2016

View reviewed changes

stephennancekivell added 2 commits November 22, 2016 20:58

pr feedback

595ed76

make GuavaFutureOpts private

b4da679

johanandren reviewed Nov 22, 2016

View reviewed changes

not a package object.

e6fa29d

johanandren reviewed Nov 22, 2016

View reviewed changes

implicit class, no implicit def.

a682466

patriknw approved these changes Nov 22, 2016

View reviewed changes

stephennancekivell added 2 commits November 23, 2016 18:19

fix implicit class of same name as object containing it.

9d80e1f

no change to trigger travis

ac18a87

patriknw merged commit 83dcecb into akka:master Nov 30, 2016

stephennancekivell deleted the cassandra-sink branch December 2, 2016 08:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Cassandra sink #56

Cassandra sink #56

stephennancekivell commented Nov 14, 2016

patriknw left a comment

patriknw Nov 14, 2016

patriknw Nov 14, 2016

patriknw Nov 14, 2016

stephennancekivell commented Nov 16, 2016

patriknw Nov 16, 2016

patriknw commented Nov 16, 2016

stephennancekivell commented Nov 17, 2016

stephennancekivell commented Nov 22, 2016

johanandren left a comment

johanandren Nov 22, 2016

johanandren Nov 22, 2016

stephennancekivell Nov 22, 2016

johanandren Nov 22, 2016

stephennancekivell Nov 22, 2016

johanandren Nov 22, 2016

johanandren Nov 22, 2016

johanandren Nov 22, 2016

stephennancekivell Nov 22, 2016

johanandren Nov 22, 2016

stephennancekivell commented Nov 22, 2016 •

edited

Loading

johanandren commented Nov 22, 2016

patriknw left a comment

stephennancekivell commented Nov 23, 2016

stephennancekivell commented Nov 30, 2016

patriknw commented Nov 30, 2016

patriknw commented Nov 30, 2016


		import scala.concurrent.{ Future, Promise }

		package object cassandra {

Cassandra sink #56

Cassandra sink #56

Conversation

stephennancekivell commented Nov 14, 2016

patriknw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephennancekivell commented Nov 16, 2016

Choose a reason for hiding this comment

patriknw commented Nov 16, 2016

stephennancekivell commented Nov 17, 2016

stephennancekivell commented Nov 22, 2016

johanandren left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephennancekivell commented Nov 22, 2016 • edited Loading

johanandren commented Nov 22, 2016

patriknw left a comment

Choose a reason for hiding this comment

stephennancekivell commented Nov 23, 2016

stephennancekivell commented Nov 30, 2016

patriknw commented Nov 30, 2016

patriknw commented Nov 30, 2016

stephennancekivell commented Nov 22, 2016 •

edited

Loading