[akka-user] Testing custom FlexiRoute

[Ivan Baisi]

I'm trying to test a FlexiRoute I've developed that dpeending on the type 
of element it receives, it routes it either the keep alive signal or the 
gnip activity to different flows. Here's the approach I'm trying:

class ActivityFlexiRouterSpec extends FlatSpec with GivenWhenThen{
  implicit val system = ActorSystem("TEST")
  implicit val materializer = 
ActorMaterializer(ActorMaterializerSettings(system))
  behavior of "ActivityFlexiRouter"

  it should "route the KeepAliveSignal to the KeepAliveSignalFlow" in {
Given("A FlowGraph an ActivityFlexiRouter and a KeepAliveSignal")
val newFlowProbe = TestSink.probe[GnipActivity]
val keepAliveFlowProbe = TestSubscriber.manualProbe[GnipActivity]
val graph = FlowGraph.closed() { implicit b =>
  val route = b.add(new ActivityFlexiRouter)

  Source.single(KeepAliveSignal) ~> route.in
  route.newActivityFlow ~> newFlowProbe
  route.keepAliveSignalFlow ~> keepAliveFlowProbe
}

When("the graph is run")
graph.run()

Then("assert something")
assert(keepAliveFlowProbe.)
  }

}

I must say neither of the newFlowProbe or keepAliveFlowProbe are working, 
but I wanted to give some idea of the couple of the things I've tried. 
Could anyone give me any hint on how to test this?

Thanks!

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] akka-http non standard return codes.

[Tomasz Kogut]

Hello,

this is my first post here so let me start with a big "Hello" to everyone 
in this fantastic community :)

As for the question I have a server to which I send request using 
akka-http. The problem this is a Microsoft IIS that is using rare response 
code:
  
HTTP/1.1 440 ..Content-Length: 48..Content-Type: application/json; 
charset=utf-8..Server: Microsoft-HTTPAPI/2.0..Date: Thu, 20 Aug 2015 
09:01:11 GMT..Connection: close{..  "Message": "Ticket expired. Or 
invalid."..}

this causes akka to throw exception:
akka.http.scaladsl.model.IllegalResponseException: Illegal response status 
code

and of course fails the Future so I don't really know what was the cause of 
the failure. Is there a way to somehow overcome this limitation?


--
BR,
Tomasz

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Event sourcing backend/plugin with scalable pub/sub

[Konrad Malawski]

I'm tempted to write a service that would use Kafka and Cassandra as backends 
and expose subscriptions per entity. Not sure yet if it's doable at scale.
An "exercise left for the ambitious reader" one might say ;-)

Capability wise definitely makes sense to me and we've seen such setups "in the 
field".



Can you point me to a link on this subject?
I don't know if / where it's been announced as part of the roadmap or if it's 
"future research". 

I've heard about it from some Cassandra people during a conference (GeeCON.org).



-- Konrad

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] [akka-streams] How to share extensions?

[Juan José Vázquez Delgado]

Hi all,

In my development shop we're currently having to develop or expand some 
Akka Streams features in the form of sources, flows and sinks. Some of this 
building blocks could be shared, for instance a source that deals with 
zipped files containing other zipped files, etc. However, we don't know if 
there're some templates or at least best practices that we should follow in 
order to share this work.

In other words, is it planned to have, in the short term, some kind of 
sandbox or extension project?. In the past, I've been comfortable for 
instance sharing extensions with the Apache Camel team where you have some 
kind of standard way to do this.

Thanks in advance, 

Juanjo

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] [akka-streams] Http source design question?

[Konrad Malawski]

Hi John,
good observation, however this has not shown up in our initial benchmarking
as a bottle-neck so far AFAIR.
We're taking a systematic aproach and first improving the perf of the
"biggest bang for the buck" elements of Akka Http.
Currently this means we're focusing on fusing and need to implement fusing
of fan-in and fan-out stages, which will save a lot of actor creation
during request handling.

>From there we'll benchmark again and see what then becomes the bottleneck
:-)

If you'd like to help out, we'd definitely welcome contributions (backed
with benchmarks in case of perf improvements).

Thanks!

On Sat, Aug 15, 2015 at 1:01 PM,  wrote:

> Is this source sound for a "flow" based on Http().superPool(...) :
>
> BlockingQueue queue = new LinkedBlockingQueue<>();
>
> Source< HttpRequest, ?> source = Source.fromIterator(() ->
>   new Iterator() {
>  public boolean hasNext() {return true;}
>
>  public HttpRequest next() {
> try {
>return queue.take();
> } catch (InterruptedException e) {}
> return null;
>  }
>  });
>
>
> So many clients will be adding async many  HttpRequests to the single
> queue and the flow (based on Http().superPool(...)) should then process
> theses requests.
>
> I do think that using a concurrent LinkedBlockingQueue does scale. But is
> there an alternative?
>
>
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>



-- 
Cheers,
Konrad 'ktoso' Malawski
Akka  @ Typesafe 

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: akka-http non standard return codes.

[André]

Hi Tomasz,

I think akka.http.ParserSettings.withCustomStatusCodes is what you're 
looking for. Its spec is here 

.

HTH
André

On Thursday, August 20, 2015 at 11:10:55 AM UTC+2, Tomasz Kogut wrote:
>
> Hello,
>
> this is my first post here so let me start with a big "Hello" to everyone 
> in this fantastic community :)
>
> As for the question I have a server to which I send request using 
> akka-http. The problem this is a Microsoft IIS that is using rare response 
> code:
>   
> HTTP/1.1 440 ..Content-Length: 48..Content-Type: application/json; 
> charset=utf-8..Server: Microsoft-HTTPAPI/2.0..Date: Thu, 20 Aug 2015 
> 09:01:11 GMT..Connection: close{..  "Message": "Ticket expired. Or 
> invalid."..}
>
> this causes akka to throw exception:
> akka.http.scaladsl.model.IllegalResponseException: Illegal response status 
> code
>
> and of course fails the Future so I don't really know what was the cause 
> of the failure. Is there a way to somehow overcome this limitation?
>
>
> --
> BR,
> Tomasz
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] [akka-streams] Http source design question?

[Johannes Rudolph]

Hi john,

AFAIU your question was not about existing code in akka-http but how to put 
a queue before a superpool, right? And no, putting a blocking queue inside 
of an Iterator is not likely the best solution because it will spend one 
thread permanently that is blocking in `queue.take` almost all the time. 
akka-http already provides `Http.singleRequest` which implements exactly 
what you are looking for so that should be your default entry point for 
running single requests. 

If this isn't enough (the best example IMO would be the need to provide 
custom queuing logic) akka-stream should provide a general facility that 
allows users to put a queue around one-to-one flows (with or without 
context).

Johannes

On Thursday, August 20, 2015 at 11:56:06 AM UTC+2, Konrad Malawski wrote:
>
> Hi John, 
> good observation, however this has not shown up in our initial 
> benchmarking as a bottle-neck so far AFAIR.
> We're taking a systematic aproach and first improving the perf of the 
> "biggest bang for the buck" elements of Akka Http.
> Currently this means we're focusing on fusing and need to implement fusing 
> of fan-in and fan-out stages, which will save a lot of actor creation 
> during request handling.
>
> From there we'll benchmark again and see what then becomes the bottleneck 
> :-)
>
> If you'd like to help out, we'd definitely welcome contributions (backed 
> with benchmarks in case of perf improvements).
>
> Thanks!
>
> On Sat, Aug 15, 2015 at 1:01 PM, > wrote:
>
>> Is this source sound for a "flow" based on Http().superPool(...) :
>>
>> BlockingQueue queue = new LinkedBlockingQueue<>();
>>
>> Source< HttpRequest, ?> source = Source.fromIterator(() ->
>>   new Iterator() {
>>  public boolean hasNext() {return true;}
>>
>>  public HttpRequest next() {
>> try {
>>return queue.take();
>> } catch (InterruptedException e) {}
>> return null;
>>  }
>>  });
>>
>>
>> So many clients will be adding async many  HttpRequests to the single 
>> queue and the flow (based on Http().superPool(...)) should then process 
>> theses requests.
>>
>> I do think that using a concurrent LinkedBlockingQueue does scale. But 
>> is there an alternative?
>>
>>
>>
>> -- 
>> >> Read the docs: http://akka.io/docs/
>> >> Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>> >> Search the archives: https://groups.google.com/group/akka-user
>> --- 
>> You received this message because you are subscribed to the Google Groups 
>> "Akka User List" group.
>> To unsubscribe from this group and stop receiving emails from it, send an 
>> email to akka-user+...@googlegroups.com .
>> To post to this group, send email to akka...@googlegroups.com 
>> .
>> Visit this group at http://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>>
>
>
>
> -- 
> Cheers,
> Konrad 'ktoso' Malawski
> Akka  @ Typesafe 
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] How to implement etop for Akka?

[Eax Melanhovich]

Hello.

There is a great tool in Erlang world called etop. It shows top N
actors and corresponding message queue length. Looks like this:

http://eax.me/files/2013/07/etop.png

I wonder does anyone know a good way to implement an HTTP endpoint
which shows some things similar? Current stack traces are not required,
only actor name and queue length will suffice. As I understand this is
not as simple as it may look, because there are stash(), become() and
other things that should be considered.

Best idea a have so far is to use Kamon and implement my own StatsD
server which would aggregate top N actors. BTW I'm not sure if Kamon
handles stash() properly.

Are there any better ideas?

-- 
Best regards,
Eax Melanhovich
http://eax.me/

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] [Akka-Streams] Testing a Partial Flow Graph

[Gabriel Volpe]

Hi all, 

Finally I found the way to do it. It wasn't easy to understand the API for 
the FlowGraph.closed() 

 
function, but taking a deeper look I understood that it needs a 
*combinedMat* value in addition to a Shape like the collector in the 
example. This is how the method is defined in the FlowGraph object:

def closed[Mat, M1, M2](g1: Graph[Shape, M1], g2: Graph[Shape, M2])(
combineMat: (M1, M2) ⇒ Mat)(buildBlock: (Builder[Mat]) ⇒ (Graph.Shape, Graph
.Shape) ⇒ Unit): RunnableGraph[Mat]

And regarding the example, now it's working doing this:

val (out0, out1) = FlowGraph.closed(collector[Event], collector[Event])((m1, 
m2) => (m1, m2)) { implicit builder => (out0, out1) =>
  val eif = builder.add(eventInputFlow)
  Source.single(someEvent) ~> eif.in
  eif.out(0) ~> out0  
  eif.out(1) ~> out1
}.run()

Cheers,
Gabriel.

El lunes, 17 de agosto de 2015, 11:34:05 (UTC+1), Gabriel Volpe escribió:
>
> Thanks for your response Roland,
>
> I've been trying to run the test as you suggested but I had no luck. The 
> test works if I do something like this (I'm testing only the first output):
>
> val out = FlowGraph.closed(collector[Event]) { implicit builder => out =>
>   val eif = builder.add(eventInputFlow)
>   Source.single(someEvent) ~> eif.in
>   eif.out(0) ~> out
>   eif.out(1) ~> Sink.ignore
> }.run()
>
> But if I try to do it in this way I get a compilation error:  
>
> val (out0, out1) = FlowGraph.closed(collector[Event], collector[Event]) { 
> implicit builder => (out0, out1) =>
>   val eif = builder.add(eventInputFlow)
>   Source.single(someEvent) ~> eif.in
>   eif.out(0) ~> out0
>   eif.out(1) ~> out1
> }.run()
>
> This is the error:
>
> error: missing parameter type
> [ERROR]   val (out0, out1) = FlowGraph.closed(collector[Event], 
> collector[Event]) { implicit builder => (out0, out1) =>
>
> And BTW I'm using the version 1.0 of Akka Streams.
>
> El sábado, 15 de agosto de 2015, 7:57:52 (UTC+1), Akka Team escribió:
>>
>> Hi Gabriel,
>>
>> testing such a Graph as a black box is rather simple: just attach a 
>> Source that emits the test inputs and attach a Sink to each output that 
>> collects all emitted elements (e.g. using 
>> Flow[...].grouped(1000).toMat(Sink.head)(Keep.right)). When running that 
>> flow, the output Futures will eventually be completed and you can assert 
>> that they contain the right collections.
>>
>> def collector[T] = Flow[T].grouped(1000).toMat(Sink.head)(Keep.right)
>> val (out0, out1) = FlowGraph.closed(collector[Event], collector[Event]) {
>>   implicit builder => (out0, out1) =>
>> val eif = builder.add(eventInputFlow)
>> Source(ev1, ev2, ev3) ~> eif.in
>> eif.out(0) ~> out0
>> eif.out(1) ~> out1
>> }.run()
>>
>> Regards,
>>
>> Roland
>>
>> On Thu, Aug 13, 2015 at 1:44 PM, Gabriel Volpe  
>> wrote:
>>
>>> Hi Viktor,
>>>
>>> I've been reading the official docs (
>>> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-testkit.html)
>>>  
>>> and researching into the scala docs but it seems there is no way to test 
>>> neither a FanInShape or a FanOutShape. The examples in the documentation 
>>> are trivial (only testing a simple Flow).
>>>
>>> I tried to wrap the FanOutShape into two Partial Flow Shapes using 
>>> Out(0) for one and Out(1) for the other one. But when I try to use this 
>>> FlowShape (using Flow.wrap(theFlowShape)) connected to a Source and a Sink 
>>> I got an error like this one:
>>>
>>> Element must not be null, rule 2.13
>>> java.lang.NullPointerException: Element must not be null, rule 2.13
>>> at 
>>> akka.stream.impl.ReactiveStreamsCompliance$.elementMustNotBeNullException(ReactiveStreamsCompliance.scala:44)
>>> at 
>>> akka.stream.impl.ReactiveStreamsCompliance$.requireNonNullElement(ReactiveStreamsCompliance.scala:70)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.push(Interpreter.scala:287)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.push(Interpreter.scala:434)
>>> at akka.stream.impl.fusing.Map.onPush(Ops.scala:24)
>>> at akka.stream.impl.fusing.Map.onPush(Ops.scala:23)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.run(Interpreter.scala:436)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.progress(Interpreter.scala:245)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.progress(Interpreter.scala:434)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter.akka$stream$impl$fusing$OneBoundedInterpreter$$execute(Interpreter.scala:580)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.execute(Interpreter.scala:241)
>>> at 
>>> akka.stream.impl.fusing.OneBoundedInterpreter$EntryState.execute(Interpreter.scala:666)
>>> at akka.stream.stage.AbstractStage.enterAndPush(Stage.scala:66)
>>> at 
>>> akka.stream.impl.fusing.Batchin

[akka-user] Re: Testing custom FlexiRoute

[Ivan Baisi]

If anyone is interested in the solution to this, I managed to find a way of 
doing it. It's probably not the optimal solution, but here it goes. 
Comments will be appreciated :)

it should "route the KeepAliveSignal and the NewActivity to proper flows" in 
{
Given("A Flow with an ActivityFlexiRouter, a KeepAliveSignal and a 
merger")
var keepAliveCounter = 0
var newTweetCounter = 0
val flow = Flow() { implicit b =>
  import FlowGraph.Implicits._
  val route = b.add(new ActivityFlexiRouter)
  val merge = b.add(Merge[GnipActivity](2))

  route.newActivityFlow ~> Flow[GnipActivity].map(x => {newTweetCounter 
= newTweetCounter + 1; x}) ~> merge.in(0)
  route.keepAliveSignalFlow ~> Flow[GnipActivity].map(x => 
{keepAliveCounter 
= keepAliveCounter + 1; x}) ~> merge.in(1)

  (route.in, merge.out)
}

When("the flow is materialized and run")
val (pub, sub) = TestSource.probe[GnipActivity]
  .via(flow)
  .toMat(TestSink.probe[GnipActivity])(Keep.both)
  .run()

Then("the keep alive flow receives the element")


sub.request(4)

pub.sendNext(KeepAliveSignal)
pub.sendNext(KeepAliveSignal)

pub.sendNext(NewTweetActivity(Tweet("nId", "l", "b", DateTime.now(), 0L, 
0L, None, None, "a", UUID.randomUUID()), Vector(),
  TweetAuthor("nIdA", "aName", "img")))

pub.sendNext(KeepAliveSignal)

pub.sendNext(NewTweetActivity(Tweet("nId", "l", "b", DateTime.now(), 0L, 
0L, None, None, "a", UUID.randomUUID()), Vector(),
  TweetAuthor("nIdA", "aName", "img")))

Thread.sleep(300)

assert(keepAliveCounter == 3)

assert(newTweetCounter == 2)

  }




On Thursday, August 20, 2015 at 9:49:50 AM UTC+2, Ivan Baisi wrote:
>
> I'm trying to test a FlexiRoute I've developed that dpeending on the type 
> of element it receives, it routes it either the keep alive signal or the 
> gnip activity to different flows. Here's the approach I'm trying:
>
> class ActivityFlexiRouterSpec extends FlatSpec with GivenWhenThen{
>   implicit val system = ActorSystem("TEST")
>   implicit val materializer = 
> ActorMaterializer(ActorMaterializerSettings(system))
>   behavior of "ActivityFlexiRouter"
>
>   it should "route the KeepAliveSignal to the KeepAliveSignalFlow" in {
> Given("A FlowGraph an ActivityFlexiRouter and a KeepAliveSignal")
> val newFlowProbe = TestSink.probe[GnipActivity]
> val keepAliveFlowProbe = TestSubscriber.manualProbe[GnipActivity]
> val graph = FlowGraph.closed() { implicit b =>
>   val route = b.add(new ActivityFlexiRouter)
>
>   Source.single(KeepAliveSignal) ~> route.in
>   route.newActivityFlow ~> newFlowProbe
>   route.keepAliveSignalFlow ~> keepAliveFlowProbe
> }
>
> When("the graph is run")
> graph.run()
>
> Then("assert something")
> assert(keepAliveFlowProbe.)
>   }
>
> }
>
> I must say neither of the newFlowProbe or keepAliveFlowProbe are working, 
> but I wanted to give some idea of the couple of the things I've tried. 
> Could anyone give me any hint on how to test this?
>
> Thanks!
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] [akka-streams] ActorSubscriber to ActorPublisher flow -- how to?

[Alexey Shuksto]

Hello there,

I could not find answer to this particular question neither in docs nor in 
this group:

Is it possible to wire Sink.actorSubscriber(...) into 
Source.actorPublisher(...) in one (Partial-) Flow? And by 'wire' I mean to 
somehow send message from ActorSubscriber, instantiated from given props, 
to the 'same flow' instance of ActorPublisher?

I was thinking about giving some unique Id to both of them and then 
publish/subscribe events through `context.system.eventStream`, but then I 
realized that all materialized flows would have same Id and messages would 
mess up... :(

In fact, I'm interested in even more simple task -- create an Actor, that 
could act as PartialFlow[In, Out]: when you push a value into flow, actor 
receives it, processes and sends further down the line.

Of course, I always could do something like Flow[In] mapAsync (data => 
(actor ? data).mapTo[classOf[Out]]) -- but I would really, really prefer to 
do this message way.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] [akka-streams] ActorSubscriber to ActorPublisher flow -- how to?

[Konrad Malawski]

Hello there Alexey,
I'm not sure I understand what you want to achieve, is it simply to connect 
such Sink to the Source?
That's simply: Source.actorPublisher(...).to(Sink.actorSubscriber(...)).run()

Or do you mean to put another processing stage between them?
If so then read about custom processing stages: 
doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-customize.html


-- 
Cheers,
Konrad Malawski
Akka @ Typesafe

On 20 August 2015 at 17:12:55, Alexey Shuksto (seig...@gmail.com) wrote:

Hello there,

I could not find answer to this particular question neither in docs nor in this 
group:

Is it possible to wire Sink.actorSubscriber(...) into 
Source.actorPublisher(...) in one (Partial-) Flow? And by 'wire' I mean to 
somehow send message from ActorSubscriber, instantiated from given props, to 
the 'same flow' instance of ActorPublisher?

I was thinking about giving some unique Id to both of them and then 
publish/subscribe events through `context.system.eventStream`, but then I 
realized that all materialized flows would have same Id and messages would mess 
up... :(

In fact, I'm interested in even more simple task -- create an Actor, that could 
act as PartialFlow[In, Out]: when you push a value into flow, actor receives 
it, processes and sends further down the line.

Of course, I always could do something like Flow[In] mapAsync (data => (actor ? 
data).mapTo[classOf[Out]]) -- but I would really, really prefer to do this 
message way.
--
>> Read the docs: http://akka.io/docs/
>> Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] How to implement etop for Akka?

[Konrad Malawski]

Hi Eax,
I think that re-using an implementation (Kamon for example) for the collection 
of data is a good first step,
you'll want to avoid re-inventing the wheel there (and the horrible pains of 
weaving instrumentation into the codebase manually ;-)).

Perhaps it's possible to re-use Kamon's infrastructure more directly, without 
going through statsd?
You could ask on their mailing list for some hints I guess.

Best idea a have so far is to use Kamon and implement my own StatsD 
server which would aggregate top N actors. BTW I'm not sure if Kamon 
handles stash() properly. 
Stash is like a separate buffer, it is not the mailbox size, thus it should not 
be reported directly as something in the mailbox, report it separately I'd say.



-- Konrad

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] [akka-streams] ActorSubscriber to ActorPublisher flow -- how to?

[Alex Shuksto]

Well, I want to simply connect them, but I want to connect Sink
.actorSubscriber(...) ~> Source.actorPublisher(...), not other way around,
like you said.

I've got Actor, that enriches HTTP Requests that our server receives. After
some enrichment, I need to be able to send request further down for
processing:

```
val source = Source.single(/*HttpRequest*/)
val sink = Sink.head[ProcessedRequest]

val enrichment: Flow[HttpRequest, EnrichedRequest] = ???
val processing: Flow[EnrichedRequest, ProcessedRequest] = ???

source.via(enrichment).via(processing).to(sink)
```

I want to be able to use our enrichment Actor in Flow[HttpRequest,
EnrichedRequest] stage. Actor is able to receive HttpRequest and reply (or
forward) ProcessedRequest, but I don't know how to create Flow from it.





чт, 20 авг. 2015 г. в 18:20, Konrad Malawski :

> Hello there Alexey,
> I'm not sure I understand what you want to achieve, is it simply to
> connect such Sink to the Source?
> That's simply:
> Source.actorPublisher(...).to(Sink.actorSubscriber(...)).run()
>
> Or do you mean to put another processing stage between them?
> If so then read about custom processing stages:
> doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-customize.html
>
>
> --
> Cheers,
> Konrad Malawski
> Akka @ Typesafe
>
> On 20 August 2015 at 17:12:55, Alexey Shuksto (seig...@gmail.com) wrote:
>
> Hello there,
>
> I could not find answer to this particular question neither in docs nor in
> this group:
>
> Is it possible to wire Sink.actorSubscriber(...) into
> Source.actorPublisher(...) in one (Partial-) Flow? And by 'wire' I mean to
> somehow send message from ActorSubscriber, instantiated from given props,
> to the 'same flow' instance of ActorPublisher?
>
> I was thinking about giving some unique Id to both of them and then
> publish/subscribe events through `context.system.eventStream`, but then I
> realized that all materialized flows would have same Id and messages would
> mess up... :(
>
> In fact, I'm interested in even more simple task -- create an Actor, that
> could act as PartialFlow[In, Out]: when you push a value into flow, actor
> receives it, processes and sends further down the line.
>
> Of course, I always could do something like Flow[In] mapAsync (data =>
> (actor ? data).mapTo[classOf[Out]]) -- but I would really, really prefer to
> do this message way.
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: Acknowledging messages upon work completion?

[Lance Arlaus]

Andrew-

How about using a simple branching flow with a broadcast and zip?

The first branch carries the message to the end of the pipeline where the 
acknowledger receives both the message and the result of processing as a 
Tuple. It can then decide whether/how to acknowledge the message.
The second branch contains the desired business logic and produces a Try 
(or other monadic data type) as its result.
This way, the processing logic has no need to pass along the message. In 
fact, you could extract and deal with just the payload itself in that 
second branch if you'd like.
Since the zip waits for elements from both branches before emitting, you'll 
have matched (Message, Try) tuples.

Here's a gist: 
https://gist.github.com/lancearlaus/e6e52fc8c7ca534cb026#file-akka-user-stream-ack-scala

And the relevant code (the handleAndAckSink method is the key):

case class Message[T](id: Long, body: T)

trait Queue {
  def acknowledge(id: Long): Unit
}


type Handler[T] = Flow[Message[T], Try[_], _]
type AckSink = Sink[(Message[_], Try[_]), Future[_]]

// A sink that acknowledges messages upon successful processing
def ackSink(queue: Queue) =
  Sink.foreach[(Message[_], Try[_])] {
case (msg, result) => result match {
  case Success(_) => queue.acknowledge(msg.id)
  case Failure(t) => {
// Do something on failure
println(t)
  }
}
  }

// The flow that wraps the handler and acknowledger sink
def handleAndAckSink[T](handler: Handler[T], ackSink: AckSink) = 
  Sink(handler, ackSink, Broadcast[Message[T]](2), Zip[Message[T], 
Try[_]])((_, mat, _, _) => mat) {
implicit b => (handler, ackSink, bcast, zip) =>

bcast~> zip.in0
bcast ~> handler ~> zip.in1
zip.out ~> ackSink

(bcast.in)
  }


class AckSpec extends FlatSpec with AkkaStreamsImplicits with Matchers with 
ScalaFutures {

  def testSource(n: Int) = Source((0 to n)).map(n => Message(n, s"message 
$n"))

  val testQueue = new Queue {
def acknowledge(id: Long) = println(s"acknowledging message $id")
  }
 
  val testHandler = Flow[Message[String]].map { msg => 
// Randomly fail
if (Random.nextBoolean) Failure(new Exception(s"failure processing 
message $msg"))
else Success(s"success processing message $msg")
  }

  "Acknowledge" should "ack messages" in {

val future = testSource(10).runWith(handleAndAckSink(testHandler, 
ackSink(testQueue)))

whenReady(future) { result =>
}

  }

}



Regards,
Lance


On Thursday, August 20, 2015 at 12:20:28 AM UTC-4, Andrew Rollins wrote:
>
> Is there an idiomatic way handle queues with Akka Streas that need to 
> acknowledge messages *after work is done* for a given message?
>
> This started from a thread on Twitter with Victor (
> https://twitter.com/viktorklang/status/634117117978107904), but it's more 
> appropriate to continue here. His last comment was "Sounds like you want 
> Flow[T, Ack[T]] such that you can close the loop at the end."
>
> I'm going to show my interpretation of that suggestion and voice my 
> concern with it. I'd love feedback.
>
> Assume we have an external queue service that provides messages through a 
> "getMessage" API. Messages can be acknowledged by calling "ack(messageId)". 
> After acknowledgement, a message is taken off the queue and won't be 
> delivered again.
>
> I'm not exactly sure how a Flow[T, Ack[T]] helps, because how is the Ack 
> being created from an arbitrary T? We need the original message identifier 
> to be passed through the stream such that we can acknowledge the message, 
> so we would need a flow along the lines of Flow[Msg, Ack]. In code, it 
> could look something like this:
>
>   trait Msg { def msgId } // incoming queue message
>   trait Ack   // ack result type
>
>   class FakeQueue {
> def receive : Msg = ???
> def ack(m: Msg) : Ack = ???
>   }
>
>   val queue = new FakeQueue
>
>   val msgSource: Source[Msg] =
> Source.apply(() => Iterator.iterate[Msg](None)(_ => queue.receive))
>
>   val flow: Flow[Msg, Ack, Unit] = 
> Flow[(Msg, String)].
>   map {msg => (deserialize(msg.body), msg) }.
>   map {case (x, msg) => (doWork(x), msg)}.
>   map {case (y, msg) => insertIntoDatabase(y); msg}.
>   map {msg => queue.ack(Msg)}
>
> This seems ok and certainly works, but I have a hangup with this. All my 
> intermediate steps need to passthrough the message to the end, but they 
> individually don't care about the message. Those stages are coupled with 
> some data they don't ultimately handle. I'd like to avoid that.
>
> In other words, in some hand-wavy sense, there is a desire to take a Flow 
> such as Flow[In, Out] which is queue agnostic and then wrap that flow with 
> something that will dequeue messages from a queue, push the "In" object to 
> the interflow, and somehow pass along the outer message such that the 
> message tied to an "In" pops out the other end with the associated "Out".
>
> I'm at a loss for how to do

Re: [akka-user] Event sourcing backend/plugin with scalable pub/sub

[Leonid Bakaleynik]

> I'm tempted to write a service that would use Kafka and Cassandra as 
> backends and expose subscriptions per entity. Not sure yet if it's doable 
> at scale.
>
> An "exercise left for the ambitious reader" one might say ;-)
>

ambitious reader with less strict deadlines :-) 

We'll probably end up writing to/reading from 2 journals, it will take less 
time to implement

Thanks!
Leonid

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Requester 2.1 released

[Justin du coeur]

For those who are interested: I've just cut release 2.1 of the Requester
library  -- the Actor-friendly
version of ask(), which allows you to safely compose requests from one
Actor to another in a thread-safe way.


Release 2.1 addresses several problems I've encountered while beating on it
hard:

When you are wrapping a RequestM[T] in a Future[T] (typical in RPC
scenarios), *Exceptions now propagate to the Future
*
correctly.

request / *? now works correctly with ActorSelections*; previously, it only
worked with ActorRefs.

Most important: when you flatMap requests together, they need to chain so
that things can "unwind" when you finally come up with a final result.
 (This is necessary for composition to work as expected.)  It turns out
that the original design for this worked recursively: it was beautiful,
elegant, and caused a StackOverflow if you flatMapped more than about 1500
levels deep.  (Yes, I'm currently doing stuff in Querki that is flatMapping
several *thousand* roundtrips in some cases.  It allows delightfully
elegant Akka code
.)
 This is *now rewritten to chain flatMap() a bit less elegantly, but in a
tail-recursive way that should be robust in the face of extreme nesting*.


At this point, I think it's getting seriously ready for prime time.  I
encourage folks to play with it, and drop me a note if you have any
thoughts...

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Event sourcing backend/plugin with scalable pub/sub

[Konrad Malawski]

I'd write to a primary journal and use the (Akka) query side to populate
the other journal. Yes they may be a bit behind, but depending on needs
that could be optimised (query side could be pushed to).

Have you looked into Akka Persistence Query yet? We implemented something
for leveldb this week that might inspire you (I'm on my phone, hard to find
links) - check prs about eventsbyperistenceid.
On Aug 20, 2015 8:36 PM, "Leonid Bakaleynik"  wrote:

>
> I'm tempted to write a service that would use Kafka and Cassandra as
>> backends and expose subscriptions per entity. Not sure yet if it's doable
>> at scale.
>>
>> An "exercise left for the ambitious reader" one might say ;-)
>>
>
> ambitious reader with less strict deadlines :-)
>
> We'll probably end up writing to/reading from 2 journals, it will take
> less time to implement
>
> Thanks!
> Leonid
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: How to achieve concurrency in akka and handle global state

[Lance Arlaus]

Atin-

Though it's tough to recommend an approach without knowing your specific 
requirements, you may want to look into Agents if they fit your use case.
Sharing an Agent among Actors is perfectly safe and they provide 
well-defined semantics for concurrent updates.
Reading the current value from an Agent is dead simple without the need to 
bounce messages around. You can either get the current value directly, or 
obtain a future that'll complete once all currently outstanding updates 
have been applied. Which you choose will depend on your requirements, of 
course.

Here's the relevant doc: 
http://doc.akka.io/docs/akka/snapshot/scala/agents.html

I wouldn't overuse Agents in an architecture, but they're a very handy tool 
to have in your toolbox.
That said, other comments on this thread still apply. The direct use of 
mutable shared state is verboten in concurrent applications! Agents safely 
wrap this state for you.

Regards,
Lance

On Tuesday, August 18, 2015 at 11:05:44 AM UTC-4, Atin Sood wrote:
>
> Re posting since I wasn't able to get any feedback on the last post. 
>
> On Tuesday, August 18, 2015 at 4:16:36 AM UTC-4, Atin Sood wrote:
>>
>> Hi 
>>
>>
>> I am new to learning akka and I am looking into a problem of trying to 
>> create a movie hall reservation system. I am trying to wrap my head around 
>> how to achieve concurrency with akka. If I end up creating multiple 
>> instances of akka actors then where/how should I share the global state of 
>> the total number of seats booked and how many are still available.
>>
>>
>> I believe it should be in the same code that starts the actor system. Or 
>> am I misunderstanding the whole concept.
>>
>>
>> In understand this is an extremely basic question but I wasn't able to 
>> find any good resources explaining this, so will really appreciate if you 
>> can point me in the right direction.
>>
>>
>> Regards
>>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Acknowledging messages upon work completion?

[Patrick McLaren]

I don't know how idiomatic this solution is, but I recently achieved something 
like this by materializing a discrete value, say `Success`, then resolving a 
`Promise` with this value.

If you want to acknowledge a particular message, it might be easier to simply 
close-over it and `map` your acknowledgment after your call to `run`.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Serialization Exception in hazelcast 3.5 with Scala

[M.Ahsen Taqi Kazmi]

I am using Hazelcast 3.5 with scala 
i have a case class Abc i am trying to store object of my class in 
hazelcast from my client  but it gives me some serialization exception

here is my class 
   

 @SerialVersionUID(1)
case class Abc( id : Int ,name : String , subjectCode : 
MutableList[Int]) extends Serialable 

when i run the client code it gives me following exception

18:33:43.274 [hz._hzInstance_1_dev.partition-operation.thread-1] ERROR 
c.h.map.impl.operation.PutOperation - [192.168.15.20]:5701 [dev] [3.5] 
java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
com.hazelcast.nio.serialization.HazelcastSerializationException: 
java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
 at 
com.hazelcast.nio.serialization.DefaultSerializers$ObjectSerializer.read(DefaultSerializers.java:201)
 
~[hazelcast-3.5.jar:3.5]
 at 
com.hazelcast.nio.serialization.StreamSerializerAdapter.read(StreamSerializerAdapter.java:41)
 
~[hazelcast-3.5.jar:3.5]
 at 
com.hazelcast.nio.serialization.SerializationServiceImpl.toObject(SerializationServiceImpl.java:276)
 
~[hazelcast-3.5.jar:3.5]
 at 
com.hazelcast.map.impl.mapstore.AbstractMapDataStore.toObject(AbstractMapDataStore.java:78)
 
~[hazelcast-3.5.jar:3.5]

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Serialization Exception in hazelcast 3.5 with Scala

[M.Ahsen Taqi Kazmi]

On Thursday, August 20, 2015 at 3:13:52 PM UTC+5, peter veentjer wrote:
>
> Do you have the MutableList class definition available on the machine 
> where you are deserializing the object?
>
> On Thu, Aug 20, 2015 at 1:07 PM, M.Ahsen Taqi Kazmi 
>  wrote:
>
>> I am using Hazelcast 3.5 with scala 
>> i have a case class Abc i am trying to store object of my class in 
>> hazelcast from my client  but it gives me some serialization exception
>>
>> here is my class 
>>
>>
>>  @SerialVersionUID(1)
>> case class Abc( id : Int ,name : String , subjectCode : 
>> MutableList[Int]) extends Serialable 
>>
>> when i run the client code it gives me following exception
>>
>> 18:33:43.274 [hz._hzInstance_1_dev.partition-operation.thread-1] 
>> ERROR c.h.map.impl.operation.PutOperation - [192.168.15.20]:5701 [dev] 
>> [3.5] java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
>> com.hazelcast.nio.serialization.HazelcastSerializationException: 
>> java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
>>  at 
>> com.hazelcast.nio.serialization.DefaultSerializers$ObjectSerializer.read(DefaultSerializers.java:201)
>>  
>> ~[hazelcast-3.5.jar:3.5]
>>  at 
>> com.hazelcast.nio.serialization.StreamSerializerAdapter.read(StreamSerializerAdapter.java:41)
>>  
>> ~[hazelcast-3.5.jar:3.5]
>>  at 
>> com.hazelcast.nio.serialization.SerializationServiceImpl.toObject(SerializationServiceImpl.java:276)
>>  
>> ~[hazelcast-3.5.jar:3.5]
>>  at 
>> com.hazelcast.map.impl.mapstore.AbstractMapDataStore.toObject(AbstractMapDataStore.java:78)
>>  
>> ~[hazelcast-3.5.jar:3.5]
>>
>> -- 
>> You received this message because you are subscribed to the Google Groups 
>> "Hazelcast" group.
>> To unsubscribe from this group and stop receiving emails from it, send an 
>> email to hazelcast+...@googlegroups.com.
>> To post to this group, send email to haze...@googlegroups.com.
>> Visit this group at http://groups.google.com/group/hazelcast.
>> To view this discussion on the web visit 
>> https://groups.google.com/d/msgid/hazelcast/d33c3ef3-ab75-4c2a-9d02-1fa2f6a3935f%40googlegroups.com
>>  
>> 
>> .
>> For more options, visit https://groups.google.com/d/optout.
>>
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Implicit value issue for FromRequestUnmarshaller in akka-http with spray-json

[charles adetiloye]

@Heiko... please how did you solve it ?

On Tuesday, July 28, 2015 at 11:45:19 PM UTC-5, Heiko Seeberger wrote:
>
> You need to show the definition of routes. I assume (speculation again) 
> that it’s a method in a different scope which doesn’t take an implicit 
> Materializer (or EC).
>
> Heiko
>
> --
>
> *Heiko Seeberger*
> Home: heikoseeberger.de
> Twitter: @hseeberger 
> Public key: keybase.io/hseeberger
>
> On 28 Jul 2015, at 23:10, Harit Himanshu  > wrote:
>
> I have similar issue, how did you people resolve it?
>
> On Friday, July 24, 2015 at 2:19:06 AM UTC-7, Kabir Idris wrote:
>>
>> Hi Heiko, 
>> Here is my code 
>>
>>  implicit val system = ActorSystem()
>>
>> implicit val executor: ExecutionContext = system.dispatcher
>> val log: LoggingAdapter = Logging(system, getClass)
>> implicit val materializer = ActorMaterializer()
>>
>>
>>   Http().bindAndHandle(routes, httpInterface, httpPort)
>>
>>
>>
>>
>> On Monday, March 16, 2015 at 12:16:37 AM UTC+1, Arthur Kushka wrote:
>>>
>>> Hi. I was faced with a typical error with spray-json and akka-http that 
>>> easy to find in Google, but none of solutions helps me.
>>>
>>> Scala fail compiling in place where I parsing request body with `entity` 
>>> directive. As error say, they can`t find implicit value for unmarshaller, 
>>> but formatter and `SprayJsonSupport` are present in context (in 
>>> documentation written that that`s enough).
>>>
>>> Below is compiler message and my code.
>>> Please help me with finding solution of this problem.
>>>
>>> [error] /home/arhelmus/techmedia/akka-http/src/main/scala/utils/
>>> CustomDirectives.scala:31: could not find implicit value for parameter 
>>> um: akka.http.unmarshalling.FromRequestUnmarshaller[utils.Credentials]
>>> [error] entity(as[Credentials]).flatMap { credentials =>
>>>
>>>
>>> package utils
>>>
>>> import akka.http.marshallers.sprayjson.SprayJsonSupport
>>> import akka.http.server.Directive1
>>> import models.User
>>> import spray.json._
>>>
>>> case class Credentials(login: String, password: String)
>>>
>>> trait Protocols extends DefaultJsonProtocol {
>>>   implicit val formatter: RootJsonFormat[Credentials] = jsonFormat2(
>>> Credentials.apply)
>>> }
>>>
>>> object CustomDirectives extends CustomDirectives
>>>
>>> trait CustomDirectives extends SprayJsonSupport with Protocols {
>>>
>>>   import akka.http.server.directives.BasicDirectives._
>>>   import akka.http.server.directives.MarshallingDirectives._
>>>   import akka.http.server.directives.ParameterDirectives._
>>>
>>>   def authenticateByToken: Directive1[User] =
>>> parameter("token").flatMap { code =>
>>>   provide(User(code, "asdsada", "asdsadasd"))
>>> }
>>>
>>>   def authenticateByCredentials: Directive1[User] = {
>>> entity(as[Credentials]).flatMap { credentials =>
>>>   provide(User(credentials.login, credentials.login, credentials.
>>> password))
>>> }
>>>   }
>>>
>>> }
>>>
>>
> -- 
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: 
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> --- 
> You received this message because you are subscribed to the Google Groups 
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an 
> email to akka-user+...@googlegroups.com .
> To post to this group, send email to akka...@googlegroups.com 
> .
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] Re: Acknowledging messages upon work completion?

[Andrew Rollins]

Lance,

I really appreciate your response, it's well thought out, and the code is a 
great illustration. I have one question, I'm guessing you probably already 
considered it so I'd like to hear your thoughts.

Using your approach, what's the best way to parallelize work that the 
handler does? Is the intention to launch several separate handleAndAckSinks 
to achieve parallelism, or something else?

I ask because the handler flow has to output one element per message in 
order to keep the message and the Try in lock step. That means the handler 
flow can't do something like break down the work into a bunch of smaller 
messages, flatten that into the stream, and utilize mapAsync in that 
flattened stream to do a bunch of parallel work (however, the handler could 
have many steps which provide some parallelism through pipelining). I'm 
just not familiar enough with Akka Streams to intuite the best approach for 
parallelize the handler.

Thanks,
Andrew

On Thursday, August 20, 2015 at 12:31:50 PM UTC-4, Lance Arlaus wrote:
>
> Andrew-
>
> How about using a simple branching flow with a broadcast and zip?
>
> The first branch carries the message to the end of the pipeline where the 
> acknowledger receives both the message and the result of processing as a 
> Tuple. It can then decide whether/how to acknowledge the message.
> The second branch contains the desired business logic and produces a Try 
> (or other monadic data type) as its result.
> This way, the processing logic has no need to pass along the message. In 
> fact, you could extract and deal with just the payload itself in that 
> second branch if you'd like.
> Since the zip waits for elements from both branches before emitting, 
> you'll have matched (Message, Try) tuples.
>
> Here's a gist: 
> https://gist.github.com/lancearlaus/e6e52fc8c7ca534cb026#file-akka-user-stream-ack-scala
>
> And the relevant code (the handleAndAckSink method is the key):
>
> case class Message[T](id: Long, body: T)
>
> trait Queue {
>   def acknowledge(id: Long): Unit
> }
>
>
> type Handler[T] = Flow[Message[T], Try[_], _]
> type AckSink = Sink[(Message[_], Try[_]), Future[_]]
>
> // A sink that acknowledges messages upon successful processing
> def ackSink(queue: Queue) =
>   Sink.foreach[(Message[_], Try[_])] {
> case (msg, result) => result match {
>   case Success(_) => queue.acknowledge(msg.id)
>   case Failure(t) => {
> // Do something on failure
> println(t)
>   }
> }
>   }
>
> // The flow that wraps the handler and acknowledger sink
> def handleAndAckSink[T](handler: Handler[T], ackSink: AckSink) = 
>   Sink(handler, ackSink, Broadcast[Message[T]](2), Zip[Message[T], 
> Try[_]])((_, mat, _, _) => mat) {
> implicit b => (handler, ackSink, bcast, zip) =>
>
> bcast~> zip.in0
> bcast ~> handler ~> zip.in1
> zip.out ~> ackSink
>
> (bcast.in)
>   }
>
>
> class AckSpec extends FlatSpec with AkkaStreamsImplicits with Matchers 
> with ScalaFutures {
>
>   def testSource(n: Int) = Source((0 to n)).map(n => Message(n, s"message 
> $n"))
>
>   val testQueue = new Queue {
> def acknowledge(id: Long) = println(s"acknowledging message $id")
>   }
>  
>   val testHandler = Flow[Message[String]].map { msg => 
> // Randomly fail
> if (Random.nextBoolean) Failure(new Exception(s"failure processing 
> message $msg"))
> else Success(s"success processing message $msg")
>   }
>
>   "Acknowledge" should "ack messages" in {
>
> val future = testSource(10).runWith(handleAndAckSink(testHandler, 
> ackSink(testQueue)))
>
> whenReady(future) { result =>
> }
>
>   }
>
> }
>
>
>
> Regards,
> Lance
>
>
> On Thursday, August 20, 2015 at 12:20:28 AM UTC-4, Andrew Rollins wrote:
>>
>> Is there an idiomatic way handle queues with Akka Streas that need to 
>> acknowledge messages *after work is done* for a given message?
>>
>> This started from a thread on Twitter with Victor (
>> https://twitter.com/viktorklang/status/634117117978107904), but it's 
>> more appropriate to continue here. His last comment was "Sounds like you 
>> want Flow[T, Ack[T]] such that you can close the loop at the end."
>>
>> I'm going to show my interpretation of that suggestion and voice my 
>> concern with it. I'd love feedback.
>>
>> Assume we have an external queue service that provides messages through a 
>> "getMessage" API. Messages can be acknowledged by calling "ack(messageId)". 
>> After acknowledgement, a message is taken off the queue and won't be 
>> delivered again.
>>
>> I'm not exactly sure how a Flow[T, Ack[T]] helps, because how is the Ack 
>> being created from an arbitrary T? We need the original message identifier 
>> to be passed through the stream such that we can acknowledge the message, 
>> so we would need a flow along the lines of Flow[Msg, Ack]. In code, it 
>> could look something like this:
>>
>>   trait Msg { def msgId } // incoming queue message
>>   trait Ack   // ack

Re: [akka-user] Serialization Exception in hazelcast 3.5 with Scala

[Viktor Klang]

How is this related to Akka?

On Thu, Aug 20, 2015 at 3:21 AM, M.Ahsen Taqi Kazmi 
wrote:

> I am using Hazelcast 3.5 with scala
> i have a case class Abc i am trying to store object of my class in
> hazelcast from my client  but it gives me some serialization exception
>
> here is my class
>
>
>  @SerialVersionUID(1)
> case class Abc( id : Int ,name : String , subjectCode :
> MutableList[Int]) extends Serialable
>
> when i run the client code it gives me following exception
>
> 18:33:43.274 [hz._hzInstance_1_dev.partition-operation.thread-1] ERROR
> c.h.map.impl.operation.PutOperation - [192.168.15.20]:5701 [dev] [3.5]
> java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
> com.hazelcast.nio.serialization.HazelcastSerializationException:
> java.lang.ClassNotFoundException: scala.collection.mutable.MutableList
>  at
> com.hazelcast.nio.serialization.DefaultSerializers$ObjectSerializer.read(DefaultSerializers.java:201)
> ~[hazelcast-3.5.jar:3.5]
>  at
> com.hazelcast.nio.serialization.StreamSerializerAdapter.read(StreamSerializerAdapter.java:41)
> ~[hazelcast-3.5.jar:3.5]
>  at
> com.hazelcast.nio.serialization.SerializationServiceImpl.toObject(SerializationServiceImpl.java:276)
> ~[hazelcast-3.5.jar:3.5]
>  at
> com.hazelcast.map.impl.mapstore.AbstractMapDataStore.toObject(AbstractMapDataStore.java:78)
> ~[hazelcast-3.5.jar:3.5]
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>



-- 
Cheers,
√

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[akka-user] How to survive long GC pause

[Ara Tatous]

I have a configuration of Akka cluster with 4 nodes. Let's call them Engine 
and Warehouse.  

In Engine:

 

*Everything is normal*

[DEBUG] [08/20/2015 07:32:25.082] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-80] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2551]

[DEBUG] [08/20/2015 07:32:25.082] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-80] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2552]

[DEBUG] [08/20/2015 07:32:25.082] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-80] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod2:2551]

[DEBUG] [08/20/2015 07:32:25.084] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-98] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat response from 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2551]

[DEBUG] [08/20/2015 07:32:25.084] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-98] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat response from 
[akka.tcp://EngineWarehouseActorSystem@compprod1:2551]

[DEBUG] [08/20/2015 07:32:25.084] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-98] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat response from 
[akka.tcp://EngineWarehouseActorSystem@compprod1:2552]

 

 

*We have GC pause here*

2015-08-20T07:32:31.959-0700: [GC2015-08-20T07:32:31.960-0700: [ParNew 
(promotion failed): 172544K->170208K(172544K), 0.8968931 
secs]2015-08-20T07:32:32.857-0700: [CMS2015-08-20T07:32:39.682-0700: 
[CMS-concurrent-sweep: 7.003/7.909 secs] [Times: user=8.78 sys=0.40, 
real=7.91 secs] 

 (concurrent mode failure): 5760566K->2755481K(6099776K), 30.9310318 secs] 
5924201K->2755481K(6272320K), [CMS Perm : 142090K->141971K(237016K)], 
31.8286597 secs] [Times: user=32.22 sys=0.38, real=31.83 secs] 

 

*We have almost 32 seconds GC pause. The following heartbeat request are 
buffered and played right after GC pause in one single second!. All the 
connection is being disassociated with other cluster members.*

 

[DEBUG] [08/20/2015 07:33:03.794] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-23] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2551]

[DEBUG] [08/20/2015 07:33:03.794] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-23] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2552]

[DEBUG] [08/20/2015 07:33:03.794] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-23] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod2:2551]

[DEBUG] [08/20/2015 07:33:03.808] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-81] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2551]

[DEBUG] [08/20/2015 07:33:03.808] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-81] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod1:2552]

[DEBUG] [08/20/2015 07:33:03.808] 
[EngineWarehouseActorSystem-akka.actor.default-dispatcher-81] 
[akka.tcp://EngineWarehouseActorSystem@compprod2:2552/system/cluster/core/daemon/heartbeatSender]
 
Cluster Node [akka.tcp://EngineWarehouseActorSystem@compprod2:2552] - 
Heartbeat to [akka.tcp://EngineWarehouseActorSystem@compprod2:2551]

[DEBUG] [08/20/2015 07:33