Vert.x Adapter #310

cstub · 2016-09-13T08:15:39Z

The Vert.x Adapter offers various facilities to connect Eventuate event-logs to the Vert.x event-bus. Events can be read from an event-log and published to the Vert.x event-bus. Furthermore events may be read from the Vert.x event-bus and persisted to an event-log.

implement an adapter-system which is provided with adapter-configurations and an instance of Vert.x
implement a publish-adapter which reads events from an event-log and publishes them to all consumers on the Vert.x event-bus
implement a send-adapter which reads events from an event-log and sends them to a single consumer on the Vert.x event-bus
provide at-least-once-delivery semantics for the send-adapter with single confirmations or batched confirmations
implement a write-adapter which reads events from the Vert.x event-bus and persists them to an event-log
closes Vert.x Adapter #290

krasserm · 2016-09-19T06:30:45Z

When compiling I'm getting lots of

[warn] Class io.vertx.codegen.annotations.VertxGen not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.Fluent not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.Fluent not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.Fluent not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.GenIgnore not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.Fluent not found - continuing with a stub.
[warn] Class io.vertx.codegen.annotations.Fluent not found - continuing with a stub.
...

How shall we deal with it?

krasserm · 2016-09-19T06:32:31Z

Please add this adapter to the build config.

krasserm · 2016-09-19T06:42:02Z

project/ProjectDependencies.scala

@@ -22,6 +22,8 @@ object ProjectDependencyVersions {
  val Log4jVersion = "2.5"
  val ProtobufVersion = "2.5.0"
  val SparkVersion = "1.6.1"
+  val VertxVersion = "3.0.0"
+  val ExampleVertxVersion = "3.3.2"


Why different Vert.x versions for the adapter and the example?

I chose different versions because the Adapter is compatible with Vert.x versions > 3.0.0, but version 3.0.0 has some logging issues and because of that the output in the examples was messed up.

Assuming that 3.3.2 is backwards-compatible to 3.0.0, why not depending on 3.3.2 in the adapter then?

It is backwards compatible, but using 3.3.2 in the adapter also has the risk of using APIs not available in Vert.x 3.0.0, which would break the compatibility of the adapter, wouldn't it?

Isn't that only an issue when using a provided configuration? If we force applications to use 3.3.2 this should be ok, unless there are breaking changes in Vert.x 3.x minor versions. Is this the case?

The adapter should use Vert.x 3.0.0 as a provided dependency to not force the user into a specific Vert.x version because of breaking changes between minor versions (see https://github.com/vert-x3/wiki/wiki/3.1.0---Breaking-changes).

Ok, I see ...

krasserm · 2016-09-19T07:38:07Z

build.sbt seems to break the diff view on github, hence my comments here:

dependsOn(logLeveldb % "it->it") should be enough, the Cassandra dependencies can be removed
The Netty dependency of Akka Remote must not be excluded as it depends on Netty 3 which cannot be replaced by Netty 4. I wonder if it is a problem at all to have both Netty 3 and 4 on the classpath, as Netty 3 is in the org.jboss.netty package and Netty 4 in the io.netty package.
The multi-jvm configs/dependencies can be removed as well unless you plan to make multi-jvm tests (see also the Akka Streams adapter configuration for a more lightweight example)

krasserm · 2016-09-19T08:08:08Z

...src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterSystemConfig.scala

+  val writeAdapters: Seq[VertxWriteAdapterConfig] =
+    configurations.collect({ case c: VertxWriteAdapterConfig => c })
+
+  def addAdapter(first: VertxAdapterConfig, rest: VertxAdapterConfig*): VertxAdapterSystemConfig =


Shouldn't a method with name addAdapter have only a single VertxAdapterConfig parameter? All usage examples also don't use the rest vararg parameter.

You're right - I will rename the varags-version to addAdapters

Why not removing the rest parameter instead? They're never used in the example

you're right - the varargs version is not necessary, I will remove the parameter.

krasserm · 2016-09-19T08:09:46Z

...src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterSystemConfig.scala

+    configurations.collect({ case c: VertxWriteAdapterConfig => c })
+
+  def addAdapter(first: VertxAdapterConfig, rest: VertxAdapterConfig*): VertxAdapterSystemConfig =
+    addAdapters(first +: rest.toVector)


copy(onfigurations = onfigurations :+ first) or is ordering significant?

Nope, ordering is not significant in this case, but does appending the element have any advantages?

No, shoudn't make a difference

krasserm · 2016-09-19T08:11:23Z

...src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterSystemConfig.scala

+}
+
+class VertxAdapterSystemConfig(private[vertx] val configurations: Seq[VertxAdapterConfig],
+  private[vertx] val codecClasses: Seq[Class[_]]) {


Why private[vertx]?

I didn't want to expose the constructor as part of the public API.

krasserm · 2016-09-19T08:15:21Z

...src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterSystemConfig.scala

+    else
+      invalid.map(c => s"Ambigious definition for adapter with id '${c._1}' given. An id may only be used once.")
+        .toVector
+        .left


This generates n error messages that only differ in ${c._1}. Why not just collecting keys with more than one config and create a single error message? What about using just Either instead of Scalaz's \/?

krasserm · 2016-09-19T08:34:17Z

...vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterConfig.scala

+case class VertxSendAdapterConfig(id: String, log: ActorRef, endpointRouter: VertxEndpointRouter, deliveryMode: DeliveryMode) extends VertxReadAdapterConfig
+case class VertxWriteAdapterConfig(id: String, log: ActorRef, endpoints: Seq[String], filter: PartialFunction[Any, Boolean]) extends VertxAdapterConfig
+
+sealed trait ConfirmationType {}


{} can be removed

krasserm · 2016-09-19T08:34:24Z

...vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterConfig.scala

+case object Single extends ConfirmationType
+case class Batch(size: Int) extends ConfirmationType
+
+sealed trait DeliveryMode {}


{} can be removed

krasserm · 2016-09-19T08:49:29Z

...ate-adapter-vertx/src/it/scala/com/rbmhtechnology/eventuate/adapter/vertx/ActorStorage.scala

+import org.scalatest.{BeforeAndAfterEach, Suite}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._


Some imports are not used here

krasserm · 2016-09-19T09:22:39Z

...rc/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/AkkaSerializationMessageCodec.scala

+
+class AkkaSerializationMessageCodec(override val name: String)(implicit system: ActorSystem) extends MessageCodec[AnyRef, AnyRef] {
+
+  val serializer = PayloadSerializationExtension(system)


You can avoid creating your own serialization extension by using Akka's SerializationExtension and defining a payload wrapper (preferrably using protobuf) so that you know the class when deserializing with Serialization.deserialize[T](bytes: Array[Byte], clazz: Class[T]).

Was there a special reason not implementing like that?

I wanted to reuse the CommonSerializer as it already uses protobuf to serialize the the payload, but this implementation needen an ExtendedActorSystem which was only available within an Extension

You can access an ExtendedActorSystem from Serialization.

krasserm · 2016-09-19T09:23:42Z

...rc/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/AkkaSerializationMessageCodec.scala

+  override def decodeFromWire(pos: Int, buffer: Buffer): AnyRef = {
+    val payloadLength = buffer.getInt(pos)
+    val payload = buffer.getBytes(pos + Integer.BYTES, pos + Integer.BYTES + payloadLength)
+    serializer.fromBinary(payload).asInstanceOf[AnyRef]


Low level position arithmetic can be avoided with a protobuf-defined payload wrapper.

According to the official Vert.x examples the payload length must be stored in the buffer to recover the part of the buffer that contains the payload in decodeFromWire.

krasserm · 2016-09-19T09:25:04Z

...rc/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/AkkaSerializationMessageCodec.scala

+
+class PayloadSerializationExtension(system: ExtendedActorSystem) extends Extension {
+
+  val serializer = new CommonSerializer(system)


When implementing the previously mentioned approach, the serializer can be moved to AkkaSerializationMessageCodec

krasserm · 2016-09-19T10:26:22Z

...ter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxAdapterSystem.scala

+        vertx.eventBus().registerDefaultCodec(c.asInstanceOf[Class[AnyRef]], AkkaSerializationMessageCodec(c))
+      } catch {
+        case e: IllegalStateException =>
+          system.log.warning(s"An adapter codec for class ${c.getName} was configured, even though a default codec was already registered for this class.")


Shouldn't we propagate this exception to the application?

As far as I remember we said that trying to override an existing message-codec-binding should not result in an error but in a warning.

krasserm · 2016-09-19T10:35:10Z

...apter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxWriteRouter.scala

+    val handler = (msg: Message[Any]) => {
+      routes.foreach { route =>
+        if (route.filter.applyOrElse(msg.body(), (_: Any) => false)) {
+          route.destinationLog ! PersistMessage(msg.body(), msg)


Why not having only a single Message parameter for PersistMessage and VertxWriterextracts the event with msg.body?

yes, a single parameter would suffice - I will change that

krasserm · 2016-09-19T11:25:42Z

...apter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxWriteRouter.scala

+        if (route.filter.applyOrElse(msg.body(), (_: Any) => false)) {
+          route.destinationLog ! PersistMessage(msg.body(), msg)
+        } else {
+          msg.reply(msg.body)


Would an ACK be sufficient (here and in VertxWriter)?

how about returning a processing-state depending on whether the event was persisted in the log or filtered? It might be useful information for the caller and should not have a big serialization/deserialization overhead since it can be modelled as an enum.

krasserm · 2016-09-19T12:29:37Z

...apter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxWriteRouter.scala

+
+  private def installMessageConsumer(endpoint: String, routes: Seq[Route]): MessageConsumer[Any] = {
+    val handler = (msg: Message[Any]) => {
+      routes.foreach { route =>


A message should have only a single reply. Allowing n routes here may erroneously lead to n replies. An endpoint should be restricted to publish to exactly one log.

krasserm · 2016-09-19T12:30:11Z

...apter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxWriteRouter.scala

+
+object VertxWriteRouter {
+
+  case class Route(sourceEndpoint: String, destinationLog: ActorRef, filter: PartialFunction[Any, Boolean])


Rename destinationLog to writer?

krasserm · 2016-09-19T12:37:53Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+  protected def produce[A](address: String, msg: Any, deliveryOptions: DeliveryOptions, handler: Handler[AsyncResult[Message[A]]]): Unit
+
+  def produce(address: String, msg: Any): Unit = {
+    produce[Unit](address, msg, new DeliveryOptions(), ((ar: AsyncResult[Message[Unit]]) => {}).asVertxHandler)


Equivalent DeliveryOptions and handler objects are created with every call. Define them in an outer scope? OTOH I'm not sure if the compiler is clever enough to detect that and optimize.

krasserm · 2016-09-19T12:38:17Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+    produce[Unit](address, msg, new DeliveryOptions(), ((ar: AsyncResult[Message[Unit]]) => {}).asVertxHandler)
+  }
+
+  def produce[A](address: String, msg: Any, timeout: FiniteDuration = 30.seconds)(implicit ec: ExecutionContext): Future[A] = {


Make default timeout configurable?

I've removed the default timeout altogether because it is not used anyway.

krasserm · 2016-09-19T12:42:59Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+  override protected def produce[A](address: String, msg: Any, deliveryOptions: DeliveryOptions, handler: Handler[AsyncResult[Message[A]]]): Unit = {
+    eventBus().publish(address, msg, deliveryOptions)
+    handler.handle(VertxFuture.succeededFuture(new MessageImpl[Any, A]()))
+  }


From a design perspective, shouldn't a MessagePublisher rather offer a publish method without a handler parameter (and use a produce method to delegate to)?

Yes, but the problem with this approach would be, that the MessagePublisher and MessageSender could not be used indistinguishable, as done in the various reader-actors.

This way a reader only has to depend on a MessageProducer and the mixed-in implementation will decide if it is a Publisher or a Sender

In this case you may want to abstract over message delivery somewhere else (e.g. on reader-level) rather than here. Completing an asynchronous handler where the return from a from a method call is sufficient is a code/design smell IMO.

krasserm · 2016-09-19T12:43:32Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+trait MessageSender extends MessageProducer {
+  override protected def produce[A](address: String, msg: Any, deliveryOptions: DeliveryOptions, handler: Handler[AsyncResult[Message[A]]]): Unit = {
+    eventBus().send(address, msg, deliveryOptions, handler)
+  }


send method here? (see also https://github.com/RBMHTechnology/eventuate/pull/310/files#r79381579)

krasserm · 2016-09-19T12:46:44Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+
+  def vertx: Vertx
+
+  protected def produce[A](address: String, msg: Any, deliveryOptions: DeliveryOptions, handler: Handler[AsyncResult[Message[A]]]): Unit


You're trying to abstract over point-to-point and pub-sub messaging here. Does that make sense?

In this case I would say yes, because from a sender point-of-view a publish can be seen as a send with an immediate response.
This enables generic handling of message-sending in the VertxReader

I consider a response semantically different from a send-acknowledgement. Responses are communicated via an asynchronous result handler, send/publish acknowledgements by returning from the send/publish method.

krasserm · 2016-09-19T12:54:17Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxReader.scala

+  }
+}
+
+trait MessageDelivery extends MessageProducer {


For reasons of symmetry to the adapter configuration, MessageDelivery sub-traits should only extend the behavior of MessageSender. At-most-once semantics is implied for MessagePublisher. This is also related to https://github.com/RBMHTechnology/eventuate/pull/310/files#r79381579

krasserm · 2016-09-19T13:02:06Z

Some thoughts regarding naming: shouldn't we rather use

bus writer instead of vert.x reader
log writer instead of vert.x writer

This would make declarations like

trait VertxReader[R, W] extends EventsourcedWriter[R, W] with ...

less confusing. WDYT?

krasserm · 2016-09-27T10:41:24Z

...r-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxEventDispatcher.scala

+    vertx.eventBus().publish(address, evt)
+}
+
+trait VertxEventSender extends VertxEventPublisher with EventSender {


extends VertxEventProducer?

krasserm · 2016-09-27T10:56:25Z

...te-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/VertxSender.scala

+}
+
+private[eventuate] class VertxSender(val id: String, val eventLog: ActorRef, val endpointRouter: EndpointRouter, val vertx: Vertx, val storageProvider: StorageProvider)
+  extends VertxEventDispatcher[Long, Long] with AtMostOnceDelivery with VertxEventSender with SequenceNumberProgressStore {


AtMostOnceDelivery causes the VertxEventDispatcher to publish on the Vert.x bus. The VertxEventSender.send() is never used here. Furthermore, when applying my previous change you anyway must use VertxEventPublisher here for making the code compile.

The problem is that the *Delivery traits are not independent of the VertxEvent* traits. Maybe you design the *Delivery traits in a way that they are only applicable to VertxEventSender. At-least-once delivery for VertxEventPublisher would/should be implied then.

krasserm · 2016-09-27T11:24:22Z

...src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/VertxAdapterSystemConfig.scala

+    if (invalid.isEmpty)
+      Right(configs)
+    else
+      Left(s"Ambigious definition for adapter(s) [${invalid.keys.map(id => s"'$id'").mkString(", ")}] given. " +


krasserm · 2016-09-27T11:31:25Z

...uate-adapter-vertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/LogWriter.scala

+    Props(new LogWriter(id, eventLog))
+}
+
+private[vertx] class LogWriter(val id: String, val eventLog: ActorRef) extends EventsourcedActor {


This actor might take a long time to replay events. Either safe periodically empty snapshots to reduce recovery time or use this new feature to start replay from a custom sequence number. In the latter case, the sequence number must be managed elsewhere.

krasserm · 2016-09-27T13:53:50Z

Awesome contribution @cstub 👍 LGTM after adding API and user documentation.

krasserm · 2016-10-04T13:19:05Z

...ertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/japi/rx/StorageProvider.scala

+import scala.concurrent.{ ExecutionContext, Future, Promise }
+
+/**
+ * A storage provider is used to persist the replication progress of individual `event producers`.


replication progress -> log read progress?

krasserm · 2016-10-04T13:24:12Z

...ertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/EventProducerConfig.scala

+   * Specifies that this `point-to-point event producer` should use `At-Least-Once` delivery semantics.
+   *
+   * @param confirmationType Confirmation type - can either be `Batch` or `Single`.
+   * @param confirmationTimeout Timeout after which events should be redelivered if not confirmation was received.


... if no confirmation ...

krasserm · 2016-10-04T13:24:56Z

...ertx/src/main/scala/com/rbmhtechnology/eventuate/adapter/vertx/api/EventProducerConfig.scala

+
+  /**
+   * Sets the destination event log all received events are persisted to.
+   * An optional filter function can be supplied. Only events passing the function are persisted.


passing the filter ...

krasserm · 2016-10-04T13:32:25Z