pangiole / akka-wamp Goto Github PK

See wamp-proto/wamp-proto#226

Python Sphinx seems to be more powerful than Markdown

I am attempting to use your actor example to subscribe to a production web socket feed. All is well except this exception (refer to trace below). I'm not sure how to reproduce this other than connecting/subscribing and then waiting for the appropriate condition to occur. Very frequently, I have encountered this exception after leaving my actor to run in the background. I can't seem to keep my connection to the production server alive. Thanks again for making this available!

[error] a.w.s.SerializationFlows - Timeout on waiting for new data
akka.wamp.serialization.DeserializeException: Timeout on waiting for new data
    at akka.wamp.serialization.JsonSerialization.make$1(JsonSerialization.scala:47)
    at akka.wamp.serialization.JsonSerialization.deserialize(JsonSerialization.scala:163)
    at akka.wamp.serialization.JsonSerializationFlows$$anonfun$2.apply(JsonSerializationFlows.scala:58)
    at akka.wamp.serialization.JsonSerializationFlows$$anonfun$2.apply(JsonSerializationFlows.scala:53)
    at akka.stream.impl.fusing.Map$$anon$8.onPush(Ops.scala:42)
    at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
    at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:710)
    at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:616)
    at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
    at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:423)
Caused by: java.io.IOException: Timeout on waiting for new data
    at akka.stream.impl.io.InputStreamAdapter$$anonfun$read$1.apply$mcI$sp(InputStreamSinkStage.scala:147)
    at akka.stream.impl.io.InputStreamAdapter$$anonfun$read$1.apply(InputStreamSinkStage.scala:132)
    at akka.stream.impl.io.InputStreamAdapter$$anonfun$read$1.apply(InputStreamSinkStage.scala:132)
    at akka.stream.impl.io.InputStreamAdapter.executeIfNotClosed(InputStreamSinkStage.scala:112)
    at akka.stream.impl.io.InputStreamAdapter.read(InputStreamSinkStage.scala:132)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.loadMore(UTF8StreamJsonParser.java:207)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._parseNumber2(UTF8StreamJsonParser.java:1470)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._parsePosNumber(UTF8StreamJsonParser.java:1378)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:852)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:748)

When the deserialize flow receives a bad incoming message it throws a DeserializeException. The stream could rather drop the offending message and keep the client connected for subsequent messages to come.

This decisions could be implemented with proper Akka Stream Supervision settings.

The actual client.Caller sends a CALL message and stores the correspondent Promise[Result] into a so called pending calls hashmap. Actually, that promise maybe fulfilled (and the pending call entry maybe removed) only if the correspondent RESULT message is later on replied.

I missed to implement an automatic mechanism which breaks the promise and remove the pending entry after a configurable timeout expires. It could be done sending a TIMEOUT message via the Akka ActorSystem scheduler.

For reasons explained in the Akka HTTP docs, it would be good to make also Akka Wamp not explicitly depend on Akka Stream anymore, but rather build it using the provided SBT scope.

Following is an excerpt of the above Akka HTTP docs, adapted to the akka-wamp case:

Users will always have to add a manual dependency to akka-stream. Users will make sure they have chosen and added a dependency to akka-stream when updating to the new version of akka-wamp (veterans may remember this policy from Spray).

tried poloniex example, it doesn work:

21:57:49.293 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisignclass2g2ca [jdk]: Certificate failed: cert = "OU=VeriSign Trust Network,OU=(c) 1998 VeriSign, Inc. - For authorized use only,OU=Class 2 Public Primary Certification Authority - G2,O=VeriSign, Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.306 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisigntsaca [jdk]: Certificate failed: cert = "CN=Thawte Timestamping CA,OU=Thawte Certification,O=Thawte,L=Durbanville,ST=Western Cape,C=ZA" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.307 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in gtecybertrustglobalca [jdk]: Certificate failed: cert = "CN=GTE CyberTrust Global Root,OU=GTE CyberTrust Solutions, Inc.,O=GTE Corporation,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.307 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisignclass3g2ca [jdk]: Certificate failed: cert = "OU=VeriSign Trust Network,OU=(c) 1998 VeriSign, Inc. - For authorized use only,OU=Class 3 Public Primary Certification Authority - G2,O=VeriSign, Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.307 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in equifaxsecureglobalebusinessca1 [jdk]: Certificate failed: cert = "CN=Equifax Secure Global eBusiness CA-1,O=Equifax Secure Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.312 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in thawtepremiumserverca [jdk]: Certificate failed: cert = "1.2.840.113549.1.9.1=#16197072656d69756d2d736572766572407468617774652e636f6d,CN=Thawte Premium Server CA,OU=Certification Services Division,O=Thawte Consulting cc,L=Cape Town,ST=Western Cape,C=ZA" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.312 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisignclass1g2ca [jdk]: Certificate failed: cert = "OU=VeriSign Trust Network,OU=(c) 1998 VeriSign, Inc. - For authorized use only,OU=Class 1 Public Primary Certification Authority - G2,O=VeriSign, Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.312 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisignclass3ca [jdk]: Certificate failed: cert = "OU=Class 3 Public Primary Certification Authority,O=VeriSign, Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.312 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in verisignclass1ca [jdk]: Certificate failed: cert = "OU=Class 1 Public Primary Certification Authority,O=VeriSign, Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.313 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in equifaxsecureebusinessca1 [jdk]: Certificate failed: cert = "CN=Equifax Secure eBusiness CA-1,O=Equifax Secure Inc.,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:49.313 |-WARN in c.t.s.s.ConfigSSLContextBuilder - validateStore: Skipping certificate with weak key size in equifaxsecureca [jdk]: Certificate failed: cert = "OU=Equifax Secure Certificate Authority,O=Equifax,C=US" failed on constraint RSA keySize < 2048, algorithm = RSA, keySize = 1024
21:57:50.714 |-ERROR in a.a.LocalActorRefProvider(akka://default) - guardian failed, shutting down system
akka.wamp.client.ClientException: WebSocket upgrade did not finish because of 'unexpected status code: 502 Bad Gateway'

As mentioned by oberstet in wamp-proto/issue/255, the router should abruptly disconnect offending clients by default.

Therefore, the disconnect-offending-peers = false shall be turned to drop-offending-messages = false

akka {
  wamp {
    router {

      # The boolean switch to drop offending messages (e.g. 
      # not deserializable or against the protocol).
      #
      # By default, offending messages will cause session to be
      # closed and transport to be disconnected. Set this switch on 
      # if you prefer to rather drop offending messages and resume.
      #
      drop-offending-messages = false
    }
  }
}

What a router shall do if client connects and sends any of the offending messages?

• Not deserializable messages!
• Either SUBSCRIBE, PUBLISH, REGISTER or CALL before any HELLO!
• Repeated HELLOs (see #21)!

I cannot read anything about the above scenario on the WAMP specification

This question has been also asked in wamp-proto/issues/255

The below code ,
session.subscribe("BTC_XMR", event => {
println(s"${event.kwargs} -> ${event.args}")
})

shows error as ,

Error:(27, 18) type mismatch;
found : Unit
required: scala.concurrent.Future[akka.Done]
println(s"${event.kwargs} -> ${event.args}")

As client.Session is not an actor, but rather shared amongst future threads, could it be not thread safe?
Please, investigate.

Some people asked for SSL/TLS support to let akka-wamp client to subscribe to POLONIEX brokers.

Details are published in https://poloniex.com/support/api/

It would be good to implement this enhancement as soon as possible alongside with a working example.

As asked by Konstantin Burkalev on the WAMP Forum, payloads could bear arguments either as indexed sequence or key-value pairs

Tobias Oberstein replied that a client shall also expect a mixed payload bearing both forms of arguments (as it can be understood by reading the WAMP protocol specification tough no examples are provided for that specific case)

Therefore, akka-wamp shall provide a better API for mixed payloads

First of all thank you for this very useful package !! May be I am not seeing it but it looks like there is no method in the high level (Futures) API where we can do something like

val to : Future[Timeout] = session.setTimeout( duration=2000)
to.onComplete {
case Sucess(x) => ....
case Failure(y) => ...
}

This is available in the underlying Actor API like this
context.system.scheduler.scheduleOnce( 2 seconds, self, IdleCheckTick)

Immediate Motivation : i wanted to publish something periodically in my akka-wamp app

Akka HTTP documentation advice to reactToConnectionFailure as explained at the bottom of the following page

Low-Level Server-Side API.

But it doesn't work as you try to connect and disconnect to our akka-wamp router by using a telnet program as client

Make the following registration possible

  for {
      session <- client
        .openSession(
          url = "ws://localhost:8080/ws",
          realm = "myapp.realm")
      registration <- session
        .register(
          procedure = "myapp.procedure.sum",
          handler = (a: Int, b: Int) => a + b)
    } 
    yield ()

The procedure handler is given just as one of the scala.Function types

Provide a Client.connect() with timeout to cope with unresponsive routers

Add the following TypeSafe configuration item for our akka-wamp library

akka {
  wamp {
    serialization {
      # The boolean switch (default is false) to validate against strict URIs 
      # rather than loose URIs
      #
      validate-strict-uris = false
    }}}

So that, it can be configured to check against strict or loose URI as described in the specification section 5.1.1

This seems to not be implemented right now. The current implementation:
throw new DeserializeException("Cannot deserialize binary message as JSON message was expected instead!")

Needs to be:
source.runFold("")(_ + )(materializer)
.map( json.deserialize()(validator, materializer) )(materializer.executionContext)

As transport disconnects (e.g. mobile devices experience disconnections quite often) the session held on the router side (with its subscription/registration entries) is lost. In fact, routers are more likely to remove session entries upon disconnections. AkkaWamp router is one of those.

Though a client, after some re-attempts, could successfully reconnect to the same router, it will have to open a new session and explicitely restore all of its previous subscriptions/registrations. That's how the majority of client APIs are asking their users to do. Clients are notified of "disconnected" events (e.g. via RxObservables or similar mechanisms) so that they can attempt restoring their session themselves.

AkkaWamp users could enjoy an "automatic session restoring" feature by default. Restoring subscriptions/registrations could be performed by AkkaWamp itself rather than put in charge to users. That behaviour could be configurable:

akka.wamp.client {

  # The boolean switch to disable automatic session restoring of all
  # subscriptions/registrations obtained until a transport disconnection
  # event occurs
  #
  manual-session-restoring = false

}

Hi,

When i run PoloniexScalaClient, it runs fine but not consuming any data.
It never prints "${event.kwargs} -> ${event.args}"

Please suggest me what to do.

Thanks.

https://github.com/angiolep/akka-wamp/blob/master/build.sbt#L49-L50

it should point to this repo

I tried but ... macros don't compile for Scala 2.11

Deliberate disconnect of a WAMP connection has not been provided yet as it seems quite difficult to implement with the actual Akka HTTP version (2.4.10 at the time of writing)

The actual client.Manager makes use of the following Akka Stream:

val (outgoingActor, upgradeResponse) =
   outgoingSource
     .via(serializationFlows.serialize)
     .viaMat(webSocketFlow)(Keep.both)
     .viaMat(serializationFlows.deserialize)(Keep.left)
     .toMat(incomingSink)(Keep.left)
     .run()

which would have caused TCP socket disconnection if we had inserted a custom stage at the beginning of the stream meant to force completion upon processing a Wamp.Disconnect command.

Please read

http://doc.akka.io/docs/akka/2.4.10/scala/stream/stream-customize.html#graphstage-scala

for further details

From release v0.8.0 the following configuration entry has been provided in reference.conf (a.k.a. the default TypeSafe configuration file the Akka ActorSystem loads during its initialization):

akka {
  wamp {
     client {
      # The boolean switch to disconnect those peers that send 
      # offending messages (e.g. not deserializable or causing
      # session failures)
      #
      # By default, offending messages are just dropped and 
      # the router resumes processing next incoming messages
      #
      disconnect-offending-peers = false
}}}

so that an Akka Wamp client created via the following factory method on the companion object:

import akka.wamp.client._
val client = Client()

• will make use of a Supervision.Resume strategy on DeserializationException
• and preserve the connection though any kind of session failure happens (such as SUBSCRIBE before open session or repeated HELLOs)

This enhancement is meant to provide a better factory method which would allow to create more client instances each of those configured with different settings (and different disconnect-offending-peers boolean switches).

val client1 = Client1() // default settings

val config2 = ...
val client2 = Client(config2)

For example, those clients configured with disconnect-offending-peers switched on

• will make use of a Supervision.Stop strategy on DeserializationException
• and disconnect as soon as any kind of session failure happens (such as SUBSCRIBE before open session or repeated HELLOs)

Router shall be configured as follows:

akka {
  wamp {
    router {
       transport {
          # Decision to be made on DeserializeException can be 
          # one of the following two alternatives:
          # 
          #   - resume
          #      The offending message is dropped and the 
          #      stream continues to process the next one
          #   
          #   - stop
          #      The transport is stopped and client disconnected
          #
          deserialize-exception = resume
        }}}}

Akkawamp 0.15.0 is not compatible with akka 2.5.2. Would it be possible to upgrade akkawamp to make it compatible with 2.5.2

What a router shall do when it receives a second HELLO message during
the lifetime of a session?

I can read from wamp-proto specification section-7.1.1.

> It is a protocol error to receive a second "HELLO" message during the
> lifetime of the session and the _Peer_ must fail the session if that
> happens.

But, what "fail the session" really means?

Does it mean the router shall dispose subscriptions/registrations of that client
and then reply with a GOODBYE("wamp.error.session_failure") message?

Or does it mean the router shall dispose subscriptions/registrations of that client
and then abruptly disconnect the transport giving no chance to the client to
understand what just happened?

router.ConnectionHandlerSpec

Test1

• The default router.ConnectionHandler should complete publish/subscribe in optimal scenario BUT I haven't yet found a way to verify EVENT messages are correctly sent/received upon PUBLISH

Test2

• The default router.ConnectionHandler should handle sessions in optimal scenario BUT I haven't yet found a way to programmatically make the router send GOODBYE

client.TranportSpec

Test3

• The client.TransportSpec should fail open session when it turns out to have been disconnected from router side BUT I haven't yet found a way to programmatically disconnect from router side.

client.SessionSpec

Test4

• A client session should close and reply GOODBYE upon receiving GOODBYE from router BUT I haven't yet found a way to programmatically make the router send GOODBYE

The actual Akka Wamp subscriber will just override the latest event handler with the most recent one given in case of multiple SUBSCRIBE to the same topic.

While the Akka Wamp router is compliant with the spec, the actual Akka Wamp client is buggy as you can understand from

• https://groups.google.com/forum/#!msg/wampws/kC878Ngc9Z0/bGRTgJWyiLYJ
• wamp-proto/wamp-proto#142

Following is an excerpt from the release candidate specification:

A Subscriber may have more than one event handlers for the same topic, that can be implemented in different ways:

• Subscriber can be smart enough to be aware, that it is already subscribed to such topic, and just add another handler for incoming events.
• Or it can simply send SUBSCRIBE message to broker (as it would be first).

Some additional hints to fix this issue are coming from ecorm who commented on Mar 15, 2015:

On the client side, if you're dealing with multiple subscriptions to the same topic, then you must have some kind of collection to keep track of all the handlers associated with a particular topic. You already have the reference count by virtue of that handler collection. It's then a trivial matter to check if that collection becomes empty when a "local" unsubscribe occurs. If the handler collection becomes empty, then you issue an UNSUBSCRIBE message to the router

This is really very simple issue. I downloaded the router 0.15.0 tar.gz file
unzipped to some $HOME/akka-wamp
then edited the conf/application.conf to change the port number from 8080 to 1080
Then I run bin/akka-wamp -Dakka.loglevel=DEBUG
It is not reading my conf changes - instead it still runs on port 8080 ?

Set(RSA)
16:46:11.983 |-DEBUG in a.i.TcpListener - Successfully bound to /127.0.0.1:8080
16:46:11.994 |-DEBUG in a.w.r.Binder - Bound to ws://127.0.0.1:8080/wamp

What am I missing here - This seems like very very elementary ?
Thanks a lot

Currently the case Status.Failure is not handled in handleConnected receive method in ConnectionHandler.scala.

In case a failure occurs ( e.g. a server error ) it leaves the ConnectionHandler in a broken state without reporting the connector. I guess it should send a Disconnect to connector and PoisonPill to self.

I can replicate/trigger this by running both client and server on a laptop and suspending and resuming.

The server will fail with:

[akka.actor.ActorSystemImpl(modulair)] Websocket handler failed with Buffer overflow (max capacity was: 4)!
akka.stream.BufferOverflowException: Buffer overflow (max capacity was: 4)!
	at akka.stream.impl.ActorRefSourceActor$$anonfun$receiveElem$1.applyOrElse(ActorRefSourceActor.scala:91)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
	at akka.actor.Actor.aroundReceive(Actor.scala:517)
	at akka.actor.Actor.aroundReceive$(Actor.scala:515)
	at akka.stream.impl.ActorRefSourceActor.akka$stream$actor$ActorPublisher$$super$aroundReceive(ActorRefSourceActor.scala:29)
	at akka.stream.actor.ActorPublisher.aroundReceive(ActorPublisher.scala:329)
	at akka.stream.actor.ActorPublisher.aroundReceive$(ActorPublisher.scala:272)
	at akka.stream.impl.ActorRefSourceActor.aroundReceive(ActorRefSourceActor.scala:29)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:590)
	at akka.actor.ActorCell.invoke(ActorCell.scala:559)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
	at akka.dispatch.Mailbox.run(Mailbox.scala:224)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
	at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
	at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
	at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

The actual failure is:

akka.http.scaladsl.model.ws.PeerClosedConnectionException: Peer closed connection with code 1011 'internal error'

But the client/connector will never get notified of a disconnect/failure

As per WAMP specification section 6.1

Keys in "Options" and "Details" MUST be of type "string"
and MUST match the regular expression "[a-z][a-z0-9_]{2,}" for WAMP
predefined keys. Implementations MAY use implementation-specific
keys that MUST match the regular expression "[a-z0-9]{3,}".
Attributes unknown to an implementation MUST be ignored.

Provide connection resilience by attempting a reconnection as soon as the transport disconnect.

Client could automatically attempt to reconnect to the same router and reopen a new session to subscribe (or register) the same topics (or procedures).

An urgent fix is required as RPC message types won't be validate correctly

The following specification excerpt

The role|dict is a dictionary describing features supported by the peer for that role.
This MUST be empty for WAMP Basic Profile implementations, and MUST be used by implementations implementing parts of the Advanced Profile to list the specific set of features they support.

Refers to what a client shall do to announce features it supports. It does NOT mean a Basic Profile router shall not validate feature announcements.

You can register a procedure URI with a partially applied function as handler.

val registration: Future[Registration] = session.flatMap(
  _.register(
    procedure = "myapp.procedure.sum",
    handler = sum _ ))

Akka Wamp will dinamically invoke the partially applied function you provide (e.g. sum _) upon receiving invocations addressed to the registered procedure URI (e.g. "myapp.procedure.sum"). It will lazily deserialize the args list from incoming payloads by using its default type bindings.

Notice
Lazy deserialization from incoming kwargs and dynamic invocation passing input values to named arguments is not yet supported

As per WAMP specification section 6.3, the parser should inspect the incoming messages by using a stream-based approach.

For JSON messages, you could use Faster Jackson Streaming Parser. It will improve performances as the parser could entirely skip the incoming payload.

Actual router behaviour is:

"Routing occurs between sessions that have joined any realm"

Expected behaviour shall rather be:

"Routing occurs between sessions that have joined the same realm"

The WAMP spec: states the format of the Events should be one of:

[EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id,Details|dict] or
[EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id,Details|dict,PUBLISH.Arguments|list] or
[EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id,Details|dict,PUBLISH.Arguments|list, PUBLISH.ArgumentKw|dict]

It's similar for CALL, INVOCATION etc.

There is currently no way to parse an EVENT that contains both Arguments|list along with ArgumentKw|dict

In case they are both present PUBLISH.ArgumentKw|dict is taken with precedence and PUBLISH.Arguments|list is not even present in the Payload.

A snippet from the JsonSerialiser:

case EVENT => {
          arr.length match {
          .....
            case 6 => Event(subscriptionId = arr(1).asId, publicationId = arr(2).asId, details = arr(3).asDict, arr(5).asSomePayload)
          }
        }

Will be good as other WAMP implementations to support messages where both parts of the Payload are present.

A potential test that should pass is:

  "succeed for valid EVENT with payload both as string and as map" in {
        s.deserialize(s"""[36,1,2,{},[{"arg0":"paolo","age":40,"arg2":true}, {"arg1":"paolo","arg2":40,"arg3":true}],{"name":"paolo","age":40,"human":true}]""") match {
          case m: Event =>
            m.subscriptionId mustBe 1
            m.publicationId mustBe 2
            m.details mustBe empty
            m.payload.value.arguments mustBe List(
              Map("arg0"->"paolo", "age"->40, "arg2"->true),
              Map("arg1"->"paolo", "arg2"->40, "arg3"->true)
            )
            m.payload.value.argumentsKw mustBe Map(
              "name"->"paolo",
              "age"->40,
              "human"->true
            )
          case _ => fail
        }
      }

I can potentially provide a pull request that solves the issue.