Re: [akka-user] Migrating from wandoulabs websockets to stock akka-io

[Sam Halliday]

Hi Roland,

I've read the documentation, several times, I've even given you feedback on
the documentation in an earlier milestone phase. Also, the documentation
for WebSockets in Akka is TODO and TODO. The documentation on the routing
directives are extremely sparse. In particular, there are no promises
around the implementation of back pressure from the new websockets.

What I'm missing is the ability to hook an existing actor system into
something that expects a Flow, with back pressure preserved. I understand
Flow, but I don't understand the implementation in terms of Actors (which
incidentally, is exactly my primary feedback on the earlier documentation).
You're now confusing me further by saying that Streams are not actors,
because I was told at the time that streams are implemented in terms of
actors.

In case you didn't pick up on it, I'm planning on moving away from
wandoulabs, not integrate it. This is the key piece, distilled into a
standalone problem.

Best regards, Sam
On 27 May 2015 7:35 am, "Roland Kuhn"  wrote:

> Hi Sam,
>
> it might be better to take a step back before potentially running in the
> wrong direction. First off, Akka HTTP offers a complete solution for
> everything HTTP (including websockets) within an ActorSystem. Before
> deciding to combine this with another tool I recommend that you explore
> first how Akka HTTP works, because it introduces several fundamentally new
> concepts. In particular, when talking about it as “Spray 2.0” it is
> important to note that everything ActorRef-related in Spray has been
> replaced by Streams—a completely different abstraction that is *not* an
> Actor. The whole underpinnings are completely rewritten in a radically
> different fashion, so don’t expect any Spray modules that live “beneath the
> surface” to seamlessly fit onto Akka HTTP.
>
> We could go into the details Wandoulabs’ websocket add-on, but I don’t see
> much value in discussing that before the basics are clear. The other piece
> of information that I’m lacking is why you would want to “retrofit”
> something in this context, it might be better to explain the ends and not
> the means in order to get help.
>
> Regards,
>
> Roland
>
> 23 maj 2015 kl. 12:38 skrev Sam Halliday :
>
> Hi all,
>
> I'm very excited that akka-io now has WebSocket support.
>
> In ENSIME, we're planning on using this wrapper over wandoulab's websockets
>
>   https://github.com/smootoo/simple-spray-websockets
>
> to easily create a REST/WebSockets endpoint with JSON marshalling for a
> sealed family, with backpressure.
>
> Smootoo's wrapper works really well, and I have had the pleasure of using
> it in a corporate environment so I trust it to be stable.
>
>
> For future proofing, it would seem sensible to move to stock akka-io for
> WebSockets, so I'm considering sending a PR to retrofit the wrapper. I have
> a couple of questions about that:
>
> 1. does akka-io's HTTP singleton actor support WebSockets now? That was
> the big caveat about using wandoulabs. It means all kinds of workarounds if
> you want to just use HTTP in the same actor system.
>
> 2. is there a migration guide for wandoulabs to akka-io? Or would it be
> best just to rewrite the wrapper from scratch on top of akka-io?
>
> 3. where is the documentation? This just has a big TODO on it
>
>
> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/client-side/websocket-support.html
>
> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/routing-dsl/websocket-support.html
>
> I can't even find any examples. I guess the key thing is the handshaking,
> which would mean rewriting this bit (and the corresponding client side
> handshake)
>
>
> https://github.com/smootoo/simple-spray-websockets/blob/master/src/main/scala/org/suecarter/websocket/WebSocket.scala#L167
>
> Best regards,
> Sam
>
> --
> >> 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.
>
>
>
>
> *Dr. Roland Kuhn*
> *Akka Tech Lead*
> Typesafe  – Reactive apps on the JVM.
> twitter: @rolandkuhn
> 
>
>  --
> >> 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 a topic in the
> Google Groups "Akka 

Re: [akka-user] Flow from an Actor

[Roland Kuhn]

> 26 maj 2015 kl. 16:02 skrev Sam Halliday :
> 
> To maybe try and formalise this a little bit more, and abstract away from 
> WebSockets (that will only muddy the water).
> 
> Lets say we have an Actor already that looks like this
> 
> sealed trait Incoming
> sealed trait Outgoing
> class SimpleActor(upstream: ActorRef) extends Actor {
>   def receive = {
> case in: Incoming =>
>// work, including some upstream ! outgoing
> case Ack =>
>// ack for the last message upstream
> case other =>
>// work, including some upstream ! outgoing
>   }
> }
> 
> How do I wrap that as a Flow[Incoming, Outgoing, Unit] ?

Short answer: by converting it to a PushPullStage (see the docs 
;
 and here are the implementations 

 of the common combinators)

Long answer: the flow control protocol described by Reactive Streams 
 is a lot more involved than your simplistic 
Ack protocol, there are failure conditions and corner cases to be considered to 
make it work reliably. Implementing that protocol yourself is a non-trivial 
task, we do not consider Publisher/Subscriber to be end-user API for that 
reason. Please trust me when I report that we spent 1.5 years together with 
some really bright engineers hammering out the details and trying to find the 
minimal specification that works, so the reason for the complexity is not that 
we complicated things, it lies within the subject matter itself.

Regards,

Roland

> 
> 
> On Tuesday, 26 May 2015 14:09:45 UTC+1, Sam Halliday wrote:
> Re: asyncyMap. I don't think that is going to work, there is no implied 
> single response to each query (which I gather is what you're suggesting)? And 
> I need some way of receiving new messages from upstream.
> 
> The existing Actor is both a sink (i.e. it consumes messages from upstream, 
> not necessarily responding to each one) and a source (i.e. it can send an 
> effectively infinite number of messages). It is using backpressure, but only 
> using its own `Ack` message.
> 
> For some context, I'm retrofitting some code that is using this WebSockets 
> layer around wandoulabs, e.g. 
> https://github.com/smootoo/simple-spray-websockets/blob/master/src/test/scala/org/suecarter/websocket/WebSocketSpec.scala#L150
>  
> 
> 
> But the newly released akka-io layer expects a Flow.
> 
> The `Ack` is being received when messages were sent directly to the I/O 
> layer. Presumably, the backpressure is implemented differently now... 
> although I am not sure how yet. That's the second problem once I can actually 
> get everything hooked up.
> 
> 
> On Tuesday, 26 May 2015 13:57:43 UTC+1, √ wrote:
> Not knowing what your actor is trying to do, what about Flow.mapAsync + ask?
> 
> -- 
> Cheers,
> √
> 
> On 26 May 2015 14:54, "Sam Halliday" > wrote:
> Hi all,
> 
> I need to interface an Actor with an API that requires a Flow.
> 
> The actor can receive a sealed trait family of inputs and will only send (a 
> different) sealed family of outputs to upstream, so I suspect that will help 
> matters.
> 
> Looking in FlowOps, it looks like I can create a Flow from a partial 
> function, but there isn't anything that would just simply take an ActorRef.
> 
> Am I missing something trivial to just upgrade an ActoRef to a Flow? 
> (Obviously there is a bunch of extra messages the actor will have to handle, 
> such as backpressure messages etc... but assume that's all taken care of)
> 
> Best regards,
> Sam
> 
> -- 
> >> 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: h

Re: [akka-user] Flow from an Actor

[Roland Kuhn]

> 26 maj 2015 kl. 14:54 skrev Sam Halliday :
> 
> Hi all,
> 
> I need to interface an Actor with an API that requires a Flow.
> 
> The actor can receive a sealed trait family of inputs and will only send (a 
> different) sealed family of outputs to upstream, so I suspect that will help 
> matters.
> 
> Looking in FlowOps, it looks like I can create a Flow from a partial 
> function, but there isn't anything that would just simply take an ActorRef.

Instead of trying to make sense of method signatures I highly recommend reading 
the documentation first—we spent considerable effort on describing the entirely 
new abstractions that we have built, and you will not understand the point 
behind the signatures without knowing what “Flow” entails.

> 
> Am I missing something trivial to just upgrade an ActoRef to a Flow? 
> (Obviously there is a bunch of extra messages the actor will have to handle, 
> such as backpressure messages etc... but assume that's all taken care of)

Yes, when just using Flows and our DSL then we construct Actors for you that 
take care of all these things. But when you write those Actors yourself, then 
obviously you need to take care of these things.

Regards,

Roland

> 
> Best regards,
> Sam
> 
> -- 
> >> 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 
> .



Dr. Roland Kuhn
Akka Tech Lead
Typesafe  – Reactive apps on the JVM.
twitter: @rolandkuhn
 

-- 
>>  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: Testing Akka Actors using mocks

[Harit Himanshu]

>
> Frown upon mocks in actor testing, though they do have their place.


This use case is perfect for mocks because the 

LogReaderDisruptor main(Array())


is long running process (which will continue to run forever.

To give you some context,  LogReaderDisruptor is entry to our current 
project as a single process. 

We are converting this to akka based project where first step is crash 
receovery, as in if Actor (running LogReaderDisruptor main(Array())) gets 
killed (because of some exception), we restart it. 

I can't comment on whether this is best design strategy, but this for sure 
adds value to our project and gives us ways to add fault tolerance to our 
application.
Does that makes sense?

Thanks

On Tuesday, May 26, 2015 at 3:56:16 PM UTC-7, Harit Himanshu wrote:
>
> I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest` 
>
> I am working on a problem where my `Actor` gives call to external system. 
>
> case object LogProcessRequest
> 
> class LProcessor extends Actor {
>   val log = Logging(context.system, this)
> 
>   def receive = {
> case LogProcessRequest =>
>   log.debug("starting log processing")
>   LogReaderDisruptor main(Array())
>   }
> }
>
> The `LogReaderDisruptor main(Array())` is a `Java` class that does many 
> other things.  
>
> The test I have currently looks like  
>
> class LProcessorSpec extends UnitTestSpec("testSystem") {
> 
>   "A mocked log processor" should {
> "be called" in  {
>   val logProcessorActor = system.actorOf(Props[LProcessor])
>   logProcessorActor ! LogProcessRequest
> }
>   }
> }
>
> where `UnitTestSpec` looks like (and inspired from [here][1]) 
>
> import akka.actor.ActorSystem
> import akka.testkit.{ImplicitSender, TestKit}
> import org.scalatest.matchers.MustMatchers
> import org.scalatest.{BeforeAndAfterAll, WordSpecLike}
> 
> abstract class UnitTestSpec(name: String)
>   extends TestKit(ActorSystem(name))
>   with WordSpecLike
>   with MustMatchers
>   with BeforeAndAfterAll
>   with ImplicitSender {
> 
>   override def afterAll() {
> system.shutdown()
>   }
> }
>
>
> **Question**   
>
> - How can I mock the call to `LogReaderDisruptor main(Array())` and verify 
> that it was called?  
>
> I am coming from `Java`, `JUnit`, `Mockito` land and something that I 
> would have done here would be  
>
> doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
> verify(logReaderDisruptor, times(1)).main(Matchers.anyVararg())
>
> I am not sure how to translate that with ScalaTest here.  
>
> Also, This code may not be idiomatic, since I am very new and learning
>
>   [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html
>

-- 
>>  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] Migrating from wandoulabs websockets to stock akka-io

[Roland Kuhn]

Hi Sam,

it might be better to take a step back before potentially running in the wrong 
direction. First off, Akka HTTP offers a complete solution for everything HTTP 
(including websockets) within an ActorSystem. Before deciding to combine this 
with another tool I recommend that you explore first how Akka HTTP works, 
because it introduces several fundamentally new concepts. In particular, when 
talking about it as “Spray 2.0” it is important to note that everything 
ActorRef-related in Spray has been replaced by Streams—a completely different 
abstraction that is not an Actor. The whole underpinnings are completely 
rewritten in a radically different fashion, so don’t expect any Spray modules 
that live “beneath the surface” to seamlessly fit onto Akka HTTP.

We could go into the details Wandoulabs’ websocket add-on, but I don’t see much 
value in discussing that before the basics are clear. The other piece of 
information that I’m lacking is why you would want to “retrofit” something in 
this context, it might be better to explain the ends and not the means in order 
to get help.

Regards,

Roland

> 23 maj 2015 kl. 12:38 skrev Sam Halliday :
> 
> Hi all,
> 
> I'm very excited that akka-io now has WebSocket support.
> 
> In ENSIME, we're planning on using this wrapper over wandoulab's websockets
> 
>   https://github.com/smootoo/simple-spray-websockets
> 
> to easily create a REST/WebSockets endpoint with JSON marshalling for a 
> sealed family, with backpressure.
> 
> Smootoo's wrapper works really well, and I have had the pleasure of using it 
> in a corporate environment so I trust it to be stable.
> 
> 
> For future proofing, it would seem sensible to move to stock akka-io for 
> WebSockets, so I'm considering sending a PR to retrofit the wrapper. I have a 
> couple of questions about that:
> 
> 1. does akka-io's HTTP singleton actor support WebSockets now? That was the 
> big caveat about using wandoulabs. It means all kinds of workarounds if you 
> want to just use HTTP in the same actor system.
> 
> 2. is there a migration guide for wandoulabs to akka-io? Or would it be best 
> just to rewrite the wrapper from scratch on top of akka-io?
> 
> 3. where is the documentation? This just has a big TODO on it
> 
>   
> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/client-side/websocket-support.html
>   
> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/routing-dsl/websocket-support.html
> 
> I can't even find any examples. I guess the key thing is the handshaking, 
> which would mean rewriting this bit (and the corresponding client side 
> handshake)
> 
>   
> https://github.com/smootoo/simple-spray-websockets/blob/master/src/main/scala/org/suecarter/websocket/WebSocket.scala#L167
> 
> Best regards,
> Sam
> 
> -- 
> >> 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 
> .



Dr. Roland Kuhn
Akka Tech Lead
Typesafe  – Reactive apps on the JVM.
twitter: @rolandkuhn
 

-- 
>>  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 force ClusterClient recover from akka.actor.StashOverflowException?

[bearrito]

Correction akka version is 2.3.10

On Tuesday, May 26, 2015 at 9:12:10 PM UTC-4, bearrito wrote:
>
> I have a clusterClient which will attempt to contact a cluster in order 
> request work. It is possible for the client to become disconnected. That 
> will eventually end up causing
>
> akka.actor.StashOverflowException: Couldn't enqueue message 
> Envelope(Publish(...))
>
> How can I force the clusterClient to either not stash or to  recover?
>
> scala 2.10.4
> akka 2.2.3
>
> -b
>

-- 
>>  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: Testing Akka Actors using mocks

[bearrito]

You could put your log processing into a trait like

trait LogProcessing {

def isImportantMessage(entry:LogProcessRequest) = ???
}

You could then directly extend your actor like

class LProcessor  extends Actor with LogProcessing {...}

or simply do class LProcessor(disruptor: LogReaderDisruptor) extends Actor 
{...}


I would probably end up writing something like the below on a first pass.



case object LogProcessRequest
case object LogProcessStarted
case object LogProcessFinished


class LProcessor(disruptor: LogReaderDisruptor) extends Actor {
  val log = Logging(context.system, this)

  def receive = {
case LogProcessRequest =>
  sender ! LogProcessStarted
  log.debug("starting log processing")
  disruptor.main(Array())
 sender ! LogProcessFinished
  }
}

You can then verify that the call is made because you will receive a 
LogProcessFinished message.

Frown upon mocks in actor testing, though they do have their place.

On Tuesday, May 26, 2015 at 9:30:23 PM UTC-4, Harit Himanshu wrote:
>
> @bearrito, I am very new to Scala/Akka world, coming from Java background. 
> I had trouble understanding your statement about 
>
> Split your processing from the actor portion
>
>
> This seems like a interesting advice, could you please explain what you 
> mean with it?
>
> Thanks a lot
> + Harit
> On Tuesday, May 26, 2015 at 6:13:05 PM UTC-7, bearrito wrote:
>>
>> Split your processing from the actor portion.
>>
>> Use a trait or pass the processor in using DI.
>>
>> On Tuesday, May 26, 2015 at 6:56:16 PM UTC-4, Harit Himanshu wrote:
>>>
>>> I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest` 
>>>
>>> I am working on a problem where my `Actor` gives call to external 
>>> system. 
>>>
>>> case object LogProcessRequest
>>> 
>>> class LProcessor extends Actor {
>>>   val log = Logging(context.system, this)
>>> 
>>>   def receive = {
>>> case LogProcessRequest =>
>>>   log.debug("starting log processing")
>>>   LogReaderDisruptor main(Array())
>>>   }
>>> }
>>>
>>> The `LogReaderDisruptor main(Array())` is a `Java` class that does many 
>>> other things.  
>>>
>>> The test I have currently looks like  
>>>
>>> class LProcessorSpec extends UnitTestSpec("testSystem") {
>>> 
>>>   "A mocked log processor" should {
>>> "be called" in  {
>>>   val logProcessorActor = system.actorOf(Props[LProcessor])
>>>   logProcessorActor ! LogProcessRequest
>>> }
>>>   }
>>> }
>>>
>>> where `UnitTestSpec` looks like (and inspired from [here][1]) 
>>>
>>> import akka.actor.ActorSystem
>>> import akka.testkit.{ImplicitSender, TestKit}
>>> import org.scalatest.matchers.MustMatchers
>>> import org.scalatest.{BeforeAndAfterAll, WordSpecLike}
>>> 
>>> abstract class UnitTestSpec(name: String)
>>>   extends TestKit(ActorSystem(name))
>>>   with WordSpecLike
>>>   with MustMatchers
>>>   with BeforeAndAfterAll
>>>   with ImplicitSender {
>>> 
>>>   override def afterAll() {
>>> system.shutdown()
>>>   }
>>> }
>>>
>>>
>>> **Question**   
>>>
>>> - How can I mock the call to `LogReaderDisruptor main(Array())` and 
>>> verify that it was called?  
>>>
>>> I am coming from `Java`, `JUnit`, `Mockito` land and something that I 
>>> would have done here would be  
>>>
>>> 
>>> doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
>>> verify(logReaderDisruptor, 
>>> times(1)).main(Matchers.anyVararg())
>>>
>>> I am not sure how to translate that with ScalaTest here.  
>>>
>>> Also, This code may not be idiomatic, since I am very new and learning
>>>
>>>   [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html
>>>
>>

-- 
>>  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: Testing Akka Actors using mocks

[Harit Himanshu]

@bearrito, I am very new to Scala/Akka world, coming from Java background. 
I had trouble understanding your statement about 

Split your processing from the actor portion


This seems like a interesting advice, could you please explain what you 
mean with it?

Thanks a lot
+ Harit
On Tuesday, May 26, 2015 at 6:13:05 PM UTC-7, bearrito wrote:
>
> Split your processing from the actor portion.
>
> Use a trait or pass the processor in using DI.
>
> On Tuesday, May 26, 2015 at 6:56:16 PM UTC-4, Harit Himanshu wrote:
>>
>> I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest` 
>>
>> I am working on a problem where my `Actor` gives call to external system. 
>>
>> case object LogProcessRequest
>> 
>> class LProcessor extends Actor {
>>   val log = Logging(context.system, this)
>> 
>>   def receive = {
>> case LogProcessRequest =>
>>   log.debug("starting log processing")
>>   LogReaderDisruptor main(Array())
>>   }
>> }
>>
>> The `LogReaderDisruptor main(Array())` is a `Java` class that does many 
>> other things.  
>>
>> The test I have currently looks like  
>>
>> class LProcessorSpec extends UnitTestSpec("testSystem") {
>> 
>>   "A mocked log processor" should {
>> "be called" in  {
>>   val logProcessorActor = system.actorOf(Props[LProcessor])
>>   logProcessorActor ! LogProcessRequest
>> }
>>   }
>> }
>>
>> where `UnitTestSpec` looks like (and inspired from [here][1]) 
>>
>> import akka.actor.ActorSystem
>> import akka.testkit.{ImplicitSender, TestKit}
>> import org.scalatest.matchers.MustMatchers
>> import org.scalatest.{BeforeAndAfterAll, WordSpecLike}
>> 
>> abstract class UnitTestSpec(name: String)
>>   extends TestKit(ActorSystem(name))
>>   with WordSpecLike
>>   with MustMatchers
>>   with BeforeAndAfterAll
>>   with ImplicitSender {
>> 
>>   override def afterAll() {
>> system.shutdown()
>>   }
>> }
>>
>>
>> **Question**   
>>
>> - How can I mock the call to `LogReaderDisruptor main(Array())` and 
>> verify that it was called?  
>>
>> I am coming from `Java`, `JUnit`, `Mockito` land and something that I 
>> would have done here would be  
>>
>> 
>> doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
>> verify(logReaderDisruptor, 
>> times(1)).main(Matchers.anyVararg())
>>
>> I am not sure how to translate that with ScalaTest here.  
>>
>> Also, This code may not be idiomatic, since I am very new and learning
>>
>>   [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html
>>
>

-- 
>>  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: Disambiguate actorRefs

[bearrito]

I use ScalaGuice with Name annotations. If you really want to enforce it 
then ScalaGuice and wrap the props or refs in a case class or value class 
as you mention.

On Tuesday, May 26, 2015 at 2:32:08 PM UTC-4, Jeff wrote:
>
> I have an actor that I need to pass several actorRefs into (remote 
> services). Is there a best practice for doing this with type safety (so a 
> user doesn't accidentally pass the wrong actorRef)? Some options I've 
> considered are using scalaz Tags and native scala Value Classes. 
>

-- 
>>  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: Testing Akka Actors using mocks

[bearrito]

Split your processing from the actor portion.

Use a trait or pass the processor in using DI.

On Tuesday, May 26, 2015 at 6:56:16 PM UTC-4, Harit Himanshu wrote:
>
> I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest` 
>
> I am working on a problem where my `Actor` gives call to external system. 
>
> case object LogProcessRequest
> 
> class LProcessor extends Actor {
>   val log = Logging(context.system, this)
> 
>   def receive = {
> case LogProcessRequest =>
>   log.debug("starting log processing")
>   LogReaderDisruptor main(Array())
>   }
> }
>
> The `LogReaderDisruptor main(Array())` is a `Java` class that does many 
> other things.  
>
> The test I have currently looks like  
>
> class LProcessorSpec extends UnitTestSpec("testSystem") {
> 
>   "A mocked log processor" should {
> "be called" in  {
>   val logProcessorActor = system.actorOf(Props[LProcessor])
>   logProcessorActor ! LogProcessRequest
> }
>   }
> }
>
> where `UnitTestSpec` looks like (and inspired from [here][1]) 
>
> import akka.actor.ActorSystem
> import akka.testkit.{ImplicitSender, TestKit}
> import org.scalatest.matchers.MustMatchers
> import org.scalatest.{BeforeAndAfterAll, WordSpecLike}
> 
> abstract class UnitTestSpec(name: String)
>   extends TestKit(ActorSystem(name))
>   with WordSpecLike
>   with MustMatchers
>   with BeforeAndAfterAll
>   with ImplicitSender {
> 
>   override def afterAll() {
> system.shutdown()
>   }
> }
>
>
> **Question**   
>
> - How can I mock the call to `LogReaderDisruptor main(Array())` and verify 
> that it was called?  
>
> I am coming from `Java`, `JUnit`, `Mockito` land and something that I 
> would have done here would be  
>
> doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
> verify(logReaderDisruptor, times(1)).main(Matchers.anyVararg())
>
> I am not sure how to translate that with ScalaTest here.  
>
> Also, This code may not be idiomatic, since I am very new and learning
>
>   [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html
>

-- 
>>  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 force ClusterClient recover from akka.actor.StashOverflowException?

[bearrito]

I have a clusterClient which will attempt to contact a cluster in order 
request work. It is possible for the client to become disconnected. That 
will eventually end up causing

akka.actor.StashOverflowException: Couldn't enqueue message 
Envelope(Publish(...))

How can I force the clusterClient to either not stash or to  recover?

scala 2.10.4
akka 2.2.3

-b

-- 
>>  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] Testing Akka Actors using mocks

[Dragisa Krsmanovic]

Instead of hardcoding, can you pass LogReaderDisruptor in the actor
constructor ? Then, when testing you can replace it by a mock.

On Tue, May 26, 2015 at 3:56 PM, Harit Himanshu <
harit.subscripti...@gmail.com> wrote:

> I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest`
>
> I am working on a problem where my `Actor` gives call to external system.
>
> case object LogProcessRequest
>
> class LProcessor extends Actor {
>   val log = Logging(context.system, this)
>
>   def receive = {
> case LogProcessRequest =>
>   log.debug("starting log processing")
>   LogReaderDisruptor main(Array())
>   }
> }
>
> The `LogReaderDisruptor main(Array())` is a `Java` class that does many
> other things.
>
> The test I have currently looks like
>
> class LProcessorSpec extends UnitTestSpec("testSystem") {
>
>   "A mocked log processor" should {
> "be called" in  {
>   val logProcessorActor = system.actorOf(Props[LProcessor])
>   logProcessorActor ! LogProcessRequest
> }
>   }
> }
>
> where `UnitTestSpec` looks like (and inspired from [here][1])
>
> import akka.actor.ActorSystem
> import akka.testkit.{ImplicitSender, TestKit}
> import org.scalatest.matchers.MustMatchers
> import org.scalatest.{BeforeAndAfterAll, WordSpecLike}
>
> abstract class UnitTestSpec(name: String)
>   extends TestKit(ActorSystem(name))
>   with WordSpecLike
>   with MustMatchers
>   with BeforeAndAfterAll
>   with ImplicitSender {
>
>   override def afterAll() {
> system.shutdown()
>   }
> }
>
>
> **Question**
>
> - How can I mock the call to `LogReaderDisruptor main(Array())` and verify
> that it was called?
>
> I am coming from `Java`, `JUnit`, `Mockito` land and something that I
> would have done here would be
>
> doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
> verify(logReaderDisruptor, times(1)).main(Matchers.anyVararg())
>
> I am not sure how to translate that with ScalaTest here.
>
> Also, This code may not be idiomatic, since I am very new and learning
>
>   [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html
>
> --
> >> 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] Testing Akka Actors using mocks

[Harit Himanshu]

I am new to entire ecosystem including `Scala`, `Akka` and `ScalaTest` 

I am working on a problem where my `Actor` gives call to external system. 

case object LogProcessRequest

class LProcessor extends Actor {
  val log = Logging(context.system, this)

  def receive = {
case LogProcessRequest =>
  log.debug("starting log processing")
  LogReaderDisruptor main(Array())
  }
}

The `LogReaderDisruptor main(Array())` is a `Java` class that does many 
other things.  

The test I have currently looks like  

class LProcessorSpec extends UnitTestSpec("testSystem") {

  "A mocked log processor" should {
"be called" in  {
  val logProcessorActor = system.actorOf(Props[LProcessor])
  logProcessorActor ! LogProcessRequest
}
  }
}

where `UnitTestSpec` looks like (and inspired from [here][1]) 

import akka.actor.ActorSystem
import akka.testkit.{ImplicitSender, TestKit}
import org.scalatest.matchers.MustMatchers
import org.scalatest.{BeforeAndAfterAll, WordSpecLike}

abstract class UnitTestSpec(name: String)
  extends TestKit(ActorSystem(name))
  with WordSpecLike
  with MustMatchers
  with BeforeAndAfterAll
  with ImplicitSender {

  override def afterAll() {
system.shutdown()
  }
}


**Question**   

- How can I mock the call to `LogReaderDisruptor main(Array())` and verify 
that it was called?  

I am coming from `Java`, `JUnit`, `Mockito` land and something that I would 
have done here would be  

doNothing().when(logReaderDisruptor).main(Matchers.anyVararg())
verify(logReaderDisruptor, times(1)).main(Matchers.anyVararg())

I am not sure how to translate that with ScalaTest here.  

Also, This code may not be idiomatic, since I am very new and learning

  [1]: http://www.superloopy.io/articles/2013/scalatest-with-akka.html

-- 
>>  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 remoting over ssl with client auth

[Jim Newsham]

For security reasons, we would like to enable two-way ssl authentication 
for our akka remoting communication.  Is this possible?

Thanks,
Jim

-- 
>>  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] Disambiguate actorRefs

[Jeff]

I have an actor that I need to pass several actorRefs into (remote 
services). Is there a best practice for doing this with type safety (so a 
user doesn't accidentally pass the wrong actorRef)? Some options I've 
considered are using scalaz Tags and native scala Value Classes. 

-- 
>>  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 Receive as parameter

[Jeff]

This is great advice, thanks!

On Saturday, May 23, 2015 at 1:58:29 AM UTC-7, rkuhn wrote:
>
> Using something like Spores you could ascertain that your partial function 
> does indeed not close over its surrounding Actor when creating anonymous 
> Actors inline, but Spores are still an experiment. An equally safe but not 
> as elegant solution is to declare the partial function in the Actor’s 
> companion object and pass the needed values into the method that 
> instantiates it (just like the `def props(...): Props` recommendation).
>
> Regards,
>
> Roland
>
> 22 maj 2015 kl. 23:57 skrev Jeff >:
>
> Following up on this question, what are the best practices around creating 
> anonymous actors, as long as you are not closing over context of a parent 
> actor?
>
> On Friday, May 22, 2015 at 12:08:45 PM UTC-7, Jeff wrote:
>>
>> Is it bad practice to pass in the Receive pf to an actor as part of the 
>> constructor arguments, assuming all vals it closes over are consts?
>>
>
> -- 
> >> 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.
>
>
>
>
> *Dr. Roland Kuhn*
> *Akka Tech Lead*
> Typesafe  – Reactive apps on the JVM.
> twitter: @rolandkuhn
> 
>  
>

-- 
>>  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][Java] How to keep ActorRef to use it in my controller, if I will stop my actor system and start it later?

[Mahitab Farid]

Hi,

I have actor system, I initialize it at the application beginning.
I need to call these actors from my controllers, so I pass the ActorRef to 
my controllers/services and keep them in local variable.

upon some actions from user the actor system could be stopped, then started 
(not restart).

keeping the references in the controller/services causing me errors after 
starting the ActorSystem as the controllers hold the old references 
(incarnations).

So, how to keep the actor references?

should I use ActorSelection with each request? would that affect the 
performance?

public class MyController {

  public static Promise myAction(String param) {

ActorSelection myActor = Akka.system().actorSelection(
"akka://application/user/myActor");

String myMessage = "hello";
Promise resultJsonObjectPromise = 
  Promise.wrap(ask(myActor, myMessage, 100)).map(
new Function() {
  public JSONObject apply(Object response) {
return ((myResponse) response).getJsonResult();
  }
});

return resultJsonObjectPromise.map(new Function() {
  public Result apply(JSONObject response) {
  return ok(response.toString());
  }
});
  }
}



or shall I keep the actor reference in a main class and each time 
controller get the actor ref from this class?


public class ActorBag {
  public static ActorRef myActorRef;
}


@ application start:

ActorBag.myActorRef = Akka.system().actorOf(
Props.create(MyActor.class, params), "MyActor");



public class MyController {

  public static Promise myAction(String param) {

ActorRef myActor = ActorBag.myActorRef;

String myMessage = "hello";
Promise resultJsonObjectPromise = 
  Promise.wrap(ask(myActor, myMessage, 100)).map(
new Function() {
  public JSONObject apply(Object response) {
return ((myResponse) response).getJsonResult();
  }
});
//same as above
  }
}


What is the best practice
ActorRef myActor = ActorBag.myActorRef;
or
ActorSelection myActor = 
Akka.system().actorSelection("akka://application/user/myActor");

or anything else
?

Regards,
   MFarid

-- 
>>  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] Re: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Viktor Klang]

Fair enough :)

I have some longer term plans providing more interesting execution engines
(not based on ExecutorService),
we'll see what the future holds. For all intents and purposes the
invitation to provide alternate implementations still stands :)

On Tue, May 26, 2015 at 7:10 PM, Guido Medina  wrote:

> It would only require a quick & dirty hack, but I'm still in the process
> of learning/developing a new project using Akka/Java 8 on a new job so time
> is not much on my side *-you know, the first 6 months of every new job-*
>
> There is no need to change the dispatcher but the Scala configurator
> should provide one of your lovely queues which I know are extremely
> optimized compared to ArrayBlockingQueue, LinkedBlocking and
> ConcurrentLinkedQueue Java counter parts, right?
>
> On Tue, May 26, 2015 at 5:54 PM, Viktor Klang 
> wrote:
>
>> Hi Guido,
>>
>> the good news is that the ExecutorService for the Dispatcher is pluggable
>> so you can provide your own, optimized version.
>>
>> Would love to hear how much of a difference it could make (JMH bench
>> would be superb).
>>
>> On Tue, May 26, 2015 at 6:35 PM, Guido Medina  wrote:
>>
>>> In fact, even more could be done here, I'm assuming there is a reason
>>> why mailboxes in general are not just blocking queues from Java, designers'
>>> decision to provide a faster queue, so, why not reuse the same logic for
>>> dispatchers?
>>>
>>> --
>>> >> 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 a topic in the
>> Google Groups "Akka User List" group.
>> To unsubscribe from this topic, visit
>> https://groups.google.com/d/topic/akka-user/Q_sgvACG8Nw/unsubscribe.
>> To unsubscribe from this group and all its topics, 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.
>



-- 
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] Re: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Guido Medina]

Sorry I replied directly to you, my mistake, pasting message here:

It would only require a quick & dirty hack, but I'm still in the process of 
learning/developing a new project using Akka/Java 8 on a new job so time is 
not much on my side *-you know, the first 6 months of every new job-*

There is no need to change the dispatcher but the Scala configurator should 
provide one of your lovely queues which I know are extremely optimized 
compared to ArrayBlockingQueue, LinkedBlocking and ConcurrentLinkedQueue 
Java counter parts, right?

-- 
>>  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] Re: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Guido Medina]

It would only require a quick & dirty hack, but I'm still in the process of
learning/developing a new project using Akka/Java 8 on a new job so time is
not much on my side *-you know, the first 6 months of every new job-*

There is no need to change the dispatcher but the Scala configurator should
provide one of your lovely queues which I know are extremely optimized
compared to ArrayBlockingQueue, LinkedBlocking and ConcurrentLinkedQueue
Java counter parts, right?

On Tue, May 26, 2015 at 5:54 PM, Viktor Klang 
wrote:

> Hi Guido,
>
> the good news is that the ExecutorService for the Dispatcher is pluggable
> so you can provide your own, optimized version.
>
> Would love to hear how much of a difference it could make (JMH bench would
> be superb).
>
> On Tue, May 26, 2015 at 6:35 PM, Guido Medina  wrote:
>
>> In fact, even more could be done here, I'm assuming there is a reason why
>> mailboxes in general are not just blocking queues from Java, designers'
>> decision to provide a faster queue, so, why not reuse the same logic for
>> dispatchers?
>>
>> --
>> >> 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 a topic in the
> Google Groups "Akka User List" group.
> To unsubscribe from this topic, visit
> https://groups.google.com/d/topic/akka-user/Q_sgvACG8Nw/unsubscribe.
> To unsubscribe from this group and all its topics, 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] Re: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Viktor Klang]

Hi Guido,

the good news is that the ExecutorService for the Dispatcher is pluggable
so you can provide your own, optimized version.

Would love to hear how much of a difference it could make (JMH bench would
be superb).

On Tue, May 26, 2015 at 6:35 PM, Guido Medina  wrote:

> In fact, even more could be done here, I'm assuming there is a reason why
> mailboxes in general are not just blocking queues from Java, designers'
> decision to provide a faster queue, so, why not reuse the same logic for
> dispatchers?
>
> --
> >> 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] state of backpressure in websockets?

[Sam Halliday]

Hi all,

I have another thread about retrofitting wandoulabs websockets to use 
akka-io, which is proving painful, but I wanted to separate out this aspect 
of the questioning.

Before I invest any more time into it, I'd like to know if the new 
websockets implementation actually implements backpressure on the server 
and client side, for both reading and writing from the socket (there are 
four channels requiring backpressure in a single client/server connection).


Even if the implementation has backpressure at the IO level, it looks like 
the only way to create a Flow from an Actor is via Sink.actorRef (plus some 
other magic with Sources and the DSL that I haven't figured out yet) ... 
and that explicitly says in the documentation

   "there is no back-pressure signal from the destination actor, i.e. if 
the actor is not consuming the messages fast enough the mailbox of the 
actor will grow"

which means that passing off to an actor backend to implement the 
websockets server is ultimately not going to have any backpressure when 
reading off the socket.

I don't know what the situation is for writing to the socket, but certainly 
this is something that my current backend library is able to handle.


So is this reactive, or what?


Best regards,
Sam

-- 
>>  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: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Guido Medina]

In fact, even more could be done here, I'm assuming there is a reason why 
mailboxes in general are not just blocking queues from Java, designers' 
decision to provide a faster queue, so, why not reuse the same logic for 
dispatchers?

-- 
>>  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: PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Guido Medina]

Sorry I mixed oranges with apples here, I just looked deeper the code and 
no, my assumptions are *not* correct, the mailbox is one thing but the 
task-queue is a different story, it would be nice at the 
ThreadPoolExecutorConfigurator Scala class to have an option for single 
thread dispatchers *aka* PinnedDispatcher so that it is not a full blown 
blocking queue due to the fact there is only one thread executing tasks, 
I'm guessing there is one or more threads on the Akka system "loop" that 
can queue tasks into the PinnedDispatcher's queue which in this case it 
will always have a single task consumer -The PinnedDispatcher only thread-

-- 
>>  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] PinnedDispatcher with SingleConsumerOnlyUnboundedMailbox behaviour

[Guido Medina]

*Quick question:* If my default mailbox is a 
*SingleConsumerOnlyUnboundedMailbox* and my dispatcher a *PinnedDispatcher*, 
does the thread pool *-in this case a single thread-* de-queue tasks from 
it or is it there an intermediary Java blocking queue?

I'm assuming dispatchers in Akka are not necessary a copy of Fork-Join and 
Thread Pool executors from Java which for *SingleConsumerOnlyUnboundedMailbox 
*and *PinnedDispatcher* would be a waste of performance, I'm assuming it is 
de-queuing tasks directly from the mailbox, are my assumptions correct?

Best regards,

Guido.

-- 
>>  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] Closing resources in the Akka Stream

[Petr Janda]

Hi guys,

I was wondering what is the best practice used in Akka Streams to clean up 
opened resources. My example use case is the stream reading lines from the 
file, streaming them to Apache Kafka 
(using https://github.com/softwaremill/reactive-kafka subscriber). See the 
example code here:

val file = io.Source.fromFile(path)
val lines = file.getLines()
val kafka = new ReactiveKafka(host = "localhost:9092", zooKeeperHost = 
"localhost:2181")
val subscriber = kafka.publish("uppercaseStrings", "groupName", new 
StringEncoder())

Source(() => lines)
  .map(_.toUpperCase)
  .to(Sink(subscriber))
  .run()


As soon as the flow is done I would like to cleanup and close the file 
input stream. One way I used to go about that is to have Sink.onComplete or 
Sink.fold although this is not viable here. Also, ideally I would like to 
close the file in case of any error. 

Could you advice on any idiomatic way to do this? 

Thanks,
~Petr

 

-- 
>>  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] Flow from an Actor

[Sam Halliday]

To maybe try and formalise this a little bit more, and abstract away from 
WebSockets (that will only muddy the water).

Lets say we have an Actor already that looks like this

sealed trait Incoming
sealed trait Outgoing
class SimpleActor(upstream: ActorRef) extends Actor {
  def receive = {
case in: Incoming =>
   // work, including some upstream ! outgoing
case Ack =>
   // ack for the last message upstream
case other =>
   // work, including some upstream ! outgoing
  }
}

How do I wrap that as a Flow[Incoming, Outgoing, Unit] ?


On Tuesday, 26 May 2015 14:09:45 UTC+1, Sam Halliday wrote:
>
> Re: asyncyMap. I don't think that is going to work, there is no implied 
> single response to each query (which I gather is what you're suggesting)? 
> And I need some way of receiving new messages from upstream.
>
> The existing Actor is both a sink (i.e. it consumes messages from 
> upstream, not necessarily responding to each one) and a source (i.e. it can 
> send an effectively infinite number of messages). It is using backpressure, 
> but only using its own `Ack` message.
>
> For some context, I'm retrofitting some code that is using this WebSockets 
> layer around wandoulabs, e.g. 
> https://github.com/smootoo/simple-spray-websockets/blob/master/src/test/scala/org/suecarter/websocket/WebSocketSpec.scala#L150
>
> But the newly released akka-io layer expects a Flow.
>
> The `Ack` is being received when messages were sent directly to the I/O 
> layer. Presumably, the backpressure is implemented differently now... 
> although I am not sure how yet. That's the second problem once I can 
> actually get everything hooked up.
>
>
> On Tuesday, 26 May 2015 13:57:43 UTC+1, √ wrote:
>>
>> Not knowing what your actor is trying to do, what about Flow.mapAsync + 
>> ask?
>>
>> -- 
>> Cheers,
>> √
>> On 26 May 2015 14:54, "Sam Halliday"  wrote:
>>
>>> Hi all,
>>>
>>> I need to interface an Actor with an API that requires a Flow.
>>>
>>> The actor can receive a sealed trait family of inputs and will only send 
>>> (a different) sealed family of outputs to upstream, so I suspect that will 
>>> help matters.
>>>
>>> Looking in FlowOps, it looks like I can create a Flow from a partial 
>>> function, but there isn't anything that would just simply take an ActorRef.
>>>
>>> Am I missing something trivial to just upgrade an ActoRef to a Flow? 
>>> (Obviously there is a bunch of extra messages the actor will have to 
>>> handle, such as backpressure messages etc... but assume that's all taken 
>>> care of)
>>>
>>> Best regards,
>>> Sam
>>>
>>> -- 
>>> >> 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.

Re: [akka-user] Cluster disconnection - "Failed to write message to the transport"

[Ran Nozik]

Hi Endre,

Thank you for your quick response.

I verified that the only protobuf version we use
is com.google.protobuf:protobuf-java:2.5.0 (no other versions in the
classpath).

I'm not sure I understood your question about the remoting. We have a
distributed system with many (backend) Android workers and one master
(frontend) node. They do not interact as client and server.

Regards,
Nozik

On Tue, May 26, 2015 at 4:21 PM, Endre Varga 
wrote:

> Caused by: com.google.protobuf.UninitializedMessageException: Message
> missing required fields:
> ... 30 more
> ]
>
> This very much looks like a serialization problem though. Do you maybe
> have a newer protobuf version on your classpath than the one Akka uses?
>
> Btw, why are you using akka-remoting between android systems? Don't forget
> that remoting and clustering are not client-server technologies but
> peer-to-peer technologies:
> http://doc.akka.io/docs/akka/2.3.11/general/remoting.html#Peer-to-Peer_vs__Client-Server
>
> -Endre
>
> On Tue, May 26, 2015 at 3:16 PM,  wrote:
>
>> Hi,
>>
>> I upgraded to 2.3.11 and the problem reproduced again.
>>
>> Thanks.
>>
>>
>> On Tuesday, May 26, 2015 at 3:12:38 PM UTC+3, √ wrote:
>>>
>>> Hi Mozik,
>>>
>>> please upgrade to the latest version and report back if you still have
>>> the same problem.
>>>
>>> On Tue, May 26, 2015 at 2:03 PM,  wrote:
>>>
 Hi Everyone,

 I've been trying to set an Akka cluster with one master node and
 multiple workers. The workers are actor systems than run on Android
 emulators.
 As a start, I work with one worker (emulator). I verify that it
 successfully joins the cluster and start sending it messages, that are
 handled successfully. After some time (from 2-3 to 30-40 minutes), however,
 it disconnects from the cluster.
 Trying to figure out what causes the problem, I noticed that even if
 the worker is idle (no messages are sent), it disconnects from the cluster
 after some time.

 In the Android logcat, the following message is displayed:

 [ClusterSystem-akka.remote.default-remote-dispatcher-5] [akka.tcp://
> ClusterSystem@127.0.0.1:2553/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FClusterSystem%4010.141.4.104%3A2551-0]
> Association with remote system [akka.tcp://
> ClusterSystem@10.141.4.104:2551] has failed, address is now gated for
> [5000] ms. Reason is: [].


 and then:

 [ClusterSystem-cluster-dispatcher-15] [akka.tcp://
 ClusterSystem@127.0.0.1:2553/system/cluster/core/daemon] Cluster Node
 [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking node(s) as
 UNREACHABLE [Member(address = akka.tcp://
 ClusterSystem@10.141.4.104:2551, status = Up)]

 and eventually:

 [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
 Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
 auto-downing unreachable node [akka.tcp://
 ClusterSystem@10.141.4.104:2551]
 [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
 Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking
 unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] as [Down]
 [ClusterSystem-cluster-dispatcher-27] [Cluster(akka://ClusterSystem)]
 Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
 removing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551]


 After I subscribed to AssociationErrorEvent, I was able to get more
 details:

 AssociationErrorEvent has occurred: AssociationError [akka.tcp://
> ClusterSystem@127.0.0.1:2553] -> [akka.tcp://
> ClusterSystem@10.141.4.104:2551]: Error [] [
> akka.remote.EndpointException:
> at
> com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
> at
> akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
> at
> akka.remote.ContainerFormats$SelectionEnvelope$Builder.addPattern(ContainerFormats.java:931)
> at
> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:45)
> at
> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:43)
> at scala.collection.Iterator$class.foreach(Iterator.scala:727)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at
> akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:43)
> at
> akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25)
> at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
> at
> ak

Re: [akka-user] Cluster disconnection - "Failed to write message to the transport"

[Endre Varga]

Caused by: com.google.protobuf.UninitializedMessageException: Message
missing required fields:
... 30 more
]

This very much looks like a serialization problem though. Do you maybe have
a newer protobuf version on your classpath than the one Akka uses?

Btw, why are you using akka-remoting between android systems? Don't forget
that remoting and clustering are not client-server technologies but
peer-to-peer technologies:
http://doc.akka.io/docs/akka/2.3.11/general/remoting.html#Peer-to-Peer_vs__Client-Server

-Endre

On Tue, May 26, 2015 at 3:16 PM,  wrote:

> Hi,
>
> I upgraded to 2.3.11 and the problem reproduced again.
>
> Thanks.
>
>
> On Tuesday, May 26, 2015 at 3:12:38 PM UTC+3, √ wrote:
>>
>> Hi Mozik,
>>
>> please upgrade to the latest version and report back if you still have
>> the same problem.
>>
>> On Tue, May 26, 2015 at 2:03 PM,  wrote:
>>
>>> Hi Everyone,
>>>
>>> I've been trying to set an Akka cluster with one master node and
>>> multiple workers. The workers are actor systems than run on Android
>>> emulators.
>>> As a start, I work with one worker (emulator). I verify that it
>>> successfully joins the cluster and start sending it messages, that are
>>> handled successfully. After some time (from 2-3 to 30-40 minutes), however,
>>> it disconnects from the cluster.
>>> Trying to figure out what causes the problem, I noticed that even if the
>>> worker is idle (no messages are sent), it disconnects from the cluster
>>> after some time.
>>>
>>> In the Android logcat, the following message is displayed:
>>>
>>> [ClusterSystem-akka.remote.default-remote-dispatcher-5] [akka.tcp://
 ClusterSystem@127.0.0.1:2553/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FClusterSystem%4010.141.4.104%3A2551-0]
 Association with remote system [akka.tcp://
 ClusterSystem@10.141.4.104:2551] has failed, address is now gated for
 [5000] ms. Reason is: [].
>>>
>>>
>>> and then:
>>>
>>> [ClusterSystem-cluster-dispatcher-15] [akka.tcp://
>>> ClusterSystem@127.0.0.1:2553/system/cluster/core/daemon] Cluster Node
>>> [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking node(s) as
>>> UNREACHABLE [Member(address = akka.tcp://ClusterSystem@10.141.4.104:2551,
>>> status = Up)]
>>>
>>> and eventually:
>>>
>>> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
>>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
>>> auto-downing unreachable node [akka.tcp://
>>> ClusterSystem@10.141.4.104:2551]
>>> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
>>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking
>>> unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] as [Down]
>>> [ClusterSystem-cluster-dispatcher-27] [Cluster(akka://ClusterSystem)]
>>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
>>> removing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551]
>>>
>>>
>>> After I subscribed to AssociationErrorEvent, I was able to get more
>>> details:
>>>
>>> AssociationErrorEvent has occurred: AssociationError [akka.tcp://
 ClusterSystem@127.0.0.1:2553] -> [akka.tcp://
 ClusterSystem@10.141.4.104:2551]: Error [] [
 akka.remote.EndpointException:
 at
 com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
 at
 akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
 at
 akka.remote.ContainerFormats$SelectionEnvelope$Builder.addPattern(ContainerFormats.java:931)
 at
 akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:45)
 at
 akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:43)
 at scala.collection.Iterator$class.foreach(Iterator.scala:727)
 at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
 at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
 at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
 at
 akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:43)
 at
 akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25)
 at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
 at
 akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
 at
 akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
 at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
 at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:841)
 at akka.remote.EndpointWriter.writeSend(Endpoint.scala:742)
 at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:717)
 at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
 at akka.remote.EndpointActor.aroundReceiv

Re: [akka-user] Cluster disconnection - "Failed to write message to the transport"

[rnozik]

Hi,

I upgraded to 2.3.11 and the problem reproduced again.

Thanks.


On Tuesday, May 26, 2015 at 3:12:38 PM UTC+3, √ wrote:
>
> Hi Mozik,
>
> please upgrade to the latest version and report back if you still have the 
> same problem.
>
> On Tue, May 26, 2015 at 2:03 PM, > wrote:
>
>> Hi Everyone,
>>
>> I've been trying to set an Akka cluster with one master node and multiple 
>> workers. The workers are actor systems than run on Android emulators.
>> As a start, I work with one worker (emulator). I verify that it 
>> successfully joins the cluster and start sending it messages, that are 
>> handled successfully. After some time (from 2-3 to 30-40 minutes), however, 
>> it disconnects from the cluster.
>> Trying to figure out what causes the problem, I noticed that even if the 
>> worker is idle (no messages are sent), it disconnects from the cluster 
>> after some time. 
>>
>> In the Android logcat, the following message is displayed:
>>
>> [ClusterSystem-akka.remote.default-remote-dispatcher-5] [akka.tcp://
>>> ClusterSystem@127.0.0.1:2553/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FClusterSystem%4010.141.4.104%3A2551-0]
>>>  
>>> Association with remote system [akka.tcp://
>>> ClusterSystem@10.141.4.104:2551] has failed, address is now gated for 
>>> [5000] ms. Reason is: [].
>>
>>
>> and then:
>>
>> [ClusterSystem-cluster-dispatcher-15] [akka.tcp://
>> ClusterSystem@127.0.0.1:2553/system/cluster/core/daemon] Cluster Node 
>> [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking node(s) as 
>> UNREACHABLE [Member(address = akka.tcp://ClusterSystem@10.141.4.104:2551, 
>> status = Up)]
>>
>> and eventually:
>>
>> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)] 
>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is 
>> auto-downing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551
>> ]
>> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)] 
>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking 
>> unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] as [Down]
>> [ClusterSystem-cluster-dispatcher-27] [Cluster(akka://ClusterSystem)] 
>> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is 
>> removing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] 
>>
>>
>> After I subscribed to AssociationErrorEvent, I was able to get more 
>> details:
>>
>> AssociationErrorEvent has occurred: AssociationError [akka.tcp://
>>> ClusterSystem@127.0.0.1:2553] -> [akka.tcp://
>>> ClusterSystem@10.141.4.104:2551]: Error [] [
>>> akka.remote.EndpointException: 
>>> at 
>>> com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
>>> at 
>>> akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
>>> at 
>>> akka.remote.ContainerFormats$SelectionEnvelope$Builder.addPattern(ContainerFormats.java:931)
>>> at 
>>> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:45)
>>> at 
>>> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:43)
>>> at scala.collection.Iterator$class.foreach(Iterator.scala:727)
>>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
>>> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
>>> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
>>> at 
>>> akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:43)
>>> at 
>>> akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25)
>>> at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
>>> at 
>>> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
>>> at 
>>> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
>>> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
>>> at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:841)
>>> at akka.remote.EndpointWriter.writeSend(Endpoint.scala:742)
>>> at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:717)
>>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:410)
>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>>> at 
>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
>>> at 
>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
>>> at 
>>> scala.concurrent.forkjoin.ForkJoinPool.

Re: [akka-user] Flow from an Actor

[Sam Halliday]

Re: asyncyMap. I don't think that is going to work, there is no implied 
single response to each query (which I gather is what you're suggesting)? 
And I need some way of receiving new messages from upstream.

The existing Actor is both a sink (i.e. it consumes messages from upstream, 
not necessarily responding to each one) and a source (i.e. it can send an 
effectively infinite number of messages). It is using backpressure, but 
only using its own `Ack` message.

For some context, I'm retrofitting some code that is using this WebSockets 
layer around wandoulabs, e.g. 
https://github.com/smootoo/simple-spray-websockets/blob/master/src/test/scala/org/suecarter/websocket/WebSocketSpec.scala#L150

But the newly released akka-io layer expects a Flow.

The `Ack` is being received when messages were sent directly to the I/O 
layer. Presumably, the backpressure is implemented differently now... 
although I am not sure how yet. That's the second problem once I can 
actually get everything hooked up.


On Tuesday, 26 May 2015 13:57:43 UTC+1, √ wrote:
>
> Not knowing what your actor is trying to do, what about Flow.mapAsync + 
> ask?
>
> -- 
> Cheers,
> √
> On 26 May 2015 14:54, "Sam Halliday" > 
> wrote:
>
>> Hi all,
>>
>> I need to interface an Actor with an API that requires a Flow.
>>
>> The actor can receive a sealed trait family of inputs and will only send 
>> (a different) sealed family of outputs to upstream, so I suspect that will 
>> help matters.
>>
>> Looking in FlowOps, it looks like I can create a Flow from a partial 
>> function, but there isn't anything that would just simply take an ActorRef.
>>
>> Am I missing something trivial to just upgrade an ActoRef to a Flow? 
>> (Obviously there is a bunch of extra messages the actor will have to 
>> handle, such as backpressure messages etc... but assume that's all taken 
>> care of)
>>
>> Best regards,
>> Sam
>>
>> -- 
>> >> 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.

Re: [akka-user] Flow from an Actor

[Viktor Klang]

Not knowing what your actor is trying to do, what about Flow.mapAsync + ask?

-- 
Cheers,
√
On 26 May 2015 14:54, "Sam Halliday"  wrote:

> Hi all,
>
> I need to interface an Actor with an API that requires a Flow.
>
> The actor can receive a sealed trait family of inputs and will only send
> (a different) sealed family of outputs to upstream, so I suspect that will
> help matters.
>
> Looking in FlowOps, it looks like I can create a Flow from a partial
> function, but there isn't anything that would just simply take an ActorRef.
>
> Am I missing something trivial to just upgrade an ActoRef to a Flow?
> (Obviously there is a bunch of extra messages the actor will have to
> handle, such as backpressure messages etc... but assume that's all taken
> care of)
>
> Best regards,
> Sam
>
> --
> >> 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] Flow from an Actor

[Sam Halliday]

Hi all,

I need to interface an Actor with an API that requires a Flow.

The actor can receive a sealed trait family of inputs and will only send (a 
different) sealed family of outputs to upstream, so I suspect that will 
help matters.

Looking in FlowOps, it looks like I can create a Flow from a partial 
function, but there isn't anything that would just simply take an ActorRef.

Am I missing something trivial to just upgrade an ActoRef to a Flow? 
(Obviously there is a bunch of extra messages the actor will have to 
handle, such as backpressure messages etc... but assume that's all taken 
care of)

Best regards,
Sam

-- 
>>  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] how does the formFields directive work?

[Ian Phillips]

If I try to follow the example from the documentation it doesn't compile.

val route =
  formFields('color, 'age.as[Int]) { (color, age) =>
complete(s"The color is '$color' and the age ten years ago was ${age - 
10}")
  }


I get the following error:

too many arguments for method formFields: (pdm: 
akka.http.scaladsl.server.directives.FormFieldDirectives.FieldMagnet)pdm.Out
formFields('color, 'age.as[Int]) { (color, age) =>
  ^
one error found

Is the documentation wrong, or (more likely) what stupid mistake am I 
making with this? In case it makes a difference here, I'm importing:

import akka.http.scaladsl.server._
import akka.http.scaladsl.server.Directives._


Cheers,
Ian.

-- 
>>  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] Cluster disconnection - "Failed to write message to the transport"

[Viktor Klang]

Hi Mozik,

please upgrade to the latest version and report back if you still have the
same problem.

On Tue, May 26, 2015 at 2:03 PM,  wrote:

> Hi Everyone,
>
> I've been trying to set an Akka cluster with one master node and multiple
> workers. The workers are actor systems than run on Android emulators.
> As a start, I work with one worker (emulator). I verify that it
> successfully joins the cluster and start sending it messages, that are
> handled successfully. After some time (from 2-3 to 30-40 minutes), however,
> it disconnects from the cluster.
> Trying to figure out what causes the problem, I noticed that even if the
> worker is idle (no messages are sent), it disconnects from the cluster
> after some time.
>
> In the Android logcat, the following message is displayed:
>
> [ClusterSystem-akka.remote.default-remote-dispatcher-5] [akka.tcp://
>> ClusterSystem@127.0.0.1:2553/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FClusterSystem%4010.141.4.104%3A2551-0]
>> Association with remote system [akka.tcp://
>> ClusterSystem@10.141.4.104:2551] has failed, address is now gated for
>> [5000] ms. Reason is: [].
>
>
> and then:
>
> [ClusterSystem-cluster-dispatcher-15] [akka.tcp://
> ClusterSystem@127.0.0.1:2553/system/cluster/core/daemon] Cluster Node
> [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking node(s) as
> UNREACHABLE [Member(address = akka.tcp://ClusterSystem@10.141.4.104:2551,
> status = Up)]
>
> and eventually:
>
> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
> auto-downing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551]
> [ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)]
> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking
> unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] as [Down]
> [ClusterSystem-cluster-dispatcher-27] [Cluster(akka://ClusterSystem)]
> Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is
> removing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551]
>
>
> After I subscribed to AssociationErrorEvent, I was able to get more
> details:
>
> AssociationErrorEvent has occurred: AssociationError [akka.tcp://
>> ClusterSystem@127.0.0.1:2553] -> [akka.tcp://
>> ClusterSystem@10.141.4.104:2551]: Error [] [
>> akka.remote.EndpointException:
>> at
>> com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
>> at
>> akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
>> at
>> akka.remote.ContainerFormats$SelectionEnvelope$Builder.addPattern(ContainerFormats.java:931)
>> at
>> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:45)
>> at
>> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:43)
>> at scala.collection.Iterator$class.foreach(Iterator.scala:727)
>> at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
>> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
>> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
>> at
>> akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:43)
>> at
>> akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25)
>> at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
>> at
>> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
>> at
>> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
>> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
>> at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:841)
>> at akka.remote.EndpointWriter.writeSend(Endpoint.scala:742)
>> at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:717)
>> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:410)
>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>> at
>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
>> at
>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
>> at
>> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
>> at
>> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>> Caused by: com.google.protobuf.UninitializedMessageException: Message
>> missing required fields:
>> ... 30 more
>> ]
>> akka.remote.EndpointExce

[akka-user] Cluster disconnection - "Failed to write message to the transport"

[rnozik]

Hi Everyone,

I've been trying to set an Akka cluster with one master node and multiple 
workers. The workers are actor systems than run on Android emulators.
As a start, I work with one worker (emulator). I verify that it 
successfully joins the cluster and start sending it messages, that are 
handled successfully. After some time (from 2-3 to 30-40 minutes), however, 
it disconnects from the cluster.
Trying to figure out what causes the problem, I noticed that even if the 
worker is idle (no messages are sent), it disconnects from the cluster 
after some time. 

In the Android logcat, the following message is displayed:

[ClusterSystem-akka.remote.default-remote-dispatcher-5] 
> [akka.tcp://ClusterSystem@127.0.0.1:2553/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FClusterSystem%4010.141.4.104%3A2551-0]
>  
> Association with remote system [akka.tcp://ClusterSystem@10.141.4.104:2551] 
> has failed, address is now gated for [5000] ms. Reason is: [].


and then:

[ClusterSystem-cluster-dispatcher-15] 
[akka.tcp://ClusterSystem@127.0.0.1:2553/system/cluster/core/daemon] 
Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking node(s) as 
UNREACHABLE [Member(address = akka.tcp://ClusterSystem@10.141.4.104:2551, 
status = Up)]

and eventually:

[ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)] 
Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is 
auto-downing unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551]
[ClusterSystem-cluster-dispatcher-26] [Cluster(akka://ClusterSystem)] 
Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Marking 
unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] as [Down]
[ClusterSystem-cluster-dispatcher-27] [Cluster(akka://ClusterSystem)] 
Cluster Node [akka.tcp://ClusterSystem@127.0.0.1:2553] - Leader is removing 
unreachable node [akka.tcp://ClusterSystem@10.141.4.104:2551] 


After I subscribed to AssociationErrorEvent, I was able to get more details:

AssociationErrorEvent has occurred: AssociationError 
> [akka.tcp://ClusterSystem@127.0.0.1:2553] -> 
> [akka.tcp://ClusterSystem@10.141.4.104:2551]: Error [] [
> akka.remote.EndpointException: 
> at 
> com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
> at 
> akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
> at 
> akka.remote.ContainerFormats$SelectionEnvelope$Builder.addPattern(ContainerFormats.java:931)
> at 
> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:45)
> at 
> akka.remote.serialization.MessageContainerSerializer$$anonfun$serializeSelection$1.apply(MessageContainerSerializer.scala:43)
> at scala.collection.Iterator$class.foreach(Iterator.scala:727)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at 
> akka.remote.serialization.MessageContainerSerializer.serializeSelection(MessageContainerSerializer.scala:43)
> at 
> akka.remote.serialization.MessageContainerSerializer.toBinary(MessageContainerSerializer.scala:25)
> at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:36)
> at 
> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
> at 
> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:842)
> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:57)
> at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:841)
> at akka.remote.EndpointWriter.writeSend(Endpoint.scala:742)
> at akka.remote.EndpointWriter$$anonfun$2.applyOrElse(Endpoint.scala:717)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:410)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at 
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
> at 
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
> at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at 
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> Caused by: com.google.protobuf.UninitializedMessageException: Message 
> missing required fields: 
> ... 30 more
> ]
> akka.remote.EndpointException: 
> at 
> com.google.protobuf.AbstractMessage$Builder.newUninitializedMessageException(AbstractMessage.java:770)
> at 
> akka.remote.ContainerFormats$Selection$Builder.build(ContainerFormats.java:1513)
> at 
> akka.remote.ContainerFo

[akka-user] Re: Migrating from wandoulabs websockets to stock akka-io

[Sam Halliday]

Thanks Arnaud!

The key is in this line

  
https://github.com/jrudolph/akka-http-scala-js-websocket-chat/blob/239af857da2f174ea1624a84b0861c42cf4d1f2d/backend/src/main/scala/example/akkawschat/Webservice.scala#L33

so on the server side, the REST endpoint upgrades to a WebSocket `Flow` 
(i.e. akka streams API) via a directive.

My legacy code is just an Actor, so I'll have to reimplement my marshalling 
layer and so on around this (or maybe marshalling is already handled).

I've had a look at the Akka Streams documentation but it is not clear to me 
how to manually create a Flow from an Actor. I may have to read this a few 
more times, but I don't think it contains the information that I need... 
there doesn't appear to be a `Flow.actorRef`

http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/stream-integrations.html



On Saturday, 23 May 2015 18:49:26 UTC+1, Arnaud Gourlay wrote:
>
> Hi Sam,
>
> I am also really interested in migrating from Wandoulabs to the new Akka 
> WS implementation.
> So far this is the best example I could find 
> https://github.com/jrudolph/akka-http-scala-js-websocket-chat
>
> Hope this helps,
> Arnaud
>
> On Saturday, May 23, 2015 at 12:38:22 PM UTC+2, Sam Halliday wrote:
>>
>> Hi all,
>>
>> I'm very excited that akka-io now has WebSocket support.
>>
>> In ENSIME, we're planning on using this wrapper over wandoulab's 
>> websockets
>>
>>   https://github.com/smootoo/simple-spray-websockets
>>
>> to easily create a REST/WebSockets endpoint with JSON marshalling for a 
>> sealed family, with backpressure.
>>
>> Smootoo's wrapper works really well, and I have had the pleasure of using 
>> it in a corporate environment so I trust it to be stable.
>>
>>
>> For future proofing, it would seem sensible to move to stock akka-io for 
>> WebSockets, so I'm considering sending a PR to retrofit the wrapper. I have 
>> a couple of questions about that:
>>
>> 1. does akka-io's HTTP singleton actor support WebSockets now? That was 
>> the big caveat about using wandoulabs. It means all kinds of workarounds if 
>> you want to just use HTTP in the same actor system.
>>
>> 2. is there a migration guide for wandoulabs to akka-io? Or would it be 
>> best just to rewrite the wrapper from scratch on top of akka-io?
>>
>> 3. where is the documentation? This just has a big TODO on it
>>
>>   
>> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/client-side/websocket-support.html
>>   
>> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/http/routing-dsl/websocket-support.html
>>
>> I can't even find any examples. I guess the key thing is the handshaking, 
>> which would mean rewriting this bit (and the corresponding client side 
>> handshake)
>>
>>   
>> https://github.com/smootoo/simple-spray-websockets/blob/master/src/main/scala/org/suecarter/websocket/WebSocket.scala#L167
>>
>> Best regards,
>> Sam
>>
>

-- 
>>  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] filter dead letters by message

[Patrik Nordwall]

ok, thanks

On Tue, May 26, 2015 at 1:03 PM, Sam Halliday 
wrote:

> re-raised as a ticket, because of the CLA
>
> https://github.com/akka/akka/issues/17572
>
>
> On Tuesday, 26 May 2015 11:32:39 UTC+1, Sam Halliday wrote:
>>
>> PR it is
>>
>> https://github.com/akka/akka/pull/17571
>>
>> Note that your scalariform version is old and hence conflicts with
>> anything using a more recent version (including ensime-sbt).
>>
>>
>> On Tuesday, 26 May 2015 10:59:57 UTC+1, Patrik Nordwall wrote:
>>>
>>>
>>>
>>> On Tue, May 26, 2015 at 11:31 AM, Sam Halliday 
>>> wrote:
>>>
 Thanks Patrick!

 DeadLetterSuppression is indeed what I need to add. Unfortunately, it's
 on one of your messages. Can you please add it to Tcp.Close?


 https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/io/Tcp.scala#L189

 Or do you really want a PR for this one line change?

>>>
>>> It would be awesome if the community would be able to help us spot these
>>> things. Preferably with a pull request or else please create an issue
>>> .
>>>
>>>


 On Tuesday, 26 May 2015 10:12:47 UTC+1, Patrik Nordwall wrote:
>
> Hi Sam,
>
> You can do that as described here
> http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
> and in the link to the Event Stream from there.
>
> However, we would like to silence logging of messages that are
> "expected" to go to deadLetters by adding the marker
> `DeadLetterSuppression` to such messages. It would be great if you (and
> other users) can open pull requests (and issues) when you find such
> messages.
>
> Thanks,
> Patrik
>
> On Fri, May 22, 2015 at 6:37 PM, Sam Halliday 
> wrote:
>
>> Hi all,
>>
>> Some messages that arrive in dead letters are of no concern at all,
>> such as `Tcp.Close` (which happens a lot when using Akka IO).
>>
>> Is there any way to filter out the logging of these particular dead
>> letter messages so that they don't clutter up the log?
>>
>> Best regards,
>> Sam
>>
>> --
>> >> 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.
>>
>
>
>
> --
>
> Patrik Nordwall
> Typesafe  -  Reactive apps on the JVM
> Twitter: @patriknw
>
>   --
 >> 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.

>>>
>>>
>>>
>>> --
>>>
>>> Patrik Nordwall
>>> Typesafe  -  Reactive apps on the JVM
>>> Twitter: @patriknw
>>>
>>>   --
> >> 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.
>



-- 

Patrik Nordwall
Typesafe  -  Reactive apps on the JVM
Twitter: @patriknw

-- 
>>  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 re

Re: [akka-user] filter dead letters by message

[Sam Halliday]

re-raised as a ticket, because of the CLA

https://github.com/akka/akka/issues/17572

On Tuesday, 26 May 2015 11:32:39 UTC+1, Sam Halliday wrote:
>
> PR it is
>
> https://github.com/akka/akka/pull/17571
>
> Note that your scalariform version is old and hence conflicts with 
> anything using a more recent version (including ensime-sbt).
>
>
> On Tuesday, 26 May 2015 10:59:57 UTC+1, Patrik Nordwall wrote:
>>
>>
>>
>> On Tue, May 26, 2015 at 11:31 AM, Sam Halliday  
>> wrote:
>>
>>> Thanks Patrick!
>>>
>>> DeadLetterSuppression is indeed what I need to add. Unfortunately, it's 
>>> on one of your messages. Can you please add it to Tcp.Close?
>>>
>>>
>>> https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/io/Tcp.scala#L189
>>>
>>> Or do you really want a PR for this one line change?
>>>
>>
>> It would be awesome if the community would be able to help us spot these 
>> things. Preferably with a pull request or else please create an issue 
>> .
>>  
>>
>>>
>>>
>>> On Tuesday, 26 May 2015 10:12:47 UTC+1, Patrik Nordwall wrote:

 Hi Sam,

 You can do that as described here 
 http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
  
 and in the link to the Event Stream from there.

 However, we would like to silence logging of messages that are 
 "expected" to go to deadLetters by adding the marker 
 `DeadLetterSuppression` to such messages. It would be great if you (and 
 other users) can open pull requests (and issues) when you find such 
 messages.

 Thanks,
 Patrik

 On Fri, May 22, 2015 at 6:37 PM, Sam Halliday  
 wrote:

> Hi all,
>
> Some messages that arrive in dead letters are of no concern at all, 
> such as `Tcp.Close` (which happens a lot when using Akka IO).
>
> Is there any way to filter out the logging of these particular dead 
> letter messages so that they don't clutter up the log?
>
> Best regards,
> Sam
>
> -- 
> >> 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.
>



 -- 

 Patrik Nordwall
 Typesafe  -  Reactive apps on the JVM
 Twitter: @patriknw

   -- 
>>> >> 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.
>>>
>>
>>
>>
>> -- 
>>
>> Patrik Nordwall
>> Typesafe  -  Reactive apps on the JVM
>> Twitter: @patriknw
>>
>>  

-- 
>>  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] filter dead letters by message

[Sam Halliday]

PR it is

https://github.com/akka/akka/pull/17571

Note that your scalariform version is old and hence conflicts with anything 
using a more recent version (including ensime-sbt).


On Tuesday, 26 May 2015 10:59:57 UTC+1, Patrik Nordwall wrote:
>
>
>
> On Tue, May 26, 2015 at 11:31 AM, Sam Halliday  > wrote:
>
>> Thanks Patrick!
>>
>> DeadLetterSuppression is indeed what I need to add. Unfortunately, it's 
>> on one of your messages. Can you please add it to Tcp.Close?
>>
>>
>> https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/io/Tcp.scala#L189
>>
>> Or do you really want a PR for this one line change?
>>
>
> It would be awesome if the community would be able to help us spot these 
> things. Preferably with a pull request or else please create an issue 
> .
>  
>
>>
>>
>> On Tuesday, 26 May 2015 10:12:47 UTC+1, Patrik Nordwall wrote:
>>>
>>> Hi Sam,
>>>
>>> You can do that as described here 
>>> http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
>>>  
>>> and in the link to the Event Stream from there.
>>>
>>> However, we would like to silence logging of messages that are 
>>> "expected" to go to deadLetters by adding the marker 
>>> `DeadLetterSuppression` to such messages. It would be great if you (and 
>>> other users) can open pull requests (and issues) when you find such 
>>> messages.
>>>
>>> Thanks,
>>> Patrik
>>>
>>> On Fri, May 22, 2015 at 6:37 PM, Sam Halliday  
>>> wrote:
>>>
 Hi all,

 Some messages that arrive in dead letters are of no concern at all, 
 such as `Tcp.Close` (which happens a lot when using Akka IO).

 Is there any way to filter out the logging of these particular dead 
 letter messages so that they don't clutter up the log?

 Best regards,
 Sam

 -- 
 >> 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.

>>>
>>>
>>>
>>> -- 
>>>
>>> Patrik Nordwall
>>> Typesafe  -  Reactive apps on the JVM
>>> Twitter: @patriknw
>>>
>>>   -- 
>> >> 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.
>>
>
>
>
> -- 
>
> Patrik Nordwall
> Typesafe  -  Reactive apps on the JVM
> Twitter: @patriknw
>
>  

-- 
>>  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 cluster with 1 seed node

[Guido Medina]

Dear Hakkers,

I have a system with micro-services and my custom cluster shard 
implementation *-a long story not to be told for now, glad to share my 
approach later-*, so I have N roles in the cluster and each role with one 
of the following types: *[singleton, evenly distributed, balanced]* where 
*evenly 
distributed* is what cluster sharding does with my own algorithm and 
*balanced* creates a circular cycler routing logic which is similar to 
round robin but dumber, it just keep cycling an atomic integer between 
routee 1..N *-I can share the code later if anyone wants it-*

So every time a micro-service comes up it subscribe to cluster events with 
initial cluster state and member up/removed and here I update a local cache *-a 
local cache per service of an address book where other actors of other 
roles are to avoid network look up-*

The problem I'm having is that because I'm using only one seed node *-a 
special micro-service that doesn't do any business logic, just coordination 
stuff-* and if I restart the seed node, other nodes don't try to rejoin it 
and now I have two clusters, one with the seed node and all of the other 
nodes, new nodes coming after this will just join the seed node cluster.

So I guess what I'm looking for is some sort of cluster seed node 
resilience configuration?

I have gone through the documentation over and over and done some googling 
for hours with no success,

Best regards,

Guido.

-- 
>>  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] Can an ActorPublisher put back pressure on the sender?

[Jeroen Rosenberg]

Ok, that makes sense or at least is consistent

On Tuesday, May 26, 2015 at 12:08:54 PM UTC+2, Patrik Nordwall wrote:
>
>
>
> On Tue, May 26, 2015 at 11:56 AM, Jeroen Rosenberg  > wrote:
>
>> Thnx! What will happen when I use Source.actorRef (as you suggested) with 
>> OverflowStrategy.backpressure?
>>
>
> An IllegalArgument("Backpressure overflowStrategy not supported") will be 
> thrown.
>
> I think that is missing from the documentation 
> .
>  
>
>>
>> On Friday, May 22, 2015 at 4:08:41 PM UTC+2, Patrik Nordwall wrote:
>>>
>>>
>>>
>>> On Thu, May 21, 2015 at 10:33 AM, Jeroen Rosenberg <
>>> jeroen.r...@gmail.com> wrote:
>>>
 I'm using Akka-Http 1.0-RC2 and I'm building a simple streaming server 
 (Chunked HTTP) and client using reactive streams / flow graphs. 

 My server looks like this (simplified version):

 object Server extends App {

   implicit val system = ActorSystem("Server")
   implicit val ec = system.dispatcher
   val (address, port) = ("127.0.0.1", 6000)

   implicit val materializer = 
 ActorFlowMaterializer(ActorFlowMaterializerSettings(system))

   val publisher = Source.actorPublisher(Props(new MyAwesomePublisher))

   val handler = Sink.foreach[Http.IncomingConnection] { con =>
   con handleWith Flow[HttpRequest].map { req =>
   HttpResponse(200).withEntity(Chunked(`application/json`, 
 publisher))
   }
   }

   (Http() bind (address, port) to handler run)
 }

 I can now consume this stream with my akka http client implementation 
 and 'slow down the stream' by applying backpressure. I deliberately slow 
 down my client side processing to trigger the backpressuring. Here's a 
 simplified version:

 class Client(processor: ActorRef) extends Actor {

   private implicit val executionContext = context.system.dispatcher
   private implicit val flowMaterializer: FlowMaterializer = 
 ActorFlowMaterializer(ActorFlowMaterializerSettings(context.system))

   val client =
 Http(context.system).outgoingConnection(host, port, settings = 
 ClientConnectionSettings(context.system))

   val decompress = Flow[ByteString].map {
 data => gunzip(data.toArray)
   }

   val buff = Flow[ByteString].buffer(1000, 
 OverflowStrategy.backpressure)

   val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20 millis, 
 context.system.scheduler)(Future.successful(x)) }

   val consumer = Flow[HttpResponse].map {
 data =>
   FlowGraph.closed() { implicit b =>
 import FlowGraph.Implicits._
 data.entity.dataBytes ~> slowFlow ~> buff ~> Sink.ignore
   }.run()
   }

   override def receive: Receive = {
 case query: String =>
   val req = HttpRequest(GET, "http://localhost:6000/api";)
 .withHeaders(
   Connection("Keep-Alive")
 )
   Source.single(req).via(client).via(consumer).to(Sink.onComplete {
 case Success(_) => println("Success!")
 case Failure(e) => println(s"Error: $e")
   }).run()
   }

 Because of 'slowFlow', I can see that my server 'slows down the stream' 
 (i.e. less throughput for this connected client). So, great!

 However, I wanted to handle the flow processing in another Actor, so I 
 used ActorPublisher and pipe the stream to it, using akka.pattern.pipe:

 class Client(processor: ActorRef) extends Actor {
   ...

   override def receive: Receive = {
 case query: String =>
   val req = HttpRequest(GET, endpoint)
 .withHeaders(
   `Accept-Encoding`(gzip),
   Connection("Keep-Alive")
 ) ~> authorize
   Source.single(req).via(client).runWith(Sink.head) pipeTo self
 case response: HttpResponse =>
   response.entity.dataBytes.map { dataByte =>
  processor ! dataByte
   }.to(Sink.ignore).run()
   }
 }

 class StreamProcessor extends ActorPublisher[ByteString] with Actor {
   override def receive: Actor.Receive = {
 case data: ByteString =>
   if (isActive && totalDemand > 0)
 onNext(data)
   }
 }

 ...
 // elsewhere I'm consuming this publisher


 val src = Source(ActorPublisher[ByteString](streamProcessor))

 FlowGraph.closed() { implicit b =>
 import FlowGraph.Implicits._

 val decompress = Flow[ByteString].map {
data => gunzip(data.toArray)
 }

 val buff = Flow[ByteString].buffer(1000, 
 OverflowStrategy.backpressure)
  val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20 
 millis, context.system.scheduler)(Future.successful(x)) }

 src ~> slowFlow ~> b

Re: [akka-user] Can an ActorPublisher put back pressure on the sender?

[Patrik Nordwall]

On Tue, May 26, 2015 at 11:56 AM, Jeroen Rosenberg <
jeroen.rosenb...@gmail.com> wrote:

> Thnx! What will happen when I use Source.actorRef (as you suggested) with
> OverflowStrategy.backpressure?
>

An IllegalArgument("Backpressure overflowStrategy not supported") will be
thrown.

I think that is missing from the documentation
.


>
> On Friday, May 22, 2015 at 4:08:41 PM UTC+2, Patrik Nordwall wrote:
>>
>>
>>
>> On Thu, May 21, 2015 at 10:33 AM, Jeroen Rosenberg > > wrote:
>>
>>> I'm using Akka-Http 1.0-RC2 and I'm building a simple streaming server
>>> (Chunked HTTP) and client using reactive streams / flow graphs.
>>>
>>> My server looks like this (simplified version):
>>>
>>> object Server extends App {
>>>
>>>   implicit val system = ActorSystem("Server")
>>>   implicit val ec = system.dispatcher
>>>   val (address, port) = ("127.0.0.1", 6000)
>>>
>>>   implicit val materializer =
>>> ActorFlowMaterializer(ActorFlowMaterializerSettings(system))
>>>
>>>   val publisher = Source.actorPublisher(Props(new MyAwesomePublisher))
>>>
>>>   val handler = Sink.foreach[Http.IncomingConnection] { con =>
>>>   con handleWith Flow[HttpRequest].map { req =>
>>>   HttpResponse(200).withEntity(Chunked(`application/json`,
>>> publisher))
>>>   }
>>>   }
>>>
>>>   (Http() bind (address, port) to handler run)
>>> }
>>>
>>> I can now consume this stream with my akka http client implementation
>>> and 'slow down the stream' by applying backpressure. I deliberately slow
>>> down my client side processing to trigger the backpressuring. Here's a
>>> simplified version:
>>>
>>> class Client(processor: ActorRef) extends Actor {
>>>
>>>   private implicit val executionContext = context.system.dispatcher
>>>   private implicit val flowMaterializer: FlowMaterializer =
>>> ActorFlowMaterializer(ActorFlowMaterializerSettings(context.system))
>>>
>>>   val client =
>>> Http(context.system).outgoingConnection(host, port, settings =
>>> ClientConnectionSettings(context.system))
>>>
>>>   val decompress = Flow[ByteString].map {
>>> data => gunzip(data.toArray)
>>>   }
>>>
>>>   val buff = Flow[ByteString].buffer(1000, OverflowStrategy.backpressure)
>>>
>>>   val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20 millis,
>>> context.system.scheduler)(Future.successful(x)) }
>>>
>>>   val consumer = Flow[HttpResponse].map {
>>> data =>
>>>   FlowGraph.closed() { implicit b =>
>>> import FlowGraph.Implicits._
>>> data.entity.dataBytes ~> slowFlow ~> buff ~> Sink.ignore
>>>   }.run()
>>>   }
>>>
>>>   override def receive: Receive = {
>>> case query: String =>
>>>   val req = HttpRequest(GET, "http://localhost:6000/api";)
>>> .withHeaders(
>>>   Connection("Keep-Alive")
>>> )
>>>   Source.single(req).via(client).via(consumer).to(Sink.onComplete {
>>> case Success(_) => println("Success!")
>>> case Failure(e) => println(s"Error: $e")
>>>   }).run()
>>>   }
>>>
>>> Because of 'slowFlow', I can see that my server 'slows down the stream'
>>> (i.e. less throughput for this connected client). So, great!
>>>
>>> However, I wanted to handle the flow processing in another Actor, so I
>>> used ActorPublisher and pipe the stream to it, using akka.pattern.pipe:
>>>
>>> class Client(processor: ActorRef) extends Actor {
>>>   ...
>>>
>>>   override def receive: Receive = {
>>> case query: String =>
>>>   val req = HttpRequest(GET, endpoint)
>>> .withHeaders(
>>>   `Accept-Encoding`(gzip),
>>>   Connection("Keep-Alive")
>>> ) ~> authorize
>>>   Source.single(req).via(client).runWith(Sink.head) pipeTo self
>>> case response: HttpResponse =>
>>>   response.entity.dataBytes.map { dataByte =>
>>>  processor ! dataByte
>>>   }.to(Sink.ignore).run()
>>>   }
>>> }
>>>
>>> class StreamProcessor extends ActorPublisher[ByteString] with Actor {
>>>   override def receive: Actor.Receive = {
>>> case data: ByteString =>
>>>   if (isActive && totalDemand > 0)
>>> onNext(data)
>>>   }
>>> }
>>>
>>> ...
>>> // elsewhere I'm consuming this publisher
>>>
>>>
>>> val src = Source(ActorPublisher[ByteString](streamProcessor))
>>>
>>> FlowGraph.closed() { implicit b =>
>>> import FlowGraph.Implicits._
>>>
>>> val decompress = Flow[ByteString].map {
>>>data => gunzip(data.toArray)
>>> }
>>>
>>> val buff = Flow[ByteString].buffer(1000,
>>> OverflowStrategy.backpressure)
>>>  val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20
>>> millis, context.system.scheduler)(Future.successful(x)) }
>>>
>>> src ~> slowFlow ~> buff ~> Sink.ignore
>>> }.run()
>>>
>>>
>>> This works fine, however in StreamProcessor (the ActorPublisher) it
>>> seems if I'm getting more data then I demand the only thing I can do is
>>> drop the messages. Can I apply backpressure here to the sender / upstream?
>>>
>>
>> Then you have to 

Re: [akka-user] filter dead letters by message

[Patrik Nordwall]

On Tue, May 26, 2015 at 11:31 AM, Sam Halliday 
wrote:

> Thanks Patrick!
>
> DeadLetterSuppression is indeed what I need to add. Unfortunately, it's on
> one of your messages. Can you please add it to Tcp.Close?
>
>
> https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/io/Tcp.scala#L189
>
> Or do you really want a PR for this one line change?
>

It would be awesome if the community would be able to help us spot these
things. Preferably with a pull request or else please create an issue
.


>
>
> On Tuesday, 26 May 2015 10:12:47 UTC+1, Patrik Nordwall wrote:
>>
>> Hi Sam,
>>
>> You can do that as described here
>> http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
>> and in the link to the Event Stream from there.
>>
>> However, we would like to silence logging of messages that are "expected"
>> to go to deadLetters by adding the marker `DeadLetterSuppression` to such
>> messages. It would be great if you (and other users) can open pull requests
>> (and issues) when you find such messages.
>>
>> Thanks,
>> Patrik
>>
>> On Fri, May 22, 2015 at 6:37 PM, Sam Halliday 
>> wrote:
>>
>>> Hi all,
>>>
>>> Some messages that arrive in dead letters are of no concern at all, such
>>> as `Tcp.Close` (which happens a lot when using Akka IO).
>>>
>>> Is there any way to filter out the logging of these particular dead
>>> letter messages so that they don't clutter up the log?
>>>
>>> Best regards,
>>> Sam
>>>
>>> --
>>> >> 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.
>>>
>>
>>
>>
>> --
>>
>> Patrik Nordwall
>> Typesafe  -  Reactive apps on the JVM
>> Twitter: @patriknw
>>
>>   --
> >> 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.
>



-- 

Patrik Nordwall
Typesafe  -  Reactive apps on the JVM
Twitter: @patriknw

-- 
>>  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] Can an ActorPublisher put back pressure on the sender?

[Jeroen Rosenberg]

Thnx! What will happen when I use Source.actorRef (as you suggested) with 
OverflowStrategy.backpressure?

On Friday, May 22, 2015 at 4:08:41 PM UTC+2, Patrik Nordwall wrote:
>
>
>
> On Thu, May 21, 2015 at 10:33 AM, Jeroen Rosenberg  > wrote:
>
>> I'm using Akka-Http 1.0-RC2 and I'm building a simple streaming server 
>> (Chunked HTTP) and client using reactive streams / flow graphs. 
>>
>> My server looks like this (simplified version):
>>
>> object Server extends App {
>>
>>   implicit val system = ActorSystem("Server")
>>   implicit val ec = system.dispatcher
>>   val (address, port) = ("127.0.0.1", 6000)
>>
>>   implicit val materializer = 
>> ActorFlowMaterializer(ActorFlowMaterializerSettings(system))
>>
>>   val publisher = Source.actorPublisher(Props(new MyAwesomePublisher))
>>
>>   val handler = Sink.foreach[Http.IncomingConnection] { con =>
>>   con handleWith Flow[HttpRequest].map { req =>
>>   HttpResponse(200).withEntity(Chunked(`application/json`, 
>> publisher))
>>   }
>>   }
>>
>>   (Http() bind (address, port) to handler run)
>> }
>>
>> I can now consume this stream with my akka http client implementation and 
>> 'slow down the stream' by applying backpressure. I deliberately slow down 
>> my client side processing to trigger the backpressuring. Here's a 
>> simplified version:
>>
>> class Client(processor: ActorRef) extends Actor {
>>
>>   private implicit val executionContext = context.system.dispatcher
>>   private implicit val flowMaterializer: FlowMaterializer = 
>> ActorFlowMaterializer(ActorFlowMaterializerSettings(context.system))
>>
>>   val client =
>> Http(context.system).outgoingConnection(host, port, settings = 
>> ClientConnectionSettings(context.system))
>>
>>   val decompress = Flow[ByteString].map {
>> data => gunzip(data.toArray)
>>   }
>>
>>   val buff = Flow[ByteString].buffer(1000, OverflowStrategy.backpressure)
>>
>>   val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20 millis, 
>> context.system.scheduler)(Future.successful(x)) }
>>
>>   val consumer = Flow[HttpResponse].map {
>> data =>
>>   FlowGraph.closed() { implicit b =>
>> import FlowGraph.Implicits._
>> data.entity.dataBytes ~> slowFlow ~> buff ~> Sink.ignore
>>   }.run()
>>   }
>>
>>   override def receive: Receive = {
>> case query: String =>
>>   val req = HttpRequest(GET, "http://localhost:6000/api";)
>> .withHeaders(
>>   Connection("Keep-Alive")
>> )
>>   Source.single(req).via(client).via(consumer).to(Sink.onComplete {
>> case Success(_) => println("Success!")
>> case Failure(e) => println(s"Error: $e")
>>   }).run()
>>   }
>>
>> Because of 'slowFlow', I can see that my server 'slows down the stream' 
>> (i.e. less throughput for this connected client). So, great!
>>
>> However, I wanted to handle the flow processing in another Actor, so I 
>> used ActorPublisher and pipe the stream to it, using akka.pattern.pipe:
>>
>> class Client(processor: ActorRef) extends Actor {
>>   ...
>>
>>   override def receive: Receive = {
>> case query: String =>
>>   val req = HttpRequest(GET, endpoint)
>> .withHeaders(
>>   `Accept-Encoding`(gzip),
>>   Connection("Keep-Alive")
>> ) ~> authorize
>>   Source.single(req).via(client).runWith(Sink.head) pipeTo self
>> case response: HttpResponse =>
>>   response.entity.dataBytes.map { dataByte =>
>>  processor ! dataByte
>>   }.to(Sink.ignore).run()
>>   }
>> }
>>
>> class StreamProcessor extends ActorPublisher[ByteString] with Actor {
>>   override def receive: Actor.Receive = {
>> case data: ByteString =>
>>   if (isActive && totalDemand > 0)
>> onNext(data)
>>   }
>> }
>>
>> ...
>> // elsewhere I'm consuming this publisher
>>
>>
>> val src = Source(ActorPublisher[ByteString](streamProcessor))
>>
>> FlowGraph.closed() { implicit b =>
>> import FlowGraph.Implicits._
>>
>> val decompress = Flow[ByteString].map {
>>data => gunzip(data.toArray)
>> }
>>
>> val buff = Flow[ByteString].buffer(1000, 
>> OverflowStrategy.backpressure)
>>  val slowFlow = Flow[ByteString].mapAsync(1) { x => after(20 millis, 
>> context.system.scheduler)(Future.successful(x)) }
>>
>> src ~> slowFlow ~> buff ~> Sink.ignore
>> }.run()
>>
>>
>> This works fine, however in StreamProcessor (the ActorPublisher) it seems 
>> if I'm getting more data then I demand the only thing I can do is drop the 
>> messages. Can I apply backpressure here to the sender / upstream?
>>
>
> Then you have to use ordinary actor messages to implement your own flow 
> control, but it would be better if you could stay within the streams domain 
> and let it handle the backpressure.
>
> By the way, lets say that you wanted something like you here implemented 
> with the StreamProcessor ActorPublisher that is dropping messages if there 
> is no demand from downstream. Then you can instead use Source.ac

Re: [akka-user] filter dead letters by message

[Sam Halliday]

Thanks Patrick!

DeadLetterSuppression is indeed what I need to add. Unfortunately, it's on 
one of your messages. Can you please add it to Tcp.Close?

https://github.com/akka/akka/blob/master/akka-actor/src/main/scala/akka/io/Tcp.scala#L189

Or do you really want a PR for this one line change?


On Tuesday, 26 May 2015 10:12:47 UTC+1, Patrik Nordwall wrote:
>
> Hi Sam,
>
> You can do that as described here 
> http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
>  
> and in the link to the Event Stream from there.
>
> However, we would like to silence logging of messages that are "expected" 
> to go to deadLetters by adding the marker `DeadLetterSuppression` to such 
> messages. It would be great if you (and other users) can open pull requests 
> (and issues) when you find such messages.
>
> Thanks,
> Patrik
>
> On Fri, May 22, 2015 at 6:37 PM, Sam Halliday  > wrote:
>
>> Hi all,
>>
>> Some messages that arrive in dead letters are of no concern at all, such 
>> as `Tcp.Close` (which happens a lot when using Akka IO).
>>
>> Is there any way to filter out the logging of these particular dead 
>> letter messages so that they don't clutter up the log?
>>
>> Best regards,
>> Sam
>>
>> -- 
>> >> 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.
>>
>
>
>
> -- 
>
> Patrik Nordwall
> Typesafe  -  Reactive apps on the JVM
> Twitter: @patriknw
>
>  

-- 
>>  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] filter dead letters by message

[Patrik Nordwall]

Hi Sam,

You can do that as described here
http://doc.akka.io/docs/akka/2.3.11/scala/logging.html#Logging_of_Dead_Letters
and in the link to the Event Stream from there.

However, we would like to silence logging of messages that are "expected"
to go to deadLetters by adding the marker `DeadLetterSuppression` to such
messages. It would be great if you (and other users) can open pull requests
(and issues) when you find such messages.

Thanks,
Patrik

On Fri, May 22, 2015 at 6:37 PM, Sam Halliday 
wrote:

> Hi all,
>
> Some messages that arrive in dead letters are of no concern at all, such
> as `Tcp.Close` (which happens a lot when using Akka IO).
>
> Is there any way to filter out the logging of these particular dead letter
> messages so that they don't clutter up the log?
>
> Best regards,
> Sam
>
> --
> >> 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.
>



-- 

Patrik Nordwall
Typesafe  -  Reactive apps on the JVM
Twitter: @patriknw

-- 
>>  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.