Re: [akka-user][deprecated] logRequestResult - how to log ALL responses (including errors / non-200 statuses) ?
Hi François,
probably the problem is that the exception is only converted into a
response when the route is "sealed". This happens automatically when you
pass a route to `bindAndHandle`. Otherwise, you will have to manually seal
the route to see responses with logRequestResponse.
See
https://doc.akka.io/docs/akka-http/current/routing-dsl/routes.html#sealing-a-route
for more information about sealing a route.
We are also working on https://github.com/akka/akka-http/issues/2216 which
will hopefully soon provide a lower-level, more reliable way to log all
outgoing responses.
Johannes
On Thu, Dec 20, 2018 at 12:09 PM François Vatel wrote:
>
> Hello,
>
> I am using latest akka-http and have routing like this:
>
> val httpRoute =
> logRequestResult(showRequestResponse _) {
> pathPrefix("rpc") {
>
> session() { session =>
>
> ...
>
> }
> } ~
> }
>
>
> def showRequestResponse(request: HttpRequest) = {
> val startTime = DateUtils.now()
> val requestId = StringUtils.uniqueRandomKey(16, true)
>
> logger.debug(s"Request($requestId) is: " + request)
>
> (response: Any) => {
> val handlingTime = (DateUtils.now() - startTime).toInt
> logger.debug(s"Response($requestId)($handlingTime ms) is: " +
response)
> None
> }
> }
>
>
> "session" is a hand-made directive which creates some kind of Session
object from cookies.
>
> The problem is that the response is not logged if I throw an Exception
(which is correctly mapped to a response with status 401). What is wrong
here?
>
> --
>
*
> ** New discussion forum: https://discuss.akka.io/ replacing akka-user
google-group soon.
> ** This group will soon be put into read-only mode, and replaced by
discuss.akka.io
> ** More details:
https://akka.io/blog/news/2018/03/13/discuss.akka.io-announced
>
*
> >>>>>>>>>>
> >>>>>>>>>> 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 https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
--
Akka Team
Lightbend - Reactive apps on the JVM
Twitter: @akkateam
--
*
** New discussion forum: https://discuss.akka.io/ replacing akka-user
google-group soon.
** This group will soon be put into read-only mode, and replaced by
discuss.akka.io
** More details: https://akka.io/blog/news/2018/03/13/discuss.akka.io-announced
*
>>>>>>>>>>
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Question on Akka router
We have a nice sample/guide showing distributed workers pulling work rather than pushing that may be interesting to you: https://developer.lightbend.com/guides/akka-distributed-workers-scala/ That does however potentially create a bottleneck so if you really want to avoid that you cannot copy the exact design of the sample, but maybe ideas in it may prove valuable. -- Johan Akka Team On Thu, Feb 22, 2018 at 10:35 PM, Grace <alohagraceal...@gmail.com> wrote: > Thanks Patrik. > > In my use case, I have a few hundreds of client nodes. I have about 300 > worker nodes across 20 boxes. Due to a legacy limitation, one worker node > can have only one actor and therefore, one worker node is only allowed to > process one task at a time. Each task runs for a few seconds. > > If we go with router inside each client, do you have a suggestion which > routing machanism we use for fairer distribution? We are using pool instead > of group and we are currently using round-robin-pool. Haven't tested under > heavy load yet. Would round robin be a good choice? > > Thanks, > Grace > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Cancel Actor job
Maybe this distributed workers guide could be good inspiration: https://developer.lightbend.com/guides/akka-distributed-workers-scala/ -- Johan Akka Team On Fri, Feb 2, 2018 at 3:10 AM, Konrad “ktoso” Malawski < konrad.malaw...@lightbend.com> wrote: > Have a look at "cluster sharding”, it’s smarter than just routers. > > -- > Cheers, > Konrad 'ktoso <http://kto.so>' Malawski > Akka <http://akka.io/> @ Lightbend <http://lightbend.com/> > > On February 2, 2018 at 11:09:19, Richard Gong (gong...@gmail.com) wrote: > > That works in a single machine. Now take a step forward to have the > workers in a cluster. The master needs to dispatch jobs to workers. I'm > looking at the document of router, https://doc.akka.io/ > docs/akka/current/cluster-usage.html?language=scala# > router-with-pool-of-remote-deployed-routees. My thought is worker code > can be dynamically deployed and join the cluster when needed. Let's say I > have these 2 components running in different nodes. > >- Master: accepts jobs and dispatches them to WorkerAdvisor. >- WorkerAdvisor: creates an actual worker to run the job. > > Now how the Master knows dispatches a job to a WorkerAdvisor which is > free. Broadcasting job to all WorkerAdvisors doesn't seem to be a good way > because we don't want a job to run in multiple places. Maintaining state of > WorkerAdvisor in Master seems odd to me as well. > -- > >>>>>>>>>> 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 https://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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Mutual Authentication using Akka HTTP
Did you see the Mutual Auth section in the Akka HTTP docs? https://doc.akka.io/docs/akka-http/10.0.11/scala/http/server-side/server-https-support.html#mutual-authentication The synthetic Tls-Session-Info header should give you access to the client certs etc. -- Johan Akka Team On Mon, Jan 29, 2018 at 4:23 PM, David Leonhartsberger <leozi...@gmail.com> wrote: > I am currently faced with the task of implementing client authentication > via public key certificates for our HTTP endpoints. > > One additional complication/requirement is to be able to identify the > client which is connecting. > My current guess for this is to extract the Common Name from the > certificate therefor I must also be able to get programmatic access to the > certificate which the client is sending. > > Does somebody know howto implement this using akka http or on top of akka > http? > > I would be very thankful for any help I can get. > > Br > David > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka Default Dispatcher creating 9000+ Worker Threads!!
If you use Await.result, Await.ready or surround logic with scala.concurrent#blocking and use a FJP dispatcher, that will fork unbounded amounts of threads for those blocks, so that could perhaps explain why you are seing so many thread. Looking at stack dumps should show you what those threads are doing. If this is the problem the solution would be to not block or to run such things on a separate thread pool based dispatcher. -- Johan Akka Team On Tue, Jan 16, 2018 at 8:37 PM, <abhil...@zilingo.com> wrote: > Hey Folks, > > I've been working on a service thats built on Scala + Play (2.5.3), which > also makes some use of Akka actors (with no remote-ing). Recently i've > noticed that the service intermittently goes down due to OOM. Looked up the > .hprof dump files (set to be created on OOM failures) and it appears that > 9000+ (!) worker threads > (akka.dispatch.MonitorableThreadFactory$AkkaForkJoinWorkerThread) > were created named 'application-akka.actor.default-dispatcher-'. > Additionally, all these worker threads have no stack trace or a history of > any task assigned to them. > > Refer attached .csv of Eclipse Memory Analyzer's thread summary. > > Note : > 1) I haven't defined any custom Dispatcher / Execution Context / Thread > Pool. > 2) As per Play's docs, i've ensured that ActorSystems are only injected > and never created, hence only play's default actor system is used. > > I assumed that a max of 64 threads would be created since only the default > actor system was used, but these many threads is very strange. > > Any help with analysing this issue would be great, more specifically any > advice on how to ascertain how/why these 'ghost' workers threads are > created would be super useful. Any tools to monitor creation of worker > threads by Akka Dispatcher? > > -- Abhilash > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-streams: Increase processing rate:
It could be a good idea to take stages out, measure the throughput while adding a stage at a time to see what actually is limiting your throughput. If it in fact is Akka HTTP, then the default configuration for the connection pool only allows 4 connections per host, maybe increasing this (akka.http.host-connection-pool.max-connections) would help. It could actually be easier to reason about the flow by using singleRequest inside a mapAsync or a mapAsyncUnordered with a parallelism matching the max-connection per host setting. It could also make sense to insert async boundaries, to let the logic before the request execute in parallel with the request being executed. -- Johan Akka Team On Tue, Jan 9, 2018 at 4:23 AM, kor <khalid.r...@gmail.com> wrote: > Hello all. I have a requirement where I need to stream records from a > couchbase database, do some json transformations and post the result to a > rest endpoint (implemented with Akka-Http). I've tested that endpoint in > isolation using gatling, and it can handle at least 200 requests/sec. The > problem I am having when I run it, I cannot get the speed to go past 10 > requests/sec. I've tested the speed that records are being pulled out of > couchbase and got a rate of about 2700 records/sec so something is causing > backpressure to be applied to the source. I've also tried posting to a > different public endpoint (probably not implemented with akka-http) and the > speed increases to about 80 requests/sec. I've posted the code below. How > can I make things go faster? Disclaimer: I am new to akka-streams. > > code: https://gist.github.com/khalidr/1a392680c0ed744b2d3e5cec1e1f193a > > Any help would be appreciated. > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Balancing + Broadcast + Priority
Late answer, but better late than never, I guess. I'd recommend that you try to model this without reaching for special custom mailboxes or dispatchers but instead try to model the use case with actors: Making sure a worker splits up its work in smaller segments, sending a trigger to itself to trigger processing of next segment is a good way to be able to share updates that many workers can use, the messages are processed in the order they are put in the inbox, so this lets the worker go through all messages sent to it since last work-segment-start-message before it continues with the next. Work pulling is also almost always a better to organize work than balancing as it allows the worker to apply back pressure, see the distributed worker sample for one example of that: https://developer.lightbend.com/guides/akka-distributed-workers-scala/ -- Johan Akka Team On Wed, Dec 13, 2017 at 3:50 PM, Arnaud Malapert <malapert.arn...@gmail.com> wrote: > Dear akka users, > > I am starting with akka and trying to translate a muti-threaded/MPI > application to the actor paradigm. > I want to manage a set of workers with the following features : > - Balancing : balance workload across workers (the router uses a Balancing > Pool <https://doc.akka.io/docs/akka/current/routing.html#balancingpool>) > - Broadcast : send important updates that help the workers to execute jobs > (reduce runtime) > - Priority : recent updates must be read before each job start. > > But, the doc says: > *Do not use **Broadcast Messages > <https://doc.akka.io/docs/akka/current/routing.html#broadcast-messages>** > when you use **BalancingPool > <https://doc.akka.io/docs/akka/current/routing.html#balancing-pool>** for > routers, as described in **Specially Handled Messages > <https://doc.akka.io/docs/akka/current/routing.html#router-special-messages>* > *. * > Here is the story of my failure to find the right solution. > > Idea 1. Create a second group router that broadcasts the updates to > the workers. > Problem. Workers miss many updates. Performance degrades. However, it > works well with the SmallestMailboxPool > <https://doc.akka.io/docs/akka/current/routing.html#smallestmailboxpool> > because each worker has its own mailbox. > > Idea 2. Use some kind of 'updates pill'. Ask an actor to send updates > to the router before and after a job is taken by a worker. > Problem. Workers miss some updates. Performance sometimes degrades. > > Idea 3. Manage the message concurrent queue by myself, as before > Problem. It gives total control but, it is more complicated, > error-prone and scales less easily. > > Idea 4. Implement my own dispatcher > Problem. A little bit complicated for a beginner, no ? > > As it seems a common use case, I may get some help here. > > Best regards, > Arnaud > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: Graph materialized value of websocket upgrade
Flow.fromSinkAndSource(sink, source) allows you to have decoupled incoming and outgoing messages. Maybe this sample can be helpful: https://markatta.com/codemonkey/blog/2016/10/02/chat-with-akka-http-websockets/ Please don't ask a new question in an only somewhat related old thread, start a new thread instead, thanks! -- Johan Akka Team On Sun, Jan 28, 2018 at 12:48 PM, Thai <lgoptimusv...@gmail.com> wrote: > Hi Mathieu, > > Sorry to re-open a quite old topic. I'm looking for an example exactly > like this one "*a websocket server for which there is no 1-1 relation > between incoming and outgoing messages. The client can send messages at any > time which may or not have a reply from the server. The server can push > messages at any time which may or not have a reply from the client*". > It's really easy in non-Akka world (e.g. spring websocket, jetty websocket > ...). Do you have any example (scala version is fine too) for me to follow? > > Regards. > > On Tuesday, May 2, 2017 at 6:35:33 AM UTC-4, Mathieu Bruyen wrote: >> >> Hello, >> >> (this is more related to akka-http but there seem to be no dedicated >> group) >> >> I am trying to setup a websocket server for which there is no 1-1 >> relation between incoming and outgoing messages. The client can send >> messages at any time which may or not have a reply from the server. The >> server can push messages at any time which may or not have a reply from the >> client. I wanted to let actors handle and produce these messages. Following >> the doc >> <http://doc.akka.io/docs/akka-http/current/java/http/server-side/websocket-support.html#handling-messages> >> I am using Sink#actorRefWithAck to push incoming messages to the >> entrypoint/router actor and let it do it's job. For outgoing messages in >> the options suggested by the doc >> <http://doc.akka.io/docs/akka/current/scala/stream/stream-integrations.html#Integrating_with_Actors> >> I was planning to use Source#queue and give this queue to all the actors >> which might need to push to the queue. I targeted this solution because it >> allows for backpressure management. >> >> -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] PartitionWith with in order merge
Hi, not sure if you have solved this already, but one idea comes to my mind, where your partitioner would inject a marker which would tell which part of the stream you need to consume next when merging. This means that at the partitioner side you would have to buffer at least two elements (hold on to the one, that will carry the merge marker, and another one that the marker applies). This approach would apply when the elements in the stream carry no data which can be used to order them. On Wed, Dec 13, 2017 at 12:54 AM, Charlie <charlieevansvimpel...@gmail.com> wrote: > Hello, > > I'm trying to use Akka streams to split a stream into two then merge the > results back into one stream, keeping the order. I cannot keep the ordering > and started to work on a custom GraphStage but not sure if that can work > for me either. > > A simplified version of what I'm trying to achieve is here: > https://gist.github.com/chaz2505/ce5bc170d483f95faa9156b04230372f, I > would like the logs to print 100,2,300,4,500,6 but currently getting > 100,300,500,2,4,6. > I make use of PartitionWith from akka stream contrib. > > For background: My use case is with kafka, I want to split a stream into > two based on the message content, do some async processing on each > sub-stream, then merge back to commit the messages (this is why I need the > ordering). The async processing is represented by separate flows (one of > which I reuse from another project). > > Any help would be greatly appreciated. > > Thanks, > > Charlie > > -- > >>>>>>>>>> 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 https://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam -- >>>>>>>>>> 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: How to measure performance of Actor Message from one actor reaching to another actor
I think the floodPipe benchmark may do what you want, commented out but you could easily uncomment and run locally: https://github.com/akka/akka/blob/master/akka-bench-jmh/src/main/scala/akka/actor/ForkJoinActorBenchmark.scala#L92 -- Johan Akka Team On Fri, Dec 15, 2017 at 11:20 AM, <kpri...@thoughtworks.com> wrote: > Ok, I just looked at TellOnlyBenchmark.scala > <https://github.com/akka/akka/blob/master/akka-bench-jmh/src/main/scala/akka/actor/TellOnlyBenchmark.scala> > closely and it just drops all the message and does not take into > consideration of time taken by message to reach receiving actor. > > > On Friday, December 15, 2017 at 2:54:55 PM UTC+5:30, > kpr...@thoughtworks.com wrote: >> >> Hello, >> >> I want to measure a performance of following use case using jmh. >> >> >> >> >> <https://lh3.googleusercontent.com/-uapjVYoC9Hw/WjOUTKRGMRI/AVo/I31SCM88jicNb5icl04Cc3171k9G-eMegCLcBGAs/s1600/Untitled%2BDiagram%2B%25281%2529.jpg>1. >> Actor A sends Msg A to Actor B >> 2. On receiving Msg A, Actor B sends Same Msg A to Actor C >> >> I just want to measure performance of Msg A from Actor A reaching to >> Actor C (Do not want to consider acknowledgment). >> Is there any way i can measure this performance? >> >> I was going through Akka jmh benchmarks and found this >> TellOnlyBenchmark.scala >> <https://github.com/akka/akka/blob/master/akka-bench-jmh/src/main/scala/akka/actor/TellOnlyBenchmark.scala> >> >> But not sure it just measures the performance of sending messages or it >> considers time required to reach message to Echo Actor? >> >> >> Thanks >> >> >> -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka Stream Tcp Messaging Server POC
What is happening when you run the code and how is that different from what
you expect?
--
Johan
Akka Team
On Mon, Aug 7, 2017 at 2:16 PM, G J <me.gaura...@gmail.com> wrote:
> I'm new Akka stream, I have to build on poc on Akka stream for instant
> messaging. So I need a help to understand what I'm missing here.
>
>
> import java.net.InetSocketAddress;
> import java.util.concurrent.CompletionStage;
>
> import akka.Done;
> import akka.actor.ActorRef;
> import akka.actor.Props;
> import akka.actor.UntypedActor;
> import akka.stream.javadsl.Sink;
> import akka.stream.javadsl.Tcp;
>
> /**
> * Created by gaurav on 28/7/17.
> */
> public class Server {
>public static void main(String[] args) {
> InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1",
> 6000);
> final Sink<Tcp.IncomingConnection, CompletionStage> handler =
> Sink.foreach(conn -> {
> System.out.println("Client connected from: " + conn.remoteAddress());
> conn.handleWith(ActorFlow.actorRef(out ->
> MyWebSocketActor.props(out)), InternalWebSocketHelper.actorMaterializer());
> });
> final CompletionStage bindingFuture =
> Tcp.get(InternalWebSocketHelper.actorSystem())
> .bind(serverAddress.getHostString(),
> serverAddress.getPort()).to(handler).run(InternalWebSocketHelper.actorMaterializer());
>
> bindingFuture.whenComplete((binding, throwable) -> {
> System.out.println("Server started, listening on: " +
> binding.localAddress());
> });
>
> bindingFuture.exceptionally(e -> {
> System.err.println("Server could not bind to " + serverAddress + " :
> " + e.getMessage());
> InternalWebSocketHelper.actorSystem().terminate();
> return null;
> });
>
>}
>
>private static class MyWebSocketActor extends UntypedActor {
>
> private final ActorRef out;
>
> public MyWebSocketActor(ActorRef out) {
> this.out = out;
> }
>
> public static Props props(ActorRef out) {
> return Props.create(MyWebSocketActor.class, () -> new
> MyWebSocketActor(out));
> }
>
> public void onReceive(Object message) throws Exception {
> out.tell(message, ActorRef.noSender());
> }
>}
> }
>
>
>
>
>
>
>
>
> /**
> * Created by gaurav on 1/8/17.
> */
>
> import java.util.function.Function;
>
> import org.reactivestreams.Publisher;
>
> import akka.NotUsed;
> import akka.actor.*;
> import akka.japi.Pair;
> import akka.stream.Materializer;
> import akka.stream.OverflowStrategy;
> import akka.stream.javadsl.*;
>
> public class ActorFlow {
>
>public static <In, Out> Flow<In, Out, NotUsed> actorRef(Function<ActorRef,
> Props> props) {
> return actorRef(props, 1000, OverflowStrategy.dropNew(),
> InternalWebSocketHelper.actorSystem(),
> InternalWebSocketHelper.actorMaterializer());
>}
>
>public static <In, Out> Flow<In, Out, NotUsed> actorRef(Function<ActorRef,
> Props> props, int bufferSize, OverflowStrategy overflowStrategy,
> ActorRefFactory factory, Materializer mat) {
>
> Pair<ActorRef, Publisher> pair = Source. actorRef(bufferSize,
> overflowStrategy)
> .toMat(Sink.asPublisher(AsPublisher.WITHOUT_FANOUT),
> Keep.both()).run(mat);
>
> return Flow.fromSinkAndSource(
>
> Sink.actorRef(factory.actorOf(Props.create(WebSocketFlowActor.class, () ->
> new WebSocketFlowActor(props, pair.first(,
> new Status.Success(new Object())),
> Source.fromPublisher(pair.second()));
>}
>
>private static class WebSocketFlowActor extends UntypedActor {
>
> private final ActorRef flowActor;
>
> public WebSocketFlowActor(Function<ActorRef, Props> props, ActorRef
> ref) {
> flowActor = context().watch(context().actorOf(props.apply(ref),
> "flowActor"));
> }
>
> @Override
> public void onReceive(Object message) throws Throwable {
> if (message instanceof Status.Success) {
> flowActor.tell(PoisonPill.getInstance(), getSelf());
> } else if (message instanceof Terminated) {
> context().stop(getSelf());
> } else {
> flowActor.tell(message, getSelf());
> }
> }
>
> @Override
> public SupervisorStrategy supervisorStrategy() {
> return SupervisorStrategy.stoppingStrategy();
> }
>}
> }
>
>
&g
Re: [akka-user] Re: PersistentActor + akka streams
No current plans about anything specifically related to interacting with persistent actors, couldn't batch + mapAsync(1)(ask-persistent-actor) do what you want? -- Johan Akka Team On Wed, Sep 13, 2017 at 3:50 PM, <techland.co...@gmail.com> wrote: > I have the same question ? > > On Wednesday, September 13, 2017 at 1:32:13 AM UTC+4:30, Yaroslav Klymko > wrote: >> >> Hi guys, >> >> I'm wondering are there any plans to provide integration between akka >> streams and PersistentActor ? >> The feature I'm interested in particular is `FlowOps.batch`, so I could >> batch commands, while events are being persisted, and then persist all >> events produced by stashed commands atomically. This will significantly >> increase throughout. >> >> Best regards. >> > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Splitting a stream to be consumed by dynamic sinks
My thought was that maybe you could achieve what you want by composing it
with existing stages, perhaps zipWithIndex+map index to key/ or use
stateFulMapConcat to decide when to change the key, and then groupBy into a
lazy sink. Can't say I'm sure that will solve your problem, but may be
worth exploring.
--
Johan
Akka Team
On Tue, Oct 31, 2017 at 9:49 PM, Jason Steenstra-Pickens <
thestein...@gmail.com> wrote:
> Hi Johan,
>
> As far as I can tell this only creates a single Sink based on the first
> element. I need a dynamic number of Sinks. In the past I have used a custom
> version of this which creates a new Source for every element (
> OneToOneOnDemandSink <https://stackoverflow.com/a/45874809/367796>).
> Neither of these solve the case above though.
>
> I could create a variation of the OneToOneOnDemandSink called
> ManyToOneOnDemandSink that takes a predicate to determine when to create
> the next Sink however this is still too specialised and we loose the
> composability. For example I would need to create three variations to cover
> the built-in SubFlow API (groupBy, splitAfter, splitWhen) which is doable
> but it is pushing all the logic into the Sink and it doesn't cover any
> custom SubFlows.
>
>
> Cheers,
> Jason
>
> On Tuesday, 31 October 2017 22:33:06 UTC+13, Akka Team wrote:
>>
>> Check if sink.lazySink.lazyInit() doesn't do what you want.
>>
>> --
>> Johan
>> Akka Team
>>
>> On Tue, Oct 31, 2017 at 12:59 AM, Jason Steenstra-Pickens <
>> thest...@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> I seem to be encountering a reoccurring problem when using Akka Streams
>>> and haven't found the right client API to use to solve it.
>>>
>>> The problem usually translates into:
>>>
>>>- I have some possibly infinite Source
>>>- I want to split it into multiple inner Sources based on some
>>>condition such as delimiter, count, or whatever
>>>- I then want to create a Sink for each inner Source dynamically and
>>>run each inner Flow
>>>- I want the backpressure, errors, completion, cancellation and
>>>stuff like that to be shared between the outer Flow and the inner Flow
>>>
>>> There are a few things come close but all seem to be for a slightly
>>> different use case, such as:
>>>
>>>- splitAt / splitWhen
>>>- lazy / lazyInit
>>>- the various hubs
>>>
>>> Here is a concrete example:
>>>
>>>1. Reading a file in 8KB parts
>>>2. Splitting the first 625 parts into a separate stream as a "chunk"
>>>3. Create a HTTP source that has a URL containing the chunk number
>>>4. Send the 625 parts to that source
>>>5. Take the next chunk from step 2
>>>
>>> An attempt using a SubFlow looks like:
>>> val chunkSize = fileResponse.chunkSize
>>> val partsPerChunk = chunkSize / partSize
>>> val counts = Source {
>>> for {
>>> chunk <- 1 to Int.MaxValue
>>> part <- 1 to partsPerChunk
>>> } yield (chunk, part)
>>> }
>>> val source = FileIO.fromPath(filePath, partSize)
>>> .zip(counts)
>>> .splitAfter({ next =>
>>> val (_, (_, part)) = next
>>> part == partsPerChunk
>>> })
>>>
>>> This is quite nice but then there doesn't seem to be a way of getting to
>>> the inner Flow even if I were to create a custom Sink.
>>>
>>> It would be really awesome if SubFlow had a function like:
>>> def to[M](f: Out => Graph[SinkShape[Out], M]): C
>>> (although probably without the materialised value since there would be
>>> multiple).
>>>
>>> Is there something obvious that I am missing?
>>>
>>>
>>> Cheers,
>>> Jason
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c
>>> urrent/additional/faq.html
>>> >>>>>>>>>> Search the archives: https://groups.google.com/grou
>>> p/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,
Re: [akka-user] Splitting a stream to be consumed by dynamic sinks
Check if sink.lazySink.lazyInit() doesn't do what you want.
--
Johan
Akka Team
On Tue, Oct 31, 2017 at 12:59 AM, Jason Steenstra-Pickens <
thestein...@gmail.com> wrote:
> Hi,
>
> I seem to be encountering a reoccurring problem when using Akka Streams
> and haven't found the right client API to use to solve it.
>
> The problem usually translates into:
>
>- I have some possibly infinite Source
>- I want to split it into multiple inner Sources based on some
>condition such as delimiter, count, or whatever
>- I then want to create a Sink for each inner Source dynamically and
>run each inner Flow
>- I want the backpressure, errors, completion, cancellation and stuff
>like that to be shared between the outer Flow and the inner Flow
>
> There are a few things come close but all seem to be for a slightly
> different use case, such as:
>
>- splitAt / splitWhen
>- lazy / lazyInit
>- the various hubs
>
> Here is a concrete example:
>
>1. Reading a file in 8KB parts
>2. Splitting the first 625 parts into a separate stream as a "chunk"
>3. Create a HTTP source that has a URL containing the chunk number
>4. Send the 625 parts to that source
>5. Take the next chunk from step 2
>
> An attempt using a SubFlow looks like:
> val chunkSize = fileResponse.chunkSize
> val partsPerChunk = chunkSize / partSize
> val counts = Source {
> for {
> chunk <- 1 to Int.MaxValue
> part <- 1 to partsPerChunk
> } yield (chunk, part)
> }
> val source = FileIO.fromPath(filePath, partSize)
> .zip(counts)
> .splitAfter({ next =>
> val (_, (_, part)) = next
> part == partsPerChunk
> })
>
> This is quite nice but then there doesn't seem to be a way of getting to
> the inner Flow even if I were to create a custom Sink.
>
> It would be really awesome if SubFlow had a function like:
> def to[M](f: Out => Graph[SinkShape[Out], M]): C
> (although probably without the materialised value since there would be
> multiple).
>
> Is there something obvious that I am missing?
>
>
> Cheers,
> Jason
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Http 2.0 request using SSL cert files in akka http
The HTTP/2 support in Akka HTTP is so far only server side, there is no
client implementation. So you would have to use some other HTTP client to
achieve that for now.
--
Johan
Akka Team
On Wed, Sep 6, 2017 at 2:50 PM, <mahesh.b...@wingify.com> wrote:
> Hi Friends,
>
> I have been using akka http for a year and found it to be useful.
>
> Recently I have a need to make Http 2.0 requests using akka http.
> Consider the below curl request. I want to convert the below curl request
> to an equivalent scala code using akka http.
>
>
> curl -d '{"aps":{"alert":{"title":"Hello from PushCrew","body":"Yay!
> Notification from Safari","action":"Click"},"url-args":["ss","ss"]}}'
> --cert 'h1.pem':'qwerty' -H "apns-topic: web.io.github.batra" --http2
> https://api.push.apple.com:443/3/device/0E705B379AD669DB060DEF1EF60A05
> 0A92F60366F97BBC08B1086241C455FE8A
>
> Notes: Its an Http 2.0 request, it uses pem encoded certificate file in
> the option "--cert"
> Let me know if you need any futher information.
>
>
>
> Thanks
> Mahesh
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: how can write and read async with socket (councurent) ?
The Akka IO API is actor based and a bit more low level, requiring more
boilerplate, the streams one is a higher level API which fits some problems
(and mindsets) very well. Performance will likely not differ much so try
both and use the one you find easiest to understand and achieve what you
want with.
--
Johan
Akka Team
On Wed, Sep 13, 2017 at 4:01 PM, <techland.co...@gmail.com> wrote:
> Someone who can help me ???
>
> On Saturday, September 2, 2017 at 3:53:20 PM UTC+4:30,
> techlan...@gmail.com wrote:
>
>> hi every one
>> I'm using akka 2.5.x how can call Socket async for read and write ?(what
>> is the best solution)
>> i need many connection concurrency for send email this is sample code
>> java ?
>>
>> public boolean sendEmail(String data, String from, String to,String subject)
>> throws IOException {
>> Socket socket;
>> InputStream inn;
>> OutputStream outt;
>> BufferedReader msg;
>> socket = new Socket(mailHost, SMTP_PORT);
>> if (socket == null) {
>> return false;
>> }
>> inn = socket.getInputStream();
>> outt = socket.getOutputStream();
>> in = new BufferedReader(new InputStreamReader(inn));
>> out = new PrintWriter(new OutputStreamWriter(outt), true);
>> if (inn == null || outt == null) {
>> log("Failed to open streams to socket.");
>> return false;
>> }
>> String initialID = in.readLine();
>> out.println("HELO " + localhost.getHostName());
>> String welcome = in.readLine();
>> out.println("MAIL From:<" + from + ">");
>> String senderOK = in.readLine();
>> out.println("RCPT TO:<" + to + ">");
>> String recipientOK = in.readLine();
>> out.println("DATA");
>> out.println("From:Ehsan test<"+from+">");
>> out.println("To:Ehsan uni<"+to+">");
>> out.println("Cc:khodam <"+to+">");
>> out.println("Subject:"+subject);
>> Date d=new Date();
>> out.println("Date:"+d.toString());
>> // this body of email
>> out.println(data);
>> out.println(".");
>> // end of email
>> String acceptedOK = in.readLine();
>> out.println("QUIT");
>> return true;
>> }
>> }
>> how can implement with akka ?
>> how can write and read async ?
>> could help me for this solution
>>
>>
>> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Interrupt actor execution recommended way
That sample looks suspiciously like calling actor internals from another thread (saveStateOnDb from inside the future) which you should never do. See explanation in the docs here: http://doc.akka.io/docs/akka/current/scala/general/jmm.html#futures-and-the-java-memory-model One way to simplify is to use the pipe-pattern and get the future result as a message to your actor, if the future is failed the exception will be wrapped in a akka.actor.Status.Failure and that is sent to the actor and you can deal with it in any way you'd like in the actor receive logic. If the correct reaction to it is to throw or stop actor or send a message to some other actor depends on the specific problem you are solving. If the actor models a request response then the right thing would be to stop and or tell someone about the failure, if it models a remote API where multiple requests flows through it you wouldn't want to stop it. If the failure could have damaged the internal state you would want to restart etc. -- Johan Akka Team On Mon, Aug 21, 2017 at 3:02 PM, Isaias Barroso <isaias.barr...@gmail.com> wrote: > Hi, all. > > Suppose I have an actor and I need to interrupt an execution based on a > Future exception, for example, like at that pseudo code: > https://gist.github.com/isaias/0445b5c32fd1905aba8cd18294672e25 > > Should throw the exception and let supervisor decide what to do? Stop the > Actor? > > Thanks a lot. > > > > > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Ask a lot of Actors inside an Iteration in an Akka-Scheduler and wait for reply of everyone results in a stop of actor-system and webserver
Calling .get on a CompletableFuture blocks the thread until the future is
completed, don't do that. User.find.findPagedList also looks suspiciously
like something blocking.
See this section of the docs for more info:
http://doc.akka.io/docs/akka/current/java/dispatchers.html#problem-blocking-on-default-dispatcher
--
Johan
Akka Team
On Fri, Sep 8, 2017 at 10:00 AM, 'Simon Jacobs' via Akka User List <
akka-user@googlegroups.com> wrote:
> Hy there,
>
> I have an akka scheduler to reindex some data in the search-engine and to
> resize the images of every user.
>
> It will be called directly by
>
> system.scheduler().scheduleOnce(FiniteDuration.create(Long.valueOf(1),
> TimeUnit.SECONDS), reindexActor, "", system.dispatcher(), null);
>
> The (flatted to one method for better readability) actor:
>
>@Override
> public void onReceive(Object msg) throws Exception
> {
> long start = System.currentTimeMillis();
> logger.info("Start reindexing of all User");
>
> if (indexing) {
> logger.info("Already reindexing. Skip");
> return;
> }
>
> try {
> int page = 0;
> PagedList users;
>
> do {
> users = User.find.findPagedList(page, COUNT_OF_ROWS);
> List active = users.getList().stream().filter(g
> -> g.isActive()).collect(Collectors.toList());
>
> List<CompletionStage> stages = new ArrayList
> <CompletionStage>(COUNT_OF_ROWS);
> for (User user : active) {
> ActorRef userActor = system.actorOf(Props.create(
> DependencyInjector.class, injector, UpdateRedisActor.class));
> userActor.tell(new UpdateRedisActor.Index(user.ge
> tId()), null);
>
> if (user.hasProfilePicture()) {
> /**
> * get the image as FilePart (imitate http
> upload)
> */
> File image = new File(user.getProfilePicture
> ().getFileName());
> FilePart filePart = new FilePart<
> Object>("", user.getFirstname(), "image/jpg", image);
>
> String randomFileName = UUID.randomUUID().
> toString();
>
> /**
> * Create new actor
> */
> ActorRef imgActor = system.actorOf(Props.
> create(DependencyInjector.class, injector, ImageActor.class));
>
> /**
> * And ask it
> */
> CompletionStage stage =
> FutureConverters
> .toJava(ask(imgActor, new ImageActor.
> Request(filePart, randomFileName, imageResizeOptionFactory.getValues()),
> ImageBroker.TEN_SECONDS * 20))
> .thenApply((response) -> ((ImageActor.
> Response) response))
> .exceptionally((e) -> {
> Logger.error("Error while
> creating picture [" + randomFileName + "]", e);
> return new ImageActor.Response(500
> );
> });
>
> /**
> * Collect stages
> */
> stages.add(stage.toCompletableFuture().
> thenApplyAsync(response -> {
> user.updateAndSetProfileImage(response.
> fileName, response.filePath);
> user.save();
> Logger.info("User " + user.getId() + "
> saved with new image");
> return response;
> }));
> }
> }
> /**
> * Wait every iteration for the actors
> */
> CompletableFuture.allOf(stages.toArray(new
> CompletableFuture[stages.size()])).toCompletableFuture().get();
> }
> while (users.hasNext());
> }
> catch (Exception e) {
> logger.error("Fatal error while reindexing: ", e);
> }
> finally {
> logger.info("Finish reindexing of all user in " + (System.
Re: [akka-user] Stream within a Stream
Not sure I read you question right, but StreamConverters.fromInputStream
together with flatMapConcat could be what you are after:
def creatingInputSource(s: String): InputStream = ???
val bytesFromInputStreams: Source[ByteString, NotUsed] =
Source(List("something")).flatMapConcat(in =>
StreamConverters.fromInputStream(() => creatingInputSource(in)))
--
Johan
Akka Team
On Sun, Sep 3, 2017 at 1:32 AM, Maatary Okouya <maatarioko...@gmail.com>
wrote:
> Hi,
>
> I am trying to figure out how to handle a situation where in one of your
> stage you need to make a call that return an InputStream, where i would
> deal with that stream as a Source of the stage that comes further down.
>
> e.g.
>
> Source.map(e => Calls that return an InputStream
> ).via(processingFlow).runwith(sink.ignore)
>
> I would like that the element going to Processing flow as those coming
> from the InputStream. This is basically a situation where i am tailing a
> file, reading each line, the line give me the information about a call i
> need to make against a CLI api, when making that call i get the Stdout as
> an InputStream from which to read the result. Result are going to be huge
> most of the time, so i can
> just collect the all thing in memory.
>
> M
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] What's the best way to get an iterator from an akka streams Source?
This sounds like two separate questions to me:
1. How to flatten Iterators of DataRow to a stream of DataRows. One thing
to think of here is whether calling next on the Iterator leads to a
blocking call somewhere which you should isolate. Capturing the state in an
object and using Source.unfoldResource to achieve this is probably the
simplest solution, as this is what it is meant for and therefore already
runs on a separate dispatcher.
2. How the StreamConverters.asJavaStream works. It will back pressure and
does not buffer. You can however introduce a buffer in front of it using
Flow.{batchWeighed, conflate, grouped, groupedWithin} to possibly make it a
bit faster.
--
Johan
Akka Team
On Sat, Sep 2, 2017 at 6:56 AM, Bwmat <bwmat.reloa...@gmail.com> wrote:
> Also, I'm curious how backpressure works in the created stream, especially
> the child Iterator; does it only buffer one element?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Restarting Akka Actor with a new State
Hi Joe,
Watching the child actor before stopping it will give you a Terminated message
which will let you know that the child terminated, so that you can start a
new one. If the id and powerPlantCfg is needed when starting anew you can
put that in the actor state, as a map for keyed by the child ActorRef for
example, so that you can look it up when the Terminated message arrives.
--
Johan
Akka Team
On Mon, Aug 7, 2017 at 9:28 PM, Joe San <codeintheo...@gmail.com> wrote:
> I have a scenario where I have to restart a child Actor from within a
> parent actor. The restart should happen with the following rules:
>
>1. The start should happen only after stop has been completed
>2. Both the stop and start should happen asynchronously
>
> I now have the following scenario:
>
>
> In my parent Actor, I have a Monix Observable which is pushing events as
> below:
>
>
> class ParentActor extends Actor {
> ...
>
> override def preStart(): Unit = {
> super.preStart()
>
> // Observable to stream events regarding PowerPlant's
> val powerPlantEventObservable =
> // For every config.database.refreshInterval in seconds
> Observable.interval(config.database.refreshInterval)
> // We ask the actor for the latest messages
> .map(_ => (dbServiceActor ?
> DBServiceActor.PowerPlantEvents).mapTo[PowerPlantEventsSeq])
> .concatMap(Observable.fromFuture(_))
> .concatMap(Observable.fromIterable(_))
>
> // Subscriber that pipes the messages to this Actor
> cancelable := powerPlantEventObservable.subscribe { update =>
> (self ? update).map(_ => Continue)
> }
> }
> }
>
>
> So what happens above is that, I'm asking another Actor called
> DBServiceActor for a List of events and when these events are available,
> I'm piping it to the ParentActor (self ? update). The receive method of the
> ParentActor looks like this and this is where I want to restart my child
> actor . asynchronously:
>
>
> override def receive: Receive = {
>
> case PowerPlantUpdateEvent(id, powerPlantCfg) =>
> log.info(s"Re-starting PowerPlant actor with id = $id and type
> ${powerPlantCfg.powerPlantType}")
>
> // I want to stop the actor first by finding it from the actor
> system
> // If it exists, do a context.stop on the Actor instance
> // Once it is stopped, I want to start it again by creating a new
> instance of this Actor
> // Once this new Actor instance is created, I want to signal my
> Monix Observer to send me the next event
> }
>
>
> Any suggestions?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Unable to write to file sink greate than 1KB with "extractDataByte" directive using CURL
Hi Mahadev,
this was answered elswhere, so for completeness sake:
1. The sample code seems to create an actor system as a response to a put
request, this is expensive and a bit strange, use the actor system you are
running Akka HTTP on top of instead.
2. no need to set IO dispatcher, the File sink is already running on that
(if needed)
3. newer use Await in routes, this may render your application completely
unresponsive as it keeps the thread for the given duration instead of
turning it back to the threadpool, in routes you can instead use onSuccess
or onComplete, or even do a complete(futureValue)
Something like this should work fine:
path("upload") {
post {
withoutSizeLimit {
extractDataBytes { bytes =>
val path = Files.createTempFile("upload", "dat")
system.log.info(s"Writing request body to $path")
onSuccess(bytes.runWith(FileIO.toPath(path))) {
case IOResult(count, Success(Done)) =>
complete(s"Successfully wrote $count bytes to $path")
case IOResult(_, Failure(ex)) => throw ex
}
}
}
}
}
--
Johan
Akka Team
On Sun, Aug 6, 2017 at 4:51 PM, <mahadev.khap...@gmail.com> wrote:
> Hello Team, With below "extractDataBytes" directive to file sink, I am
> unable to upload a file greater than 1 KB size. Where as below "fileUpload"
> directive to file sink is successful in uploading 700MB file! with same
> akka-server configuration. What is wrong with extractDataBytes code? Any
> specific configuration for "extractDataBytes" is required? Please assist.
>
> *extractDataBytes:*
>
> *curl -k -i -X PUT --header token -d @myfile700MB.tar
> https://abc/uploadContent <https://abc/uploadContent>*
>
> put {
>
> implicit val system = ActorSystem("RepositoryService")
> implicit val materializer = ActorMaterializer()
> implicit val executionContext: ExecutionContextExecutor =
> system.dispatchers.lookup("akka.stream.default-blocking-io-dispatcher")
>
>withoutSizeLimit {
> (extractDataBytes & extractMaterializer) { (bytes,mat) =>
>
> val uploadTmp = File.createTempFile("uploadTMP", ".txt")
> logger.info(" Temp absolute path is : " +
> Paths.get(uploadTmp.getAbsolutePath) + " " +
> Paths.get(uploadTmp.getAbsolutePath).toString)
> try {
>
> import scala.concurrent.duration._
> import akka.stream.scaladsl.FileIO
>
> val action = bytes.runWith(FileIO.toPath(uploadTmp.toPath)
>
> .withAttributes(ActorAttributes.dispatcher("akka.stream.default-blocking-io-dispatcher")))(mat).map
> {
> case ior if ior.wasSuccessful => {
> val end = System.currentTimeMillis
> logger.info(StatusCodes.OK + s"${ior.count}" + "bytes
> written")
> complete(StatusCodes.OK, "Successfully written to FILE. ")
> }
> case ior => complete(StatusCodes.EnhanceYourCalm,
> ior.getError.toString)
> }
> Await.result(action, 30.seconds)
>
> } catch {
> case ex: Throwable => {
> logger.info("Exception in writing to file: " + ex.getMessage)
> complete(StatusCodes.OK, "Exception in writing to file: ")
> }
> }
> }
> }
>
> *fileUpload*
>
> *curl -k -i -X PUT --header token--form "csv=@myfile700MB.tar"
> https://abc/uploadContent <https://abc/uploadContent>*
>
> put {
> val uploadFile = File.createTempFile("uploadFile", ".txt")
> (extractRequestContext & extractMaterializer) { (ctx,mat) =>
> implicit val materializer = ctx.materializer
> implicit val ec = ctx.executionContext
>
> fileUpload("csv") {
> case (metadata, byteSource) =>
>
> try {
> import scala.concurrent.duration._
> import akka.stream.scaladsl.FileIO
>
> val action =
> byteSource.runWith(FileIO.toPath(uploadFile.toPath))(mat).map {
> case ior if ior.wasSuccessful => {
> val end = System.currentTimeMillis
> logger.info(StatusCodes.OK + s"${ior.count}" + "bytes
> written")
> complete(StatusCodes.OK, "Successfully written to FILE. ")
> }
> case ior => complete(StatusCodes.EnhanceYourCalm,
> ior.getError.
Re: [akka-user] [akka-stream] How to provide flow with 'last seen element' backup?
I misread it as wanting the last value to pass through the stream before
completion, but I guess it was, read the last value that passed by in a
stream.
There's no stage that does that out of the box, so providing your own safe
concurrent field of some kind, or firing of the elements to an actor that
keeps the latest value would probably be easiest, creating a stage that
does the same into some materialized value would also be possible (if you
think about the case where a stream is materialized multiple times you
wouldn't want all streams updating the same last-seen-value).
--
Johan
Akka Team
On Mon, Aug 7, 2017 at 8:51 PM, Alexey Shuksto <seig...@gmail.com> wrote:
> Hi, Johan,
>
> Thanks for your reply, but would not `Sink.last` complete it's Future only
> after Source fully completed? I need to check for last passed value
> in-between:
>
> ```
> val bh = Source.unfold(0)(v => Some((v+1, v))).delay(1.minute).runWith(
> BroadcastHub.sink)
> bh.runWith(Sink.ignore)
> ```
>
> So, if `bh` would be materialized after 2.5 minutes, it would not only
> observer value `3` after another 30 seconds, but also value `2` just after
> materialization.
>
> понедельник, 7 августа 2017 г., 15:49:06 UTC+3 пользователь Akka Team
> написал:
>>
>> You can achieve this by using Sink.last which will materialize into a
>> Future[T] that completes with the last element emitted.
>>
>> It does however not complete the promise with the last element on
>> failures, this can be remedied by prepending it with a
>> .recoverWithRetries(0, { case _ => Source.empty }) making sure any
>> upstream failure leads to a completion of the stream.
>>
>> --
>> Johan
>> Akka Team
>>
>> On Sat, Jul 29, 2017 at 8:17 AM, Alexey Shuksto <sei...@gmail.com> wrote:
>>
>>> Hello hAkkers,
>>>
>>> In our project we use some number of flows that follow same building
>>> logic:
>>> ```
>>> val input: Source[T, NotUsed] = ???
>>> val output = input.runWith(BroadcastHub.sink)
>>>
>>> output.runWith(Sink.ignore)
>>> ```
>>>
>>> Sink.ignore here used to prevent BroadcastHub inner buffer overflow and,
>>> also, to dump out old stream elements as they are usually do not matter.
>>>
>>> But in some rare cases we need to provide to clients of our source 'the
>>> last seen element' -- the last element which passed our flow before client
>>> materialization of BroadcastHub' Source.
>>>
>>> Currently it is done via some private volatile var:
>>>
>>> ```
>>> @volatile private var last: Option[T] = None
>>> _output.runForeach(e => last = Some(e))
>>>
>>> val output = _output.prepend(Source.lazily(() => last).collect { case
>>> Some(e} => e})
>>> ```
>>>
>>> I wonder maybe there is better way to do so, without mutable state and
>>> stuff?
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c
>>> urrent/additional/faq.html
>>> >>>>>>>>>> Search the archives: https://groups.google.com/grou
>>> p/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 https://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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: [Akka-typed] Mutable Behavior testing with EffectfulActorContext
As far as I can see the supervisor never sends any messages to the child, so that would be why the child inbox is empty? -- Johan Akka Team On Mon, Aug 7, 2017 at 9:32 AM, Nishant Vishwakarma <nishant...@gmail.com> wrote: > Hi Konrad, > > Thanks for the reply. Meanwhile I am trying to use StubbedActorContext to > achieve the desired results. I am not getting a stackoverflow error and I > am successfully getting number of children and their inboxes. However there > are no messages. One of the child actor sends itself a message to > initialize which in turn should reply the supervisor. I am not getting any > messages on the selfInbox or the respective childInbox. I have the test > available as a gist here > <https://gist.github.com/NISHANTV12/c33cd9e5476186796f044030571530e4> > I will be grateful if you could help. > > Thanks and Regards, > Nishant > > > On Sunday, August 6, 2017 at 8:22:07 PM UTC+5:30, Nishant Vishwakarma > wrote: >> >> Hi, >> >> I am trying to test a mutable behavior which spawns a child actor. I am >> trying to use *EffectfulActorContext* for testing various effects and to >> avail the child inboxes for testing interaction with the spawned child >> actor. >> >> However I am getting a StackOverflow where the child actor is spawned. >> >> An exception or error caused a run to abort. >> java.lang.StackOverflowError >> at akka.typed.internal.ActorContextImpl.spawnAnonymous(ActorContextImpl. >> scala:51) >> at akka.typed.internal.ActorContextImpl.spawnAnonymous$(ActorContextImpl >> .scala:50) >> at akka.typed.testkit.EffectfulActorContext.spawnAnonymous(Effects.scala >> :81) >> >> I do not get such exception if I use *StubbedActorContext*. However in >> doing so I loose the testing features of effects. >> I have created a gist of some sample code resembling my original code >> here >> <https://gist.github.com/NISHANTV12/c33cd9e5476186796f044030571530e4>. >> >> It will be great if someone could help me with it. >> >> Thanks and Regards, >> Nishant >> >> >> >> -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Exception details in Akka Typed supervision
The current Restarter implementation does not allow any more detailed
decisions as we thought it may be enough and becomes nicely declarative.
Note though that you can install multiple restarter for different exception
types, so if all you want is to have different strategies for different
types of exceptions that is doable.
This means that currently if you want to do some more logic with an
exception you would have to catch it.
--
Johan
Akka Team
On Sat, Jul 29, 2017 at 5:53 PM, NISHANT VISHWAKARMA <nishant...@gmail.com>
wrote:
> Hi Team,
>
> I need to get the details of an exception thrown from a child actor in the
> supervisor.
>
> In the untyped world I could have a decider function like below to get the
> exception details
>
> OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
> case *ex: ArithmeticException* => println(s"Child is in trouble
> ${ex.getMessage}") // Or any other task
> Resume
> }
>
> However in the typed world I can only do something like -
>
> Actor
>.supervise(Child.behavior())
>.*onFailure[RuntimeException]*(SupervisorStrategy.restartWithLimit(1,
> FiniteDuration(1, "seconds")).withLoggingEnabled(true)),"child")
>
> This doesn't give a way to get hold of the exception value in case there
> is a need to do any other task apart from logging.
>
> Is there a way I can get the exception value in the supervisor? If it's no
> longer supported, is it to encourage any best practices that I am not aware
> of ?
>
> Thanks and Regards,
> Nishant
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka Typed and Futures - is there a pipe-to alternative?
Just tell would be enough: other ! GetUpdate(ctx.self)
--
Johan
Akka Team
On Fri, Aug 4, 2017 at 4:44 PM, Qux <angizi...@gmail.com> wrote:
> Hi,
> I have something like this:
>
> Actor.immutable[Msg]((ctx, msg) =>
> msg match {
> case DoUpdate =>
> val update: Future[Update] = other ? (GetUpdate(_))
>
> Actor.same
> case Update =>
> //...
> Actor.same
> case _ =>
> //...
> Actor.same
> }
> with
>
> trait Msg
> case object DoUpdate extends Msg
> case object Update extends Msg
>
>
> whats the best way do pipe the Update Message on completion? Or should I use
> a other pattern in this case?
>
>
> Thanks!
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: Akka to throttle service method calls
If it is used in blocking Spring controllers it would not matter, but you should not use it inside of an actor for example. With async, and possibly in general, it would probably be better to detect to many concurrent requests and respond with some form of service-overloaded status to the HTTP request. If done by just queueing up requests (or blocking the incoming thread) and you have a consistent higher inflow of requests than the limited rate those queued requests may fill up your heap and crash the JVM or make the HTTP endpoint unavailable by taking up all the threads. -- Johan Akka Team On Tue, Aug 8, 2017 at 4:37 PM, Damien Gallagher <damien.gallag...@gmail.com > wrote: > Would you recommend not using this then so Richard? > > > On Tuesday, August 8, 2017 at 3:34:37 PM UTC+1, richard wrote: >> >> A quick glance at the code indicates that the RateLimiter is blocking on >> a timer, which is an Akka antipattern. >> >> The documentation also includes this line >> >>> rateLimiter.acquire(); // may wait >>> >>> >> https://github.com/google/guava/blob/master/guava/src/com/ >> google/common/util/concurrent/RateLimiter.java >> >> >> On Tuesday, August 8, 2017 at 8:34:21 AM UTC-5, Damien Gallagher wrote: >>> >>> Thanks Rob >>> >>> On Tuesday, August 8, 2017 at 2:17:09 PM UTC+1, Rob Crawford wrote: >>>> >>>> The code using it has been in production for three years, and hasn't >>>> seen any issues with it. >>>> >>>> On Tuesday, August 8, 2017 at 7:05:59 AM UTC-4, Damien Gallagher wrote: >>>>> >>>>> Hi Rob >>>>> >>>>> Is there any concern with this library being marked as a Beta library? >>>>> >>>>> Thanks >>>>> Damien >>>>> >>>>> On Friday, July 21, 2017 at 9:25:04 PM UTC+1, Rob Crawford wrote: >>>>>> >>>>>> Take a look at Guava's RateLimiter class. I've used it for >>>>>> controlling the rate at which a web service was called, and it worked >>>>>> perfectly. >>>>>> >>>>>> On Monday, July 17, 2017 at 7:15:26 AM UTC-4, Damien Gallagher wrote: >>>>>>> >>>>>>> Hi >>>>>>> >>>>>>> I have a java 8 Spring based project that I want to use akka in. >>>>>>> >>>>>>> I have my RestControllers calling out to service classes. >>>>>>> Some of these service classes call out to third party web services >>>>>>> that have throttling requirements. >>>>>>> >>>>>>> Basically I want to throttle calls to the service classes from my >>>>>>> Rest Controllers using Akka streams. >>>>>>> >>>>>>> 1. Is this use case possible? >>>>>>> 2. If not - would you recommend a different way of doing this >>>>>>> (unfortunately I wouldnt get the time to change all my rest services to >>>>>>> use >>>>>>> the Akka http module) >>>>>>> 3. If possible - can someone provide a quick code sample on how to >>>>>>> implement this? >>>>>>> >>>>>>> Thank you >>>>>>> Damien >>>>>>> >>>>>>> >>>>>>> -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-HTTP: How does backpressure work with Futures and execution contexts?
Each connection is a stream of Request => Response, having a slow Future
inside that stream will backpressure so the same client cannot push another
request through that connection.
The Akka HTTP server by default limits the number of concurrent connections
to 1024 (you can set it using akka.http.server.max-connections) meaning
that at any given time there can be 1024 Request => Response calls running,
however if those calls are all starting a Future on an execution context
with 4 threads there is only 4 threads available to execute, and 1020
requests would queue up to be executed when a thread is available.
If you want to limit the number of requests to a specific number and fail
additional requests rather than queue them up you can do that with a custom
directive like this:
https://gist.github.com/johanandren/b87a9ed63b4c3e95432dc0497fd73fdb
--
Johan
Akka Team
On Tue, Aug 1, 2017 at 1:39 PM, Josh F <jof...@gmail.com> wrote:
> Hi all,
>
> I am just trying out akka-http and haven't been able to find much
> information about how backpressure works when we use futures and run
> operations async.
>
> For example, if I create a REST API with a route that looks like this:
> ```
> val route =
> (path("users") & post & entity(as[NewUser])) { user =>
> complete(createNewUser(user))
> }
> ```
> where the method `createNewUser(user)` spawns an async operation which
> interacts with a database and returns a Future[User].
>
> Say the database slows down and there is a backlog of pending requests,
> does akka-http recognize that the futures are taking a long time to
> complete, and then slow down the number of calls to createNewUser?
>
> Also how does the execution context used to execute the futures affect the
> backpressure? For example, if my database operations take place on an
> execution context which only supports 4 concurrent operations, will
> akka-http ensure that there are at most 4 pending futures/calls to
> createNewUser at any time?
>
> Thanks for any insights on this!
>
> Josh
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: How to deal with Future in a customized Sink?
Using the async is how it should be done, you should never use Await or
call the internal methods of the GraphStageLogic from another thread as you
do in your sample code.
Create the async callback in the onPush method though, not inside the
future onComplete callback. You can let the callback accept a
Try[YourThing] and do the pattern matching on the success/failure there and
then just use pass the invoke method as parameter to Future.onComplete.
--
Johan
Akka Team
On Sun, Aug 6, 2017 at 12:34 AM, Gary Struthers <agilej...@earthlink.net>
wrote:
> GraphStageLogic has getAsyncCallback. You define its handler method which
> returns an AsyncCallback. Then call AsyncCallback.invoke This is needed
> to re-enter the GraphStage whenever you return a Future in a custom stage.
>
>
> scalaRSF.onComplete {
>
> case Success(rs) => {
>
> val successCallback = getAsyncCallback{
>
> (_: Unit) => {
>
> if(mustFinish) completeStage() else pull(in)
>
> waitForHandler = false
>
> }
>
> }
>
> successCallback.invoke(rs)
>
> }
>
> Gary Struthers
>
> https://garyaiki.github.io/dendrites/
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] [akka-stream] How to provide flow with 'last seen element' backup?
You can achieve this by using Sink.last which will materialize into a
Future[T] that completes with the last element emitted.
It does however not complete the promise with the last element on failures,
this can be remedied by prepending it with a .recoverWithRetries(0, { case
_ => Source.empty }) making sure any upstream failure leads to a completion
of the stream.
--
Johan
Akka Team
On Sat, Jul 29, 2017 at 8:17 AM, Alexey Shuksto <seig...@gmail.com> wrote:
> Hello hAkkers,
>
> In our project we use some number of flows that follow same building logic:
> ```
> val input: Source[T, NotUsed] = ???
> val output = input.runWith(BroadcastHub.sink)
>
> output.runWith(Sink.ignore)
> ```
>
> Sink.ignore here used to prevent BroadcastHub inner buffer overflow and,
> also, to dump out old stream elements as they are usually do not matter.
>
> But in some rare cases we need to provide to clients of our source 'the
> last seen element' -- the last element which passed our flow before client
> materialization of BroadcastHub' Source.
>
> Currently it is done via some private volatile var:
>
> ```
> @volatile private var last: Option[T] = None
> _output.runForeach(e => last = Some(e))
>
> val output = _output.prepend(Source.lazily(() => last).collect { case
> Some(e} => e})
> ```
>
> I wonder maybe there is better way to do so, without mutable state and
> stuff?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Does Akka directive "fileUpload" copies the entire input entity file into memory?
Hi,
You are putting the entire file into memory by running
byteSource.runFold(ByteString.empty) { case (acc, i) => acc ++ i
}.map(s => s.utf8String)
This means the body is first completely read into memory as a ByteString
and then transformed into a String.
To stream the byte chunks directly to a file you would do something like
this:
val futureWriteResult: Future[IOResult] =
byteSource.runWith(FileIO.toPath(uploadFile))
--
Johan
Akka Team
On Fri, Aug 4, 2017 at 7:01 AM, <mahadev.khap...@gmail.com> wrote:
> Does Akka directive "fileUpload" copies the entire input entity file into
> memory? Is there a way to do multipart upload of large file as 1 GB without
> consuming 1 GB of memory?
>
> My below code to upload 300MB is consuming 300MB of memory. Is this
> expected behavior with "fileUpload" directive?
>
>
> curl -k -i -X PUT --header --form "csv=@myfile300MB.txt"
> https://abc/uploadContent
>
> val uploadFile = File.createTempFile("uploadFile", ".txt")
> extractRequestContext { ctx =>
> implicit val materializer = ctx.materializer
> implicit val ec = ctx.executionContext
> fileUpload("csv") {
> case (metadata, byteSource) =>
> val sumF = byteSource.runFold(ByteString.empty) { case (acc, i) =>
> acc ++ i }.map(s => s.utf8String)
>
> onSuccess(sumF) { sum =>
> Files.write(Paths.get(uploadFile.getAbsolutePath),
> sum.toString.getBytes)
> logger.info(StatusCodes.OK + "Successfully completed fileUpload ")
> complete(s"Successfully completed fileUpload") }}
> }
>
> Thanks *johanandren <https://github.com/johanandren>*for your below response:
>
> Hi @mahadev-khapali <https://github.com/mahadev-khapali>, we use the
> issue tracker for bugs and feature requests, please use the mailing list
> https://groups.google.com/forum/#!forum/akka-user or the gitter chat
> https://gitter.im/akka/akka for questions like this, thanks!
>
> (And yes, if you use fold like that you are collecting the entire upload
> into memory, you should instead stream it directly to a file sink if you
> want to avoid that)
>
>
> Can you somebody suggest how do I directly stream to a file sink so that I
> dont upload into memroy?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] No response from remote for outbound association
Akka remoting is peer to peer, meaning that either side could initiate the connection, looks like there are problems doing that. I'd look into and make sure it is possible to connect between the machines on the specified ports. You could verify outside of Akka with telnet or netcat or some tool like that. -- Johan Akka Team On Thu, Jul 20, 2017 at 9:21 AM, Olivier Pharabot < olivier.phara...@gmail.com> wrote: > All, > > I am facing a very confusing issue between two systems, and i came to a > conclusion that may be someone could confirm (or not). > I am running Akka 2.5.3 in Java, the context is a CI build running > integration tests. > I have 2 systems: A (=QA system) and B (=CI system). B knows A through dns > and B uses IP address for identification. > The flow is : A is started, then B tries to resolve an actor through > ActorSelection to A. When A and B are running on the same host, all runs > fine. > The issue is when A and B are on different hosts. A is behind a firewall > with only the relevant port opened. I am not 100% sure of the firewall > settings on host B. > On the first run, all is fine and ActorSelection returns the ActorRef, but > on the second run ActorSelection fails in timeout and I get the below > exception on host A, while no other exception is visible in host B. > > What I am thinking is that when B is restarted, A considers it as a > "reconnection" differently then during first run "connection" and might > follow different path. Even if the System internal Id is different. > In this "reconnection" use case, it looks like there might be a connection > attempt A -> B while in the first "connection" only B -> A happens. > This would explain why when running on the same host all runs fine. If my > thinking is correct, then i would have to investigate the firewall rules on > host B, which is beyond my control and i would need this confirmation > before going further. > > [DEBUG] [2017-07-19 17:22:16,317] [17:22:16.317UTC] [SystemA] > [akka.tcp://SystemA@hostA:2550/system/endpointManager/ > reliableEndpointWriter-akka.tcp%3A%2F%2FSystemB%40172.17.0.3%3A2557-3/endpointWriter] > [SystemA-akka.remote.default-remote-dispatcher-90] [] > akka.remote.EndpointWriter - AssociationError [akka.tcp://SystemA@hostA:2550] > -> [akka.tcp://SystemB@172.17.0.3:2557]: Error [Association failed with > [akka.tcp://SystemB@172.17.0.3:2557]] [ > akka.remote.EndpointAssociationException: Association failed with > [akka.tcp://Bl@172.17.0.3:2557] > Caused by: java.util.concurrent.TimeoutException: No response from remote > for outbound association. Associate timed out after [15000 ms]. > at akka.remote.transport.ProtocolStateActor$$anonfun$2.applyOrElse( > AkkaProtocolTransport.scala:366) > at akka.remote.transport.ProtocolStateActor$$anonfun$2.applyOrElse( > AkkaProtocolTransport.scala:340) > at scala.runtime.AbstractPartialFunction.apply( > AbstractPartialFunction.scala:34) > at akka.actor.FSM.processEvent(FSM.scala:663) > at akka.actor.FSM.processEvent$(FSM.scala:660) > at akka.remote.transport.ProtocolStateActor.processEvent( > AkkaProtocolTransport.scala:285) > at akka.actor.FSM.akka$actor$FSM$$processMsg(FSM.scala:657) > at akka.actor.FSM$$anonfun$receive$1.applyOrElse(FSM.scala:629) > at akka.actor.Actor.aroundReceive(Actor.scala:513) > at akka.actor.Actor.aroundReceive$(Actor.scala:511) > at akka.remote.transport.ProtocolStateActor.aroundReceive( > AkkaProtocolTransport.scala:285) > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:527) > at akka.actor.ActorCell.invoke(ActorCell.scala:496) > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257) > at akka.dispatch.Mailbox.run(Mailbox.scala:224) > at akka.dispatch.Mailbox.exec(Mailbox.scala:234) > at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) > at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue. > runTask(ForkJoinPool.java:1339) > at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) > at akka.dispatch.forkjoin.ForkJoinWorkerThread.run( > ForkJoinWorkerThread.java:107) > ] > [INFO ] [2017-07-19 17:22:16,318] [17:22:16.317UTC] [SystemA] > [akka.tcp://SystemA@hostA:2550/system/transports/akkaprotocolmanager.tcp0/ > akkaProtocol-tcp%3A%2F%2FSystemB%40172.17.0.3%3A2557-4] > [SystemA-akka.remote.default-remote-dispatcher-75] [] > akka.remote.transport.ProtocolStateActor - No response from remote for > outbound association. Associate timed out after [15000 ms]. > [WARN ] [2017-07-19 17:22:16,318] [17:22:16.318UTC] [SystemA] > [akka.tcp://SystemA@hostA:2550/system/endpointManager/ > reliableEndpointWriter-akka.tcp%3A%2F%2FSystemB%40172.17.0.3%3A2557-3] > [SystemA-akka.remote.default-remote-dispatcher-90] [] > akka.re
Re: [akka-user] Cannot stop actor running akka stream
Hi,
There is nowhere in that shared snippet where you actually use the kill
switch to stop the stream?
A reasonable place to use the kill switch would be the postStop hook of the
actor.
This 3 part series of articles by Colin Breck may help:
http://blog.colinbreck.com/integrating-akka-streams-and-akka-actors-part-i/
--
Johan
Akka Team
On Thu, Jul 6, 2017 at 7:43 PM, Lap Ming Lee <lapming@gmail.com> wrote:
> I cannot stop an actor running akka stream even with killswitch. How do I
> properly stop an actor running an akka stream?
>
> private def startStream(maybeOffset: Option[ProcessedEventOffset]): Unit = {
> implicit val materializer = ActorMaterializer()
>
> val publisher = SomePublisher
>
> val (killSwitch, done) = Source
> .fromPublisher(publisher)
> .viaMat(KillSwitches.single)(Keep.right)
> .toMat(Sink.ignore)(Keep.both)
> .run()
>
> done.onComplete {
> case Success(_) =>
> context.stop(self)
>
> case Failure(e) => // receive an error from the stream
> log.error(e, "Error while processing stream")
> context.stop(self)
> }
>
> maybeSwitch = Some(killSwitch)
> }
>
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Learn Akka From Samples & Examples
The getting started guide is a good place to get started ;) http://doc.akka.io/docs/akka/current/java/guide/introduction.html -- Johan Akka Team On Thu, Jul 13, 2017 at 7:02 PM, Bharath Kandaswami <bharath.so...@gmail.com > wrote: > Hi Folks, > > I am a newbie to Akka, just wanted to explore and playaround with Akka a > little bit to see if I can use Akka in place of traditional Java > development tools, can someone show me the right direction to reach there. > > Any help is deeply appreciated. > > Kind Regards, > KK>Bharath. > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Need help understanding the warning on my console
At the end of each test or test suite, you stop the actor system that was
running, usually using TestKit.shutdownActorSystem, this method takes a
timeout, if it hasn't manage to terminate the actor system before this
timeout the call fail with the error message you describe.
All actors are stopped when before the ActorSystem shutdown completes. If
there is an actor blocking or simply not completing the processing of a
message (let's say it does a while(true) { ... } for example, this will
hinder the shutdown. You can figure out exactly what in your code is
blocking/hindering actor shutdown by taking a thread dump, you should be
able to see your own logic in one of the stack traces hinting you towards
what is wrong.
--
Johan
Akka Team
On Thu, Jul 13, 2017 at 4:59 PM, usermda <shruti.palshi...@gmail.com> wrote:
> Hello
>
> I started getting this warning with a test failure (intermittent) that I
> have no clue what it means. Any help on what the warning indicates will be
> helpful. Here is the warning
> I need some help understanding this warning
>
> [WARN] [07/13/2017 09:35:53.465] [main] [ActorSystem(default)] Failed
> to stop [default] within [10 seconds]
> -> / LocalActorRefProvider$$anon$1 class
> akka.actor.LocalActorRefProvider$Guardian
> status=0 1 children
>⌊-> system LocalActorRef class
> akka.actor.LocalActorRefProvider$SystemGuardian
> status=2 Terminating(UserRequest)
>|toDie: Actor[akka://default/system/
> deadLetterListener#-47810243]
>| Actor[akka://default/system/testActor1#-1858382714]
>| Actor[akka://default/system/testActor2#-1013315176]
>| Actor[akka://default/system/testActor3#-639523170]
>⌊-> deadLetterListener RepointableActorRef null status=1
> terminated
>⌊-> testActor1 RepointableActorRef null status=1 terminated
>⌊-> testActor2 RepointableActorRef null status=1 terminated
>⌊-> testActor3 RepointableActorRef class akka.testkit.TestActor
> status=0 Terminating(UserRequest)
>| |toDie: TestActor[akka://default/
> system/testActor3/1ae159bf-59a1-4899-884d-105705d1470b]
>| ⌊-> 1ae159bf-59a1-4899-884d-105705d1470b TestActorRef
> class myParentActor status=0 1 children
>| ⌊-> $a LocalActorRef class myFirstChild status=2 no
> children
>⌊-> testActor4 RepointableActorRef class akka.testkit.TestActor
> status=0 no children
>⌊-> testActor5 RepointableActorRef class akka.testkit.TestActor
> status=0 1 children
>| ⌊-> dc2a094f-a12e-4221-8875-e400065b522f TestActorRef
> class MyParentActor status=0 no children
>⌊-> testActor6 RepointableActorRef class akka.testkit.TestActor
> status=0 no children
>⌊-> testActor7 RepointableActorRef class akka.testkit.TestActor
> status=0 1 children
>| ⌊-> 643d6fc5-bcd6-4c27-9da6-3f04c59bccfd TestActorRef
> class MyParentActor status=0 no children
>⌊-> testActor8 RepointableActorRef class akka.testkit.TestActor
> status=0 no children
>
>
> I have a test that is expecting a certain message after the myParentActor
> has completed a task that is timing out. I am trying to figure out if they
> are related
>
>
> I am not sure what more info to provide
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: formField recent change
Likely because of the fix for this bug:
https://github.com/akka/akka-http/issues/1134
Before 10.0.8 Akka HTTP would just blindly guess the charset UTF-8 even if
the request did not specify a charset, while now it has a specific type
representing a missing charset instead to give user server logic a chance
decide what to do.
--
Johan
Akka Team
On Tue, Jun 27, 2017 at 6:33 PM, Christophe Pache <chpa...@gmail.com> wrote:
> The difference seams to come from that (using request logging):
> HttpEntity.Strict(application/x-www-form-urlencoded; charset=UTF-8,foo=bar
> )
> becomes
> HttpEntity.Strict(application/x-www-form-urlencoded,ByteString(102, 111,
> 111, 61, 98, 97, 114))
>
> so is fixed using the charset.
>
> Le lundi 26 juin 2017 07:53:29 UTC+2, Christophe Pache a écrit :
>>
>> Hello everyone!
>>
>> I'm using forms in a akka-http app and I have an issue with
>> `formFieldMap` directive: it start to fail after a akka update (akka-http
>> 10.0.6 -> 10.0.8 and akka 2.5.1 -> 2.5.3).
>> I feel like the FormData was not required before this change (as some of
>> my test use also akka-http, was fine to just add an empty FormData and it
>> solved the issue).
>>
>>
>> My main question is: if I don't provide a form data in my request, would
>> a directive formFieldMap be taken or not? What about the parameterMap which
>> seams taken even if empty?
>>
>> I'm stuck currently trying to make test using basic HttpURLConnection work
>> with the update (was working before). Have someone already tested
>> formFieldMap/Seq without akka-http test support? Here is part of code I'm
>> using to test:
>>
>> val conn: HttpURLConnection = createHttpConnection(urlEndpoint)
>> conn.setRequestMethod("GET")
>> conn.setRequestProperty("Content-Type", "application/x-www-form-urlencoded")
>> val postDataBytes: Array[Byte] = "foo=bar".getBytes(*"UTF-8"*)
>> conn.setRequestProperty("Content-Length",
>> String.valueOf(postDataBytes.length))
>> conn.getOutputStream.write(postDataBytes)
>>
>> val response: Int = conn.getResponseCode
>> val inputStream: InputStream = conn.getInputStream
>> (readUtf8InputStream(inputStream), conn.getHeaderFields)
>>
>>
>>
>>
>> Many thanks for any hint and have a nice day!
>> Christophe
>>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] consuming PersistentFSM.StateChangeEvent via PersistentQuery
Just in case someone has the same problem and comes here, Lantz has since reported a ticket that you can follow about this: https://github.com/akka/akka/issues/23072 -- Johan Akka Team On Wed, May 31, 2017 at 12:08 PM, lantz moore <lantz.mo...@gmail.com> wrote: > i'd like to consume the event stream of a PersistentFSM. in particular, > i'd like to be able to index/monitor the state changes. however, > StateChangeEvent is private to the persistence package (which makes sense, > no need to leak the implementation); so you can't really match on it from > the event stream. > > would it be reasonable to ask for an extractor that would at least allow > us to match on the name of a state change? > > thanks. > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] supervisorStrategy for
Hi,
#1 sender is deadLetters
You are closing over the method this.sender() from the actor where you
defined the strategy, so you are not calling sender() in the RouterActor
which is what would return the routee actor where the exception happened.
#2. All routees restarted
http://doc.akka.io/docs/akka/current/scala/routing.html#supervision
> the router’s supervisor will treat the error as an error with the router
> itself. Therefore a directive to stop or restart will cause the router
> *itself* to stop or restart. The router, in turn, will cause its children
> to stop and restart.
--
Johan
Akka Team
On Sat, May 6, 2017 at 4:12 PM, Urvi <upatel...@gmail.com> wrote:
> Hello,
>
> I have a MasterActor where a router is defined and that is where I defined
> the following supervisionStrategy:
>
> override val supervisorStrategy = OneForOneStrategy() {
>case _: FileNotFoundException =>
>println("** Failed actor in FileNotFound: " + sender)
>Restart
> }
>
> I then passed this supervisionStrategy to the router:
>
> val router = context.actorOf(RemoteRouterConfig(RoundRobinPool(3,
> supervisorStrategy=supervisorStrategy),
> addresses).props(Props(classOf[MapActor],
> reduceActor)), "router")
>
> I think this part is right, based on the Akka documentation.
>
> I have 2 issues so far:
>
> Issue #1:
>
> I noticed that sender is deadLetters in the print statement. Why is
> that? I expected sender to be one of the routees.
>
> Issue #2:
>
> I put a print statement in the postRestart method and I see that only one
> routee is restarted when a FileNotFound exception is encountered and the
> processing of the continues. That seems like the correct behavior
>
> If I change the supervisorStrategy to Stop (shown below), I see that ALL
> actors stop and I do not ANY processing taking place, at all. I expected
> to see processing occur by all actors. When the failed actor hits the
> FileNotFound exception, I thought that ONLY that actor should be stopped
> and the others should continue processing.
>
> override val supervisorStrategy = OneForOneStrategy() {
>case _: FileNotFoundException =>
>println("** Failed actor in FileNotFound: " + sender)
>Stop
> }
>
> Given the above observations, I am afraid I have not set this up
> properly. Thanks in advance
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Alpakka docs unavailable
Thanks for letting us know, looking into it. -- Johan Akka Team On Fri, Jun 9, 2017 at 10:37 AM, 'Michal Borowiecki' via Akka User List < akka-user@googlegroups.com> wrote: > http://developer.lightbend.com/docs/alpakka/current/ started giving a 404. > > I looked if it moved somewhere but couldn't find it. > > http://doc.akka.io/docs/akka/2.5.2/java/common/other-modules.html > > https://github.com/akka/alpakka > both point to the broken link. > > Has it moved or is it a temporary outage? > > Cheers, > Michał > -- > <http://www.openbet.com/> Michal Borowiecki > Senior Software Engineer L4 > T: +44 208 742 1600 > > > +44 203 249 8448 > > > > E: michal.borowie...@openbet.com > W: www.openbet.com > OpenBet Ltd > > Chiswick Park Building 9 > > 566 Chiswick High Rd > > London > > W4 5XT > > UK > <https://www.openbet.com/email_promo> > This message is confidential and intended only for the addressee. If you > have received this message in error, please immediately notify the > postmas...@openbet.com and delete it from your system as well as any > copies. The content of e-mails as well as traffic data may be monitored by > OpenBet for employment and security purposes. To protect the environment > please do not print this e-mail unless necessary. OpenBet Ltd. Registered > Office: Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT, > United Kingdom. A company registered in England and Wales. Registered no. > 3134634. VAT no. GB927523612 > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: How should I config the use-dispatcher-for-io
When re-configuring the netty execution context you must make sure that the used execution context has enough threads for netty, in newer versions of Akka you should also see this warning hinting about that: WARNING: Failed to get all worker threads ready within 10 second(s). Make sure to specify the executor which has more threads than the requested workerCount. If unsure, use Executors.newCachedThreadPool(). -- Johan Akka Team On Fri, Jun 2, 2017 at 5:15 AM, Yutao <szqs...@gmail.com> wrote: > > Here is the Exception: > > Exception in thread "main" java.util.concurrent.TimeoutException: Futures > timed out after [1 milliseconds] > [ERROR] [06/02/2017 11:11:55.295] [main] [Remoting] Remoting error: > [Startup timed out. This is usually related to actor system host setting or > host name resolution misconfiguration.] [ > akka.remote.RemoteTransportException: Startup timed out. This is usually > related to actor system host setting or host name resolution > misconfiguration. > at akka.remote.Remoting.akka$remote$Remoting$$notifyError( > Remoting.scala:135) > at akka.remote.Remoting.start(Remoting.scala:197) > at akka.remote.RemoteActorRefProvider.init(RemoteActorRefProvider.scala: > 184) > at akka.cluster.ClusterActorRefProvider.init( > ClusterActorRefProvider.scala:58) > at akka.actor.ActorSystemImpl.liftedTree2$1(ActorSystem.scala:621) > at akka.actor.ActorSystemImpl._start$lzycompute(ActorSystem.scala:618) > at akka.actor.ActorSystemImpl._start(ActorSystem.scala:618) > at akka.actor.ActorSystemImpl.start(ActorSystem.scala:635) > at akka.actor.ActorSystem$.apply(ActorSystem.scala:142) > at akka.actor.ActorSystem$.apply(ActorSystem.scala:119) > at akka.actor.ActorSystem$.create(ActorSystem.scala:67) > at akka.actor.ActorSystem.create(ActorSystem.scala) > at com.huawei.akkacluster.splitbrain.plus.StartCluster. > startup(StartCluster.java:52) > at com.huawei.akkacluster.splitbrain.plus.StartCluster. > main(StartCluster.java:31) > Caused by: java.util.concurrent.TimeoutException: Futures timed out after > [1 milliseconds] > at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219) > at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223) > at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107) > at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn( > BlockContext.scala:53) > at scala.concurrent.Await$.result(package.scala:107) > at akka.remote.Remoting.start(Remoting.scala:179) > ... 12 more > ] > at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219) > at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223) > at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107) > at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn( > BlockContext.scala:53) > at scala.concurrent.Await$.result(package.scala:107) > at akka.remote.Remoting.start(Remoting.scala:179) > at akka.remote.RemoteActorRefProvider.init(RemoteActorRefProvider.scala: > 184) > at akka.cluster.ClusterActorRefProvider.init( > ClusterActorRefProvider.scala:58) > at akka.actor.ActorSystemImpl.liftedTree2$1(ActorSystem.scala:621) > at akka.actor.ActorSystemImpl._start$lzycompute(ActorSystem.scala:618) > at akka.actor.ActorSystemImpl._start(ActorSystem.scala:618) > at akka.actor.ActorSystemImpl.start(ActorSystem.scala:635) > at akka.actor.ActorSystem$.apply(ActorSystem.scala:142) > at akka.actor.ActorSystem$.apply(ActorSystem.scala:119) > at akka.actor.ActorSystem$.create(ActorSystem.scala:67) > at akka.actor.ActorSystem.create(ActorSystem.scala) > at com.huawei.akkacluster.splitbrain.plus.StartCluster. > startup(StartCluster.java:52) > at com.huawei.akkacluster.splitbrain.plus.StartCluster. > main(StartCluster.java:31) > > 在 2017年6月2日星期五 UTC+8上午10:28:31，Yutao写道： > >> I config use-dispatcher-for-io to "akka.remote.default-remote-dispatcher", >> but Actorsystem start failed and throws Futures timeout Exception. >> >> >> <https://lh3.googleusercontent.com/-T3fHJ07z-DI/WTDMkE6jw2I/ANs/EMAv9vdfdYY_mxe3sdyfwKjgsACCfCqiwCLcB/s1600/2017-06-02_102428.png> >> >> -- > >>>>>>>>>> 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
Re: [akka-user] Akka-http connection pool rotating connections (time to live)
There is no max-time-to-live for a connection, only akka.http.host-connection-pool.idle-timeout but that requires there to be no requests going through the connection for a timeout before it is terminated. I'd expect that when you deploy the new version, the old version get undeployed and those connections are killed by the server so that new requests through the pool triggers new requests which should then end up in the new service? -- Johan Akka Team On Tue, May 30, 2017 at 9:21 PM, Jorge Fuentes <jfuentes...@gmail.com> wrote: > > Hi here, > > I have a service running on production, which calls a dependency service > using : > > Http().newHostConnectionPoolHttps[Promise[HttpResponse]] > > Each time a new stack of the dependency service is deployed it use same > domain, my service DNS cache is updated but problem is connections inside > the pool lives on forever and only resolves DNS once at initiation time. > Thus it keeps calling the old dependency stack. > > I am searching for away to set max connection's time to live or some way > to recreate the connections > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] beginner question on akka streams flow design.
One way would be mapAsync(1)(elem => ask-actor-and-combine-elem-with-token). If asking for every element is too expensive you could make a custom graph stage which asks the actor on materialisation to get the first token, but then also registers for updates from the actor, that can be done using the stage actor ref (see docs here: http://doc.akka.io/docs/akka/current/scala/stream/stream-customize.html#integration-with-actors ) -- Johan Akka Team On Thu, Jun 1, 2017 at 7:21 PM, Mohnish Kodnani <mohnish.kodn...@gmail.com> wrote: > Hi , > I have the following use case that I want to model using akka streams and > I am new to akka streams so would like to know how/what is the best way to > model it. > > I have a kafka consumer which reads from one topic massages the message > and sends it to a RESTful external service. > This external restful service takes a token that needs to be sent with > every message in the header. This token expires every roughly 24 hours. > > I was thinking of creating an FSM actor that will act as a Token Manager. > It will be in Active state where it can send the current token on receiving > say a GetToken message. Then every 24 hours it receives a scheduled event > for FetchToken, then it will go into Fetching mode, where it will retrieve > a new token from another service and in that state it should keep queuing > the requests for GetToken. > > Now, how do i hook this actor (TokenManager) into my stream so that it can > provide a token for every message from kafka but block the stream until it > gets a new token when it expires. > Or if there is another solution to what I am thinking instead of creating > a separate actor. > > Thanks > Mohnish. > > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: ActorSystem Uncaught fatal error shutting down ActorSystem !
I think this is because of https://github.com/akka/akka/issues/22324 It was marked as in progress but it seems there never was a PR submitted to fix that. -- Johan Akka Team On Wed, Jun 7, 2017 at 12:16 PM, <ciet...@gmail.com> wrote: > yes, *situation 1 *, the log has tell me the reason cause the error , > and i know how to solve *situation 1 * . > but *situation 2,3,4 *, din't tell me the reason ( the stack ), id dont > hnow how solve this problem. > > 在 2017年6月7日星期三 UTC+8下午5:00:07，Johannes Rudolph写道： >> >> >> The rest of the stack here will tell you where the problem comes from: >> >> On Wed, Jun 7, 2017 at 10:40 AM, <cie...@gmail.com> wrote: >> >>> * situation 1 : ( has tell me the stack, and tell me *Shutdown finished * >>> )* >>> INFO | jvm 1| 2017/04/07 15:49:52 | java.lang.NoClassDefFoundError: >>> Lws/protos/EnumsProtos$HardTypeEnum; >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.lang.Class.getDeclaredFields0(Native Method) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.lang.Class.privateGetDeclaredFields(Class.java:2583) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.lang.Class.getDeclaredField(Class.java:2068) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.io.ObjectStreamClass.getDeclaredSUID(ObjectStreamClass.java:1703) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.io.ObjectStreamClass.access$700(ObjectStreamClass.java:72) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.io.ObjectStreamClass$2.run(ObjectStreamClass.java:484) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.io.ObjectStreamClass$2.run(ObjectStreamClass.java:472) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.security.AccessController.doPrivileged(Native Method) >>> INFO | jvm 1| 2017/04/07 15:49:52 | at >>> java.io.ObjectStreamClass.(ObjectStreamClass.java:472) >>> .. >>> >>> >> It might be related to actor messages received over remoting that cannot >> be deserialized because classpaths differ between the two nodes. >> >> Johannes >> >> -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Questions about Cluster Aware Router
The error message says that you have not configured serialization for the message sample.distributeddata.ReplicatedCache.PutInCache which something tries to send between nodes. -- Johan Akka Team On Tue, May 30, 2017 at 10:37 PM, Mainak Ghosh <famousmai...@gmail.com> wrote: > Hello, > > I am recently playing around with cluster aware routers and distributed > data and I have a few questions. I extended the multi jvm test code > ReplicatedCacheSpec to create a cluster aware router which creates a pool > of ReplicatedCache worker routees. Here is the code: > https://gist.github.com/mghosh4/55f96a7402960923c528ea69a103c5af > > When I try running this code: I get the following exception: > > Failed to serialize remote message [class akka.actor.ActorSelectionMessage] > using serializer [class akka.remote.serialization.MessageContainerSerializer]. > Transient association error (association remains live) > [JVM-1] at akka.remote.MessageSerializer$ > .serialize(MessageSerializer.scala:62) > [JVM-1] at akka.remote.EndpointWriter.$anonfun$serializeMessage$1( > Endpoint.scala:895) > [JVM-1] at scala.util.DynamicVariable.withValue(DynamicVariable. > scala:58) > [JVM-1] at akka.remote.EndpointWriter.serializeMessage(Endpoint. > scala:895) > [JVM-1] at akka.remote.EndpointWriter. > writeSend(Endpoint.scala:786) > [JVM-1] at akka.remote.EndpointWriter$$anonfun$4.applyOrElse( > Endpoint.scala:761) > [JVM-1] at akka.actor.Actor.aroundReceive(Actor.scala:513) > [JVM-1] at akka.actor.Actor.aroundReceive$(Actor.scala:511) > [JVM-1] at akka.remote.EndpointActor.aroundReceive(Endpoint.scala: > 452) > [JVM-1] at akka.actor.ActorCell.receiveMessage(ActorCell. > scala:519) > [JVM-1] at akka.actor.ActorCell.invoke(ActorCell.scala:488) > [JVM-1] at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257) > [JVM-1] at akka.dispatch.Mailbox.run(Mailbox.scala:224) > [JVM-1] at akka.dispatch.Mailbox.exec(Mailbox.scala:234) > [JVM-1] at akka.dispatch.forkjoin.ForkJoinTask.doExec( > ForkJoinTask.java:260) > [JVM-1] at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue. > runTask(ForkJoinPool.java:1339) > [JVM-1] at akka.dispatch.forkjoin.ForkJoinPool.runWorker( > ForkJoinPool.java:1979) > [JVM-1] at akka.dispatch.forkjoin.ForkJoinWorkerThread.run( > ForkJoinWorkerThread.java:107) > [JVM-1] Caused by: java.io.NotSerializableException: No configured > serialization-bindings for class [sample.distributeddata. > ReplicatedCache$PutInCache] > [JVM-1] at akka.serialization.Serialization.serializerFor( > Serialization.scala:235) > [JVM-1] at akka.serialization.Serialization.findSerializerFor( > Serialization.scala:211) > [JVM-1] at akka.remote.serialization.MessageContainerSerializer. > serializeSelection(MessageContainerSerializer.scala:35) > > When I change line 83 in the gist from > > context.actorOf(FromConfig.props(ReplicatedCache.props()), name = " > replicatedCache") > > to > > context.actorOf(ReplicatedCache.props), name = "replicatedCache") > > this error goes away. Is there something that I do not understand? > > Finally, I tried adding a resizer block in the configuration. I saw an > exception saying something like resizer and cluster block cannot work > together. Is that a current limitation? > > Thanks, > Mainak > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Detecting connection failures - documentation appears to be incorrect
Seems like a regression to me. I have opened an issue to track it here:
https://github.com/akka/akka-http/issues/1179
--
Johan
Akka Team
On Tue, May 2, 2017 at 8:33 PM, richard <harold.richard.sea...@gmail.com>
wrote:
> This document
> <http://doc.akka.io/docs/akka-http/10.0.5/scala/http/low-level-server-side-api.html#connection-failures>
> describes
> how to capture a connection failure that is caused by the client.
>
> However, I found that the recovery clause of reactToConnectionFailure is
> never executed.
>
> I tested by replacing
>
> .map { request =>
> // simple streaming (!) "echo" response:
> HttpResponse(entity = HttpEntity(ContentTypes.`text/plain(UTF-8)`,
> request.entity.dataBytes))
> }
>
> with
>
> .mapAsync(1) { request =>
> *Future.never*
> }
>
> so I have time to break the connection from the client side.
>
>
> Akka logging indicates
>
>> [DEBUG] [05/02/2017 13:04:33.645] [default-akka.actor.default-dispatcher-6]
>> [akka://default/system/IO-TCP/selectors/$a/3] Closing connection due to
>> IO error java.io.IOException: An established connection was aborted by the
>> software in your host machine
>
>
> The .log() operation simply indicates that "Upstream finished".
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-Http TrustStoreConfig Password?
I think that is probably an oversight. Please open a ticket over at the ssl-config project: https://github.com/typesafehub/ssl-config -- Johan Akka Team On Thu, Mar 23, 2017 at 3:57 PM, Aaron Broad <aaron.br...@gmail.com> wrote: > Hi all, > > I'm creating actors, where each Actor makes HTTP calls using a potentially > different keystore and truststore. > > I've successfully configured variable keystores with something like: > > sslConfig = AkkaSSLConfig(context.system).map(s => >s.withKeyManagerConfig( > s.keyManagerConfig.withKeyStoreConfigs( > Seq[KeyStoreCofig]( > KeyStoreConfig.getInstance( > Optional.empty[String](), > Optional.of(keystorePath)) > .withPassword(Some(keystorePassword)) > ))) > > However, when I go to do the same with a truststore... I don't see a > corresponding withPassword builder option? > > Am I missing something? > > How can I programmatically assign a password to a trust store in the > AkkaSSLConfig? > > thanks in advance, > Aaron Broad > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka http compilation error due to _marshaller
You need to have a marshaller that can turn your ServerSentEvents into data
to send over the wire. I'd recommend you to read up on the basics here:
http://doc.akka.io/docs/akka-http/10.0.7/scala/http/common/marshalling.html#marshalling
The SSE connector from Alpakka which you seem to be using deals with
connecting to a server and handling the streaming response back, but you
are trying use classes from it to stream events from your server. There is
another section in the docs about streaming body responses that should also
be helpful:
http://doc.akka.io/docs/akka-http/10.0.7/scala/http/routing-dsl/source-streaming-support.html#source-streaming
--
Johan
Akka Team
On Wed, Apr 26, 2017 at 1:22 PM, <nikhl...@ebctechnologies.com> wrote:
> Hi,
>
> I am using akka http for graphql sangria subscription.
> It is working fine in example application but while integrating graphql
> subscription code with this
> <https://github.com/sangria-graphql/sangria-relay-playground> simple
> query mutation code it is giving compilation error.
> Below is the code i'm trying to execute:
>
> complete(
> executor.prepare(queryAst, ctx, (), operation, variables)
> .map { preparedQuery ⇒
> ToResponseMarshallable(preparedQuery.execute()
> .map(result ⇒ ServerSentEvent(result.compactPrint))
> .recover { case NonFatal(error) ⇒
> logger.error(error, "Unexpected error during event stream
> processing.")
> ServerSentEvent(error.getMessage)
> })
> }
>
> It is showing error inline:
>
>
> ToResponseMarshallable(preparedQuery.execute()
>
>
> and i'm getting:
>
>
> could not find implicit value for parameter _marshaller:
> akka.http.scaladsl.marshalling.ToResponseMarshaller[akka.stream.scaladsl.Source[de.heikoseeberger.akkasse.ServerSentEvent,akka.NotUsed]]
>
>
> Can some one help me to resolve this error
>
>
> Thanks
>
>
>
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Making akka.pattern.BackoffSupervisor extendable
I think that would be more API surface than we'd like, but please open a ticket over at https://github.com/akka/akka/issues and we can discuss if it is worth doing there. -- Johan Akka Team On Tue, Mar 14, 2017 at 11:14 PM, Richard Ney <kamisama@gmail.com> wrote: > I was wondering if anyone has thought of making the BackoffSupervisor > extendable so the consumer can change the behavior of the child life-cycle? > I had a case where I wanted to control which Props was used to create the > child-actor in a Shard based on the first message. The application this was > specific to a Persistent actor that I want to change Recovery behavior when > restoring the actor after passivation if I determined I wanted to reset the > Actor's state. > > My final solution was to pull the BackoffSupervisor class out of the Akka > distribution and make a new MyBackoffSupervisor class with the 'final' > removed so I could override preStart() and then add in a new initialization > message handler to control the initial creation of the Actor before > changing over to the normal message handler. > > -Richard > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: ProducerStage and CassandraSink
The problem is that Future is not really suited for asynchronous work since there is no way to chain actions onto it, the only thing you can do is to poll or block until it is completed. It would have to be CompletableFuture. -- Johan Akka Team On Fri, Apr 28, 2017 at 5:54 PM, Richard Rodseth <rrods...@gmail.com> wrote: > Thanks. There *is* a version of send in Kafka that returns a Java Future. > > public java.util.concurrent.Future <https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/RecordMetadata.html>> > send(ProducerRecord > <https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/ProducerRecord.html> > <https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html>,V > > <https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html>> > record) > > Either way, I wish one could write with maximum possible throughput > without having a "parallelism" value to tune, but I guess Kafka would have > to provide a Reactive Streams compatible API, or Akka would have to drop > down to the Kafka protocol level, assuming that is still a thing. > > On Fri, Apr 28, 2017 at 7:09 AM, Akka Team <akka.offic...@gmail.com> > wrote: > >> The Cassandra session provides an async method session.executeAsync which >> returns a CompletableFuture/Future which makes it usable with mapAsync >> while Kafka has a callback based async api, where you trigger an action and >> pass a callback to execute when the action has completed. It would not be >> possible to use the latter with mapAsync. >> >> -- >> Johan >> Akka Team >> >> On Sat, Apr 8, 2017 at 1:08 AM, Richard Rodseth <rrods...@gmail.com> >> wrote: >> >>> I'm still curious what, if any, advantages the ProducerStage as in >>> reactive-kafka has over using mapAsync as in CassandraSink. Anyone? >>> >>> >>> On Fri, Mar 31, 2017 at 7:40 AM, Richard Rodseth <rrods...@gmail.com> >>> wrote: >>> >>>> I was looking at the implementation of the reactive-kafka >>>> ProducerStage. Very interesting. >>>> >>>> https://github.com/akka/reactive-kafka/blob/master/core/src/ >>>> main/scala/akka/kafka/internal/ProducerStage.scala >>>> >>>> It seems this pattern could be used for any external API that returns a >>>> Future. >>>> >>>> By contrast, the Alpakka CassandraSink uses mapAsyncUnordered with a >>>> user-supplied parallelism value. >>>> >>>> https://github.com/akka/alpakka/blob/master/cassandra/src/ma >>>> in/scala/akka/stream/alpakka/cassandra/scaladsl/CassandraSink.scala >>>> >>>> Am I correct in understanding that the ProducerStage generates demand >>>> (absence of backpressure) only as Futures complete, whereas the >>>> CassandraSink generates backpressure (absence of demand) when the >>>> parallelism value is reached. >>>> >>>> In the ProducerStage case, would the max number of in-flight Futures be >>>> limited by the internal buffer size, as described here? : >>>> >>>> http://doc.akka.io/docs/akka/2.4.17/scala/stream/stream-rate >>>> .html#Internal_buffers_and_their_effect >>>> >>>> Thanks. >>>> >>> >>> -- >>> >>>>>>>>>> Read the docs: http://akka.io/docs/ >>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>> urrent/additional/faq.html >>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>> p/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 https://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/c >> urrent/additional/faq.html >> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user >> --- >> You received this message because
Re: [akka-user] Akka FSM with predicate in Java
I tried to reproduce your problem, but I cannot see any problems with the
predicate not being honoured. Here's a gist with what I did to reproduce
it: https://gist.github.com/johanandren/7d01d19211867df0c308ba5fb1294162
--
Johan
Akka Team
On Mon, Apr 24, 2017 at 10:35 AM, Ankit Thakur <ankit...@gmail.com> wrote:
> Hi Guys,
>
> I have implemented a persistent FSM in Java, wherein i am evaluating a
> condition and post the validation of the same i decide to *stay* or *goTo*
> another state. I am a version just before the 2.5.0, so i am not very sure
> if the below mentioned issue is resolved or not.
>
> The issue is that i am not able to evaluate a predicate with *matchEvent *but
> the same works with *.event *construct. Below is a sample of the code.
>
> when(State.A,
> matchEvent(SomeEvent.class,
> (someEvent, domainEvent) -> {
> return ((some condition) ? false : true;
> }, (someEvent, domainEvent) ->
> {
> \* do something here *\
> return stay().
> applying(someEvent).
> andThen(\* do something here *\);
> }).
> event((SomeEvent.class,
> ((someEvent, domainEvent) -> {
> return (((some condition) ? true : false;
> }, (someEvent, domainEvent) ->
> {
>\* do something here *\
> return goTo(State.B).
> applying(someEvent).
> andThen(\* do something here *\);
> }));
>
> As mentioned above, the above piece of code does not honor the *condition
> *when coupled with *matchEvent*, but does work in case of *.event*.
>
> when(State.A,
> matchEvent(HackEventToLetFSMProcessPredicate.class,
> (hackEventToLetFSMProcessPredicate, domainEvent) -> {
> return ((some condition) ? false : true;
> }, (hackEventToLetFSMProcessPredicate, domainEvent) ->
> {
> \* do something here *\
> return stay().
> applying(someEvent).
> andThen(\* do something here *\);
> }).
> event((SomeEvent.class,
> ((someEvent, domain) -> {
> return (((some condition) ? true : false;
> }, (someEvent, domainEvent) ->
> {
>\* do something here *\
> return stay().
> applying(someEvent).
> andThen(\* do something here *\);
> }).
> event((SomeEvent.class,
> ((someEvent, domainEvent) -> {
> return (((some condition) ? true : false;
> }, (stateA, someEvent) ->
> {
>\* do something here *\
> return goTo(State.B).
> applying(someEvent).
> andThen(\* do something here *\);
>}
> ));
>
> Any help or update regarding this issue/bug is highly appreciated.
>
> Regards,
> Ankit
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] EndpointReader actor is a remoting bottleneck in case of heavy deserialisation
That is correct, there is one writer and one reader per connection in the stable remoting, so if serialization is slow/heavy that may be a bottleneck. In the new remoting, "artery", we have made it possible to run multiple in and outbound "lanes", but that functionality is not yet hardened enough to use. You can see the settings in reference.conf here: http://doc.akka.io/docs/akka/2.5.0/general/configuration.html#config-akka-remote-artery - inbound-lanes and outbound-lanes. -- Johan Akka Team On Fri, Apr 28, 2017 at 12:45 PM, Yaroslav Klymko <t3h...@gmail.com> wrote: > Hi guys, > > Do I understand correctly that we have a single instance of > `EndpointReader` actor per connection ? > Thus deserialisation of remote messages being done sequentially, so we can > hit this bottleneck much faster then a network throughput limits? > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-HTTP, Error: "Sending an 2xx 'early' response before end of request was received"
Looks like you are doing the right thing there, consuming the entity before
responding, could it be that the create methods throw an exception perhaps?
That would lead to the request body not being consumed as far as I can see.
--
Johan
Akka Team
On Thu, Feb 16, 2017 at 6:42 AM, Vasiliy Levykin <levykin.vasi...@gmail.com>
wrote:
> val uploadF = Future {
> Files.createDirectories(destPath.getParent)
> Files.createTempFile(destPath.getParent, "", ".tmp")
> }(filesystemEc) flatMap { tmpPath =>
>
> withTemporaryFile(tmpPath){ () =>
> for {
> _ <- ctx.request.entity.withSizeLimit(
> MaxUploadableSize).dataBytes.runWith(FileIO.toPath(tmpPath))
> lastModifiedTime <- commit(tmpPath, destPath)
> } yield lastModifiedTime
> }
> }
>
--
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] akka stream kafka newbie question
You should combine the same versions of Scala. If you use
akka-stream-kafka_2.12 then the Akka version you use (and all other
libraries written in Scala you use in fact) must have the same version of
Scala since the major versions of Scala (2.11, 2.12) are not binary
compatible. In addition to this you must also combine this with a version
of Akka that the driver works with. Easiest is probably to just use the
latest reactive-kaka and let it pull in the right versions of Akka as
transitive dependencies, you can do that by just having:
com.typesafe.akka
akka-stream-kafka_2.11
0.15
In your maven pomfile, if you want to explicitly list Akka as well,
you could then see in your IDE which versions that kaka-stream-kafka pulls
in and add those versions.
--
Johan
Akka Team
On Fri, Apr 28, 2017 at 4:38 PM, Shannon Ma <shanno...@gmail.com> wrote:
> With 2.12, i had to update the code,
>
>
> return new ProducerMessage.Message<SpecificRecord, SpecificRecord,
> ConsumerMessage.Committable>(
> new ProducerRecord<SpecificRecord, SpecificRecord>("akkatest",
> msg.record().key(), msg.record().value()), msg.committableOffset());
>
>
> but getting
>
> Exception in thread "main" java.lang.NoSuchMethodError:
> scala.Product.$init$(Lscala/Product;)V
> at akka.util.Timeout.(Timeout.scala:13)
> at akka.actor.ActorSystem$Settings.(ActorSystem.scala:328)
> at akka.actor.ActorSystemImpl.(ActorSystem.scala:683)
> at akka.actor.ActorSystem$.apply(ActorSystem.scala:245)
> at akka.actor.ActorSystem$.apply(ActorSystem.scala:288)
> at akka.actor.ActorSystem$.apply(ActorSystem.scala:233)
> at akka.actor.ActorSystem$.apply(ActorSystem.scala:224)
> at akka.actor.ActorSystem$.create(ActorSystem.scala:159)
> at akka.actor.ActorSystem.create(ActorSystem.scala)
> at com.javacodegeeks.camel.AkkaTest.main(AkkaTest.java:36)
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: ProducerStage and CassandraSink
The Cassandra session provides an async method session.executeAsync which returns a CompletableFuture/Future which makes it usable with mapAsync while Kafka has a callback based async api, where you trigger an action and pass a callback to execute when the action has completed. It would not be possible to use the latter with mapAsync. -- Johan Akka Team On Sat, Apr 8, 2017 at 1:08 AM, Richard Rodseth <rrods...@gmail.com> wrote: > I'm still curious what, if any, advantages the ProducerStage as in > reactive-kafka has over using mapAsync as in CassandraSink. Anyone? > > > On Fri, Mar 31, 2017 at 7:40 AM, Richard Rodseth <rrods...@gmail.com> > wrote: > >> I was looking at the implementation of the reactive-kafka ProducerStage. >> Very interesting. >> >> https://github.com/akka/reactive-kafka/blob/master/core/src/ >> main/scala/akka/kafka/internal/ProducerStage.scala >> >> It seems this pattern could be used for any external API that returns a >> Future. >> >> By contrast, the Alpakka CassandraSink uses mapAsyncUnordered with a >> user-supplied parallelism value. >> >> https://github.com/akka/alpakka/blob/master/cassandra/src/ >> main/scala/akka/stream/alpakka/cassandra/scaladsl/CassandraSink.scala >> >> Am I correct in understanding that the ProducerStage generates demand >> (absence of backpressure) only as Futures complete, whereas the >> CassandraSink generates backpressure (absence of demand) when the >> parallelism value is reached. >> >> In the ProducerStage case, would the max number of in-flight Futures be >> limited by the internal buffer size, as described here? : >> >> http://doc.akka.io/docs/akka/2.4.17/scala/stream/stream-rate >> .html#Internal_buffers_and_their_effect >> >> Thanks. >> > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] need some help extracting path without segments using mapInnerRoute
The RequestContext only contains the part of the path that is not yet
matched, it doesn't really know anything else about how you consumed the
path. "hello"/Segment/"world" would consume up to the end of "world", so I
cannot see how you would achieve the getName method as you describe it. I
think you would have devise your own path-like directive to achieve this,
using mapRequestContext and mapRequest would let you modify the request
path and the consumed path.
--
Johan
Akka Team
On Sun, Mar 19, 2017 at 8:55 PM, AL <123...@gmail.com> wrote:
> Hi,
> Is there an option to extract uri full without the segments values using ?
> simplified example:
>
> val route = get{
> path("hello"/Segment/"world"){ segment =>
> ctx.complete{
> StatusCodes.OK
> }
> }
> }
>
> I am want to collect the Api requests without the segment
>
> def getName(ctx: RequestContext): String = {
> ???
> }val extractor =
> mapInnerRoute { route => ctx =>
> try {
> ctx.complete(getName(ctx))
> } catch {
> case NonFatal(e) => ctx.complete(s"Got ${e.getClass.getSimpleName}
> '${e.getMessage}'")
> }
> }
> val checkedRoute = extractor{
> route}
>
> The function getName should return the uri path WITHOUT the segment value
> e.g:
>
> "hello.segment.world"
>
> "return api uri without segment value" in {
> Get("/hello/beautiful/world") ~> checkedRoute ~> check {
> responseAs[String] shouldEqual "hello.segment.world"
> }
> }
>
>
> is that possible with mapInnerRoute ? or I have to explicitly implement
> and call it in each route ?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Graceful Termination
Hi,
When I try those exact sources it does gracefully shutdown after processing
all of them.
Your what you want it to do does not match what it actually does however,
if you want each state to process sequentially in an actor you will need to
have some protocol for completion of the processing so that the top level
actor does not send the next state until the previous one was completed.
And for the state actor it needs to know when all its cities are done
processing.
A few additional general notes:
Don't use standard library types such as String as messages unless you have
a very good reason to, create your own message classes, using generic
classes makes applications hard to follow and is error prone.
I see you make the threads sleep, it is very important to never do that, as
the dispatcher (thread pool) that the actors are dispatched on has few
threads, this way you will starve the thread pool with the first n actors
that run, and no other actor can run until their sleeps has completed.
--
Johan
Akka Team
On Mon, Mar 27, 2017 at 4:45 AM, <kongunattuara...@gmail.com> wrote:
> Hello,
>
> I am trying to achieve this use case part of my learning exercise of akka.
> Calculate total number of streets in all the cities of all states. I have a
> database which contains all states, cities and streets i need. I am trying
> to achieve this in Java
>
> Here is what i have
>
> *Configuration:*
> akka.actor.deployment {
> /CityActor{
> router = random-pool
> nr-of-instances = 10
> }
> /StateActor {
> router = random-pool
> nr-of-instances = 1
> }
> }
>
>
> Intention of the above configuration is process one state at a time and
> within each state 10 cities are processed in parallel. If needed in future,
> i would like to process multiple states in parallel as well.
>
> *Main:*
> public static void main(String[] args) {
>
> try {
>
> Config conf = ConfigFactory
> .parseReader(
> new FileReader(ClassLoader.getSystemResource("config/
> forum.conf").getFile()))
> .withFallback(ConfigFactory.load());
>
> System.out.println(conf);
>
> final ActorSystem system = ActorSystem.create("AkkaApp", conf);
>
> final ActorRef masterActor = system.actorOf(Props.create(MasterActor.class),
> "Migrate");
>
> masterActor.tell("", ActorRef.noSender());
>
> } catch (Exception e) {
> e.printStackTrace();
> }
> }
>
> *MasterActor:*
> public class MasterActor extends UntypedActor {
>
> private final ActorRef randomRouter = getContext().system()
> .actorOf(Props.create(StateActor.class).withRouter(new
> akka.routing.FromConfig()), "StateActor");
>
> @Override
> public void onReceive(Object message) throws Exception {
>
> if (message instanceof String) {
>
> getContext().watch(randomRouter);
>
> for (String aState : getStates()) {
> randomRouter.tell(aState, getSelf());
> }
> randomRouter.tell(new Broadcast(PoisonPill.getInstance()), getSelf());
>
> } else if (message instanceof Terminated) {
>
> Terminated ater = (Terminated) message;
>
> if (ater.getActor().equals(randomRouter)) {
> getContext().system().terminate();
> }
> }
> }
>
> public List getStates() {
> return new ArrayList(Arrays.asList("CA", "MA", "TA", "NJ", "NY"));
> };
>
> }
>
> *StateActor:*
> public class StateActor extends UntypedActor {
>
> private final ActorRef randomRouter = getContext().system()
> .actorOf(Props.create(CityActor.class).withRouter(new
> akka.routing.FromConfig()), "CityActor");
>
> @Override
> public void onReceive(Object message) throws Exception {
> if (message instanceof String) {
> System.out.println("Processing state " + message);
>
> for (String aCity : getCitiesForState((String) message)) {
> randomRouter.tell(aCity, getSelf());
> }
> Thread.sleep(1000);
> }
> }
>
> public List getCitiesForState(String stateName) {
> return new ArrayList(Arrays.asList("Springfield-" + stateName,
> "Salem-" + stateName,
> "Franklin-" + stateName, "Clinton-" + stateName, "Georgetown-" +
> stateName));
> };
> }
>
>
> *CityActor:*
> public class CityActor extends UntypedActor {
>
> @Override
> public void onReceive(Object message) throws Exception {
> if (message instanceof String) {
> System.out.println("Processing city " + message);
> Thread.sleep(1000);
> }
> }
> }
>
> Am i implementing this correctly?
>
> I cannot get the program to shutdown gracefully. Not sure whether i
> implemented the PosionPill logic properly. I get message like dead letters
Re: [akka-user] Diffrence between two way to init Akka Router ?
If you look at the sources of Pool.props(props) all it does is calling props.withRouter(this), so the two options are pretty much identical so you can choose whichever you like best. -- Johan Akka Team On Wed, Apr 12, 2017 at 6:27 PM, Tu Pham Phuong <pham...@gmail.com> wrote: > My system run with Play framework 2.5 and Akka build-in. I believe there > are two ways to init a route. > > 1: > > ActorRef router = akkaService.getActorSystem().actorOf(new > RoundRobinPool(poolSize) .props(Props.create(ProfileActor.class)), > "ProfileActor"); > > 2: > > ActorRef router = akkaService.getActorSystem().actorOf(new > Props(ProfileActor.class).withRouter( new RoundRobinRouter(poolSize)), > "ProfileActor")); > > So I have 2 questions: > >- - Are they difference? >- - Which way is better? > > Thank you so much ! > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Is Akka Multi Node Testing available in Java
Note that the Java akka-testkit and multi-node-testkit support are different things though (ticket 18109 is about akka-testkit). Support for multi-ivm tests in Java would require both new Java APIs and tooling support (maven plugin?). The tooling support part is quite complicated and the reason why there only is SBT+Scala support right now. Adding a Java API and allowing that to be used with SBT would be easier though. -- Johan Akka Team On Thu, Apr 27, 2017 at 1:10 PM, 'Francesco laTorre' via Akka User List < akka-user@googlegroups.com> wrote: > Moreover, a very good opportunity for a community contribution : > https://github.com/akka/akka/issues/18109 > > Cheers, > Francesco > > On 27 April 2017 at 03:06, Dai Yinhua <yhdai.2...@gmail.com> wrote: > >> OK, thanks. >> >> On Tuesday, 25 April 2017 16:59:10 UTC+8, Akka Team wrote: >>> >>> Hi Dai, >>> >>> I'm afraid it is currently only possible to run the multi-node tests >>> with Scala + SBT >>> >>> -- >>> Johan >>> Akka Team >>> >>> On Wed, Mar 22, 2017 at 3:11 AM, Dai Yinhua <yhdai...@gmail.com> wrote: >>> >>>> Anybody knows how to run akka multi node testing in java? >>>> Is there any example code? >>>> >>>> -- >>>> >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>>> urrent/additional/faq.html >>>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>>> p/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 https://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/c >> urrent/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 https://groups.google.com/group/akka-user. >> For more options, visit https://groups.google.com/d/optout. >> > > > > -- > <http://www.openbet.com/> Francesco laTorre > Senior Developer > T: +44 208 742 1600 > +44 203 249 8394 > > E: francesco.lato...@openbet.com > W: www.openbet.com > OpenBet Ltd > Chiswick Park Building 9 > 566 Chiswick High Rd > London > W4 5XT > <https://www.openbet.com/email_promo> > This message is confidential and intended only for the addressee. If you > have received this message in error, please immediately notify the > postmas...@openbet.com and delete it from your system as well as any > copies. The content of e-mails as well as traffic data may be monitored by > OpenBet for employment and security purposes. To protect the environment > please do not print this e-mail unless necessary. OpenBet Ltd. Registered > Office: Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT, > United Kingdom. A company registered in England and Wales. Registered no. > 3134634. VAT no. GB927523612 > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Multi Host Master with Remote Worker
These two sample projects shows an example of this: https://github.com/typesafehub/activator-akka-distributed-workers https://github.com/typesafehub/activator-akka-distributed-workers-java -- Johan Akka Team On Fri, Mar 3, 2017 at 9:52 PM, <learnspr...@gmail.com> wrote: > How do i create Master on Node 1 and deploy worker on different nodes , > Any help for configuration or example will be a great help > > Thanks, > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c > urrent/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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka queue stream recover and drop element back in top
Recover will catch an exception coming from upstream and continue by
emitting a single element before completing. Indeed does not sound like
what you are after.
Maybe the retry stream in akka-stream-contrib could give you some
inspiration on how to solve your problem:
https://github.com/akka/akka-stream-contrib/blob/master/contrib/src/main/scala/akka/stream/contrib/Retry.scala
--
Johan
Akka Team
On Wed, Apr 5, 2017 at 5:15 PM, scallahan via Akka User List <
akka-user@googlegroups.com> wrote:
> Hey all,
>
> In my example below I have a source queue. In the middle I have a flow
> that can fail for a few different network related reason and all I want to
> do is drop the request back in the top of the funnel. But I am not entirely
> sure how I can get access to the element that went into the flow and caused
> the failure.
>
> val queue = Source.queue[ArchiveValidationRequest](50,
> OverflowStrategy.fail)
> queue.via(alwaysOnFlow).recover {
> case ex: Throwable =>
> error("Element failed. Reinserting back into queue to be processed")
> queue.offer(???)
> }
>
> This seems like a fairly basic pattern but its not super obvious how to
> get that failed request. recoverWithRetries doesn't seem to be what I want
> here as I just want to capture the failed element and just try to process
> it later.
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Offset type and resumable projections for 2.5
I don't find that sample odd or doing something inherently bad.
You might choose to serialize the whole Offset subtype, but you might also
decide to serialize the long-value which will take considerable less space,
may be possible to read directly from a database etc. I consider that
design decisions, the sample is just there to show that you would start
from a stored offset and store offset as the query progress.
--
Johan
Akka Team
On Thu, Apr 13, 2017 at 11:30 PM, Richard Rodseth <rrods...@gmail.com>
wrote:
> Congrats on the release of 2.5.
>
> Isn't the documentation and accompanying sample for Resumable Projections
> a bit odd?
>
> http://doc.akka.io/docs/akka/2.5.0/scala/persistence-query.
> html#Resumable_projections
> https://github.com/akka/akka/blob/master/akka-docs/rst/
> scala/code/docs/persistence/query/PersistenceQueryDocSpec.scala
>
> Why would we be constructing a Sequence? Wouldn't the offset be stored in
> a plugin-agnostic way and just passed back? In my case, the offset is in a
> Protobuf persistence event, and I'm assuming I would have to convert it
> (whether Long or TimeUUID) to a ByteString and use a Protobuf field of type
> "bytes".
>
>
>1. bidProjection.latestOffset.foreach { startFromOffset =>
>2. readJournal
>3. .eventsByTag("bid", Sequence(startFromOffset))
>4. .mapAsync(8) { envelope => (writer ? envelope.event).map(_ =>
>envelope.offset) }
>5. .mapAsync(1) { offset => bidProjection.saveProgress(offset) }
>6. .runWith(Sink.ignore)
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Apply different functions on SubFlows
Partition allows you to look at a message and decide which of its outlets
to pass the element to, this way you can design the logic of each sub
stream independently. Something like this:
val flow: Flow[Int, Int, NotUsed] = Flow.fromGraph(GraphDSL.create() {
implicit builder =>
import GraphDSL.Implicits._
val partition = builder.add(Partition[Int](2, _ % 2))
val merge = builder.add(Merge[Int](2))
val processEven = builder.add(Flow.fromFunction((n: Int) => n + 100))
val processOdd = builder.add(Flow.fromFunction((n: Int) => n - 1))
partition.out(0) ~> processEven ~> merge.in(0)
partition.out(1) ~> processOdd ~> merge.in(1)
FlowShape(partition.in, merge.out)
})
--
Johan
Akka Team
On Tue, Mar 21, 2017 at 1:18 PM, Vishal John <vishalj...@gmail.com> wrote:
>
> Hello all,
>
>
> I have defined my application flow like this,
>
> *val flow = Flow[StringConsumerRecord].*
> * map(tup => new Ticket(tup.value.toLong)).*
> * mapConcat(applyBusinessLogic(_)).*
> * groupBy(4, _._1).*
> * groupedWithin(100, 5.second).*
> * mergeSubstreams.*
> * throttle(1, 1.second, 1, ThrottleMode.shaping)*
>
>
> Basically I don't want to apply `*groupedWithin(100, 5.second)*` for all
> the sub flows.
>
> Based on the type of the grouped data, for one sub stream I want to do
> *groupedWithin(10,
> 10.second) *and for another *groupedWithin(1000, 1.second) etc... *and
> finally merge all. Basically I am trying to do some sort to pattern
> matching on the sub-flows and take different actions.
>
> Can someone give any pointers on how to achieve this.
>
>
> -thanks
> Vishal
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Could not access type LoggingAdapter in value akka.event
Akka 2.3 has recently been end-of-lifed, please upgrade to 2.5.0 or 2.4.17. Spray has been end of lifed, superseded by Akka HTTP, so you should definitely think about migrating. The problem you are seeing is probably spray or some other library depending on a newer Akka version than 2.3.14 -- Johan Akka Team On Fri, Mar 24, 2017 at 5:47 AM, Ravikumar Maddi <ravi.ma...@gmail.com> wrote: > up vote > down votefavorite > <http://stackoverflow.com/questions/42990024/could-not-access-type-loggingadapter-in-value-akka-event#> > > I migrated from Scala 2.10.8 to 2.11.7 and Akka 2.1.2 to 2.3.14. I did lot > of changes in *.sbt and source also. Now, I am getting some issue. I check > online, but I did not get acutual cause. I think the spray version is not > suite to Akka version. But which version of Spray I have to use for Akka > 2.3.14, I did not get any information yet online and Akka documents also. > > Please help me. > > The Issue is: > > [error] missing or invalid dependency detected while loading class file > 'LoggingContext.class'.[error] Could not access type LoggingAdapter in value > akka.event,[error] because it (or its dependencies) are missing. Check your > build definition for[error] missing or conflicting dependencies. (Re-run with > `-Ylog-classpath` to see the problematic classpath.)[error] A full rebuild > may help if 'LoggingContext.class' was compiled against an incompatible > version of akka.event. > > My Project dependency is this: > > https://www.dropbox.com/s/tp1ruackt9tued6/dependencyTree2.txt?dl=0 > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Idiomatic way to use akka stream Source within Spark
I think this question has a higher chance of getting an answer if asked in
the Spark community/mailing list.
The linked project seems to be more about feeding data from Akka Streams
into spark than running Akka Streams inside of Spark.
--
Johan
Akka Team
On Fri, Mar 17, 2017 at 8:49 PM, Kyrylo Stokoz <k.sto...@gmail.com> wrote:
>
> Hi All,
>
> I`m trying to figure out how one should use Sources within Spark Jobs
> keeping all benefits of Spark.
> Consider following snippet:
>
> val items: Seq[String] = Seq("a", "b", "c")
> sparkSession.sparkContext.parallelize(items, 10).flatMap { item =>
> val subItems: Source[String, _] = f(item)
> // (1)
> ???
> }.map { subItem =>
> f1(subItem)
> }.reduce(_ + _)
>
>
> There is this project https://github.com/lloydmeta/sparkka-streams which
> tries to bridge akka streams and Spark Streaming.
> I also created an IteratorSinkStage which materialized into Iterator
> similar as InputStreamSinkStage which is materialized into InputStream, but
> i`m not sure if this is best solution for this problem.
>
> What would be the best way to work with akka Source within Spark
> environment?
>
> Regards,
> Kyrylo
>
>
>
>
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] How to get all the Actors that are holding a particular web socket endpoint using Play framework?
For more Play Framework specific answers I'd ask on the play mailing list instead. In general: If it has to be actors representing the websockets I'd let them register to a registry/broadcast actor that will keep a list of current web sockets (using watch to know if they terminate for example) and forward messages to all currently living web socket actors. In Akka Streams BroadcastHub will allow you to do something like this without involving actors directly. -- Johan Akka Team On Sun, Mar 19, 2017 at 1:08 PM, kant kodali <kanth...@gmail.com> wrote: > Hi All, > > I have a message queue which doesn't have Actor concepts or anything so > what I plan to do is that on the application startup I want to start the > consumer which will then keep getting messages from the queue. Now, Play > framework creates an Actor for every Web Socket Connection and I want to be > able to group all the Actors that hold the Web Socket connection for a > particular ws endpoint so that I can broadcast all the messages that I > received from message queue on particular topic to those group of Actors. > > For Example the following end points will have an Actor created every time > a client initiates a request to any of the end point below. so lets call > them Foo actors and Bar Actors. > > ws://localshost/foo > > ws://localshost/bar > > https://www.playframework.com/documentation/2.5.x/JavaWebSockets > > Now all I want to do is this > > Pseudo code: > > messages = ReceiveMessagesFromQueue; // This is a live stream and it > never stops. > for message in messages: > if message has key1: >List foo_list = getAllFooActors >broadcast(message, foo_list) > else if message has key2: >List bar_list = getAllBarActors >broadcast(message, bar_list) > > Please Let me know if you have any suggestions on the best way to approach > this problem using Play Framework (I am using the latest version). > > Thanks! > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Is Akka Multi Node Testing available in Java
Hi Dai, I'm afraid it is currently only possible to run the multi-node tests with Scala + SBT -- Johan Akka Team On Wed, Mar 22, 2017 at 3:11 AM, Dai Yinhua <yhdai.2...@gmail.com> wrote: > Anybody knows how to run akka multi node testing in java? > Is there any example code? > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: Akka HTTP path matching fundamentally broken? Surely not?
Did you read the docs about the various path directives and how they differ?
http://doc.akka.io/docs/akka-http/10.0.4/scala/http/routing-dsl/directives/path-directives/index.html
--
Johan
Akka Team
On Fri, Mar 17, 2017 at 12:47 PM, Alan Burlison <alan.burli...@gmail.com>
wrote:
> pathPrefix("root") {
>>
>
> I can bodge around this with:
>
> pathPrefix("^root$".r)
>
> but that's unspeakably vile.
>
>
> --
> Alan Burlison
> --
>
> --
>
>> Read the docs: http://akka.io/docs/
>>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c
>>>>>>>>>>> urrent/additional/faq.html
>>>>>>>>>>> Search the archives: https://groups.google.com/grou
>>>>>>>>>>> p/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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: [Akka streams] - filter with mapAsync strange behavior
Hi Alexandre,
This is how the mapAsync stage works, the parallelism parameter controls
the top number of Futures/CompletionStages the stage should have processing
at once, so when the previous Future/CompletionStage completes and it's
value is emitted downstream it will immediately start the next future.
If you look at the doc of the method it actually even says so, just that it
takes some careful reading to see that:
* '''Backpressures when''' the number of futures reaches the
configured parallelism and the downstream
* backpressures or the first future is not completed
I think we can improve the wording so it is more clear though, I have
opened a ticket for that: https://github.com/akka/akka/issues/22525
Cheers
--
Johan
Akka Team
On Fri, Mar 10, 2017 at 9:50 AM, Alexandre Delegue <
alexandre.dele...@serli.com> wrote:
> Hi,
>
> should open an issue for this ?
>
> Thanks.
>
>
> Le mardi 7 mars 2017 08:20:35 UTC+1, Alexandre Delegue a écrit :
>>
>> Hi,
>>
>> Executing this code
>>
>> Source.range(1, 4)
>> .mapAsync(1, e -> CompletableFuture.supplyAsync(() -> {
>> System.out.println("Num "+e);
>> return e;
>> }))
>> .filter(any -> true)
>> .take(1)
>> .runWith(Sink.foreach(e -> { System.out.println("End "+e); }), mat);
>>
>>
>> Will print
>> Num 1
>> End 1
>> Num 2
>>
>> I don't understand why the async call in mapAsync is executed for number
>> 2 while the stream should be completed.
>>
>> This work fine
>>
>> Source.range(1, 4)
>> .mapAsync(1, e -> CompletableFuture.supplyAsync(() -> {
>> System.out.println("Num "+e);
>> return e;
>> }))
>> .take(1)
>> .runWith(Sink.foreach(e -> { System.out.println("End "+e); }), mat);
>>
>>
>> So it seems that the filter method pull next element even if the stream
>> is completed.
>>
>> Any idea? Is there a way to avoid that ?
>>
>> Thanks,
>> Alexandre.
>>
>>
>> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] akka cluster, dual cluster routers against the same pool of actors
The routers does not guarantee delivery, you are likely firing up the
system and sending messages before the group has learned about the actor on
the other nodes.
Here's an adaption of your config in a fully working sample that waits a
bit to give the groups time to have found the receiving pools that works as
expected:
https://gist.github.com/johanandren/35ed29f320ec3a0cece0946fcd70eb89
Cheers
--
Johan
Akka Team
On Tue, Mar 7, 2017 at 6:03 PM, Arkaitz Jimenez <arkaitz.jime...@teamaol.com
> wrote:
> /ClusterAwareActor {
> router = random-group
> routees.paths = ["/user/EchoGroup"]
> cluster {
> enabled = on
> allow-local-routees = off
> }
> }
> /EchoGroup {
> router = random-pool
> nr-of-instances = 1
> pool-dispatcher {
> executor = "fork-join-executor"
> fork-join-executor {
> parallelism-min = 20
> parallelism-max = 20
> }
> }
>
>
> }
--
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] PersistentActor and BackoffSupervisor problen
Hi Alan, I think the sample in the doc section referred to already uses the right factory method and describes why, but please PR the docs with something that makes it more clear if you want! (http://doc.akka.io/docs/akka/current/scala/persistence.html#Failures) -- Johan Akka Team On Wed, Mar 8, 2017 at 4:55 PM, Alan Burlison <alan.burli...@gmail.com> wrote: > On 08/03/17 15:07, Akka Team wrote: > > What it says is, do not use this factory method for persistent actors but >> instead the Backoff.onStop factory method, not that the back off >> supervisor >> cannot be used with persistent actors. >> > > Ahah, that makes sense - thanks for the clarification. Perhaps the docs (I > mean http://doc.akka.io/docs/akka/current/scala/persistence.html rather > than the scaladocs) could be tweaked slightly to make the distinction clear? > > > -- > Alan Burlison > -- > > -- > >> Read the docs: http://akka.io/docs/ >>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>>>>>>>>>> urrent/additional/faq.html >>>>>>>>>>> Search the archives: https://groups.google.com/grou >>>>>>>>>>> p/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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Sharding not working in some deployment
Hi, If you want to limit which nodes that will host shard regions you must assign them a role and specify that role with akka.cluster.sharding.role. Then you must start sharding on all nodes with that role. Maybe this is the problem you are having. Sharding has got pretty extensive debug logging, enable debug logging and maybe it can help you figure it out. Cheers -- Johan Akka Team On Tue, Feb 28, 2017 at 11:07 AM, Dai Yinhua <yhdai.2...@gmail.com> wrote: > Hi Akka team, > > I am developing an application with cluster sharding. > And I have below deployment: > server1(3 jvms): seed 1, seed 2, endpoint > server2(2 jvms): worker 1, worker 2 > server3(2 jvms): worker 1, worker 2 > > The seed instance will act only the role of seed, no shard region is > started. > The endpoint instance will start shardregion proxy, which is used to > accept external requests. > The worker instance will start shard region to do actual work. > > But the sharding function is not working with this deployment, I get error > messages to send message to the singleton shardcoordinator, looks like > shard coordinator is not started. > > I realised that the singleton actor is running on the leader node, so in > my case, the lead node is the seed 1, is the error caused by the shard > region is not started in the seed node? > When I use worker 1 as the seed node, the sharding function is working > correctly. > > Can you help to check what is the problem? > Thank you. > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] PersistentActor and BackoffSupervisor problen
Hi Alan, That is most certainly in the docs and not only a source comment: http://doc.akka.io/api/akka/2.4/akka/pattern/Backoff$.html#onFailure(childProps:akka.actor.Props,childName:String,minBackoff:scala.concurrent.duration.FiniteDuration,maxBackoff:scala.concurrent.duration.FiniteDuration,randomFactor:Double):akka.pattern.BackoffOptions What it says is, do not use this factory method for persistent actors but instead the Backoff.onStop factory method, not that the back off supervisor cannot be used with persistent actors. -- Johan Akka Team On Wed, Mar 8, 2017 at 4:00 PM, Alan Burlison <alan.burli...@gmail.com> wrote: > On 08/03/17 14:45, Richard Rodseth wrote: > > I think they are referring to child actors. i.e. it's OK for actor A to >> have a backoff supervisor (A can be persistent or not), but if it has a >> child which is a persistent actor then the caveats apply. >> > > Hmm, perhaps - but it doesn't read that way to me. > > > -- > Alan Burlison > -- > > -- > >> Read the docs: http://akka.io/docs/ >>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>>>>>>>>>> urrent/additional/faq.html >>>>>>>>>>> Search the archives: https://groups.google.com/grou >>>>>>>>>>> p/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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Balancing TCP requests
It looks correct to me, what is it that makes you think it is not the
server closing the connection as the exception says?
--
Johan
Akka Team
On Thu, Jan 26, 2017 at 3:20 PM, Maksym Besida <maks.bes...@gmail.com>
wrote:
> Based on this paragraph
> <http://doc.akka.io/docs/akka/2.4.16/scala/stream/stream-cookbook.html#Balancing_jobs_to_a_fixed_pool_of_workers>
> I've
> implemented a balancer for tcp outgoing requests.
>
> def outGoing(host: String, port: Int) =
> Tcp().outgoingConnection(InetSocketAddress.createUnresolved(host, port),
> halfClose = true)
>
>
> def balancer(endpoints: Seq[(String, Int)]): Flow[ByteString, ByteString,
> NotUsed] = {
> Flow.fromGraph(GraphDSL.create(){ implicit b =>
> import GraphDSL.Implicits._
> val balancer = b.add(Balance[ByteString](endpoints.length,
> waitForAllDownstreams = true))
> val merge = b.add(Merge[ByteString](endpoints.length))
>
> endpoints.foreach { case(host, port) =>
> balancer ~> outGoing(host, port).async ~> merge
> }
>
> FlowShape(balancer.in, merge.out)
> })
> }
>
>
> I use it as follows
>
> source.via(balancer(Seq(("localhost", 22002), ("localhost",
> 22002.runWith(Sink.ignore)
>
>
> I'm not interseted in result so I run it with Sink.ignore. I open 2 tcp
> connections to the same destination host for the test purpose(in reality it
> will go to different hosts)
>
> When I produce a lot of elments in my source I eventually(not after first
> message) get next error and stream fails
> akka.stream.StreamTcpException: The connection closed with error: An
> existing connection was forcibly closed by the remote host
>
> Can anybody suggest what is wrong with the example?
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Database query does not cancel when streaming rows using a DatabasePublisher
>From the slick docs:
Execution of the DBIOAction does not start until a Subscriber is attached
to the stream. Only a singleSubscriber is supported, and any further
attempts to subscribe again will fail ...
I'm guessing what you actually want is not to create your publisher once
upon application start and subscribe to that multiple time like you do now
but rather once for every request that comes in to webservice/csv (so
inside the block of the get directive.
--
Johan
Akka Team
On Wed, Feb 1, 2017 at 3:27 PM, Mark Goldenstein <bigande...@gmail.com>
wrote:
> Hi guys!
>
> (Cross-posting this since I did not get a response in the Slick group.)
>
> I use akka-http together with Slick to stream rows from a mysql database
> as csv via http.
>
> Streaming works as expected unless I disconnect the http connection
> prematurely. In this case I would expect that the sql query should stop
> executing since there is no downstream demand anymore. However, I can see
> in the mysql processes that the query is still running and it only stops
> when I shut down the webserver (or when the query finishes fetching all the
> rows).
>
> If I turn on debug logs I see repeatedly, while streaming:
>
> 22:05:17.738 [db-8] DEBUG slick.basic.BasicBackend.stream - Suspending
> streaming action with continuation (more data available)
> 22:05:17.743 [default-akka.actor.default-dispatcher-8] DEBUG
> slick.basic.BasicBackend.stream - Scheduling stream continuation after
> transition from demand = 0
> 22:05:17.743 [db-9] DEBUG slick.basic.BasicBackend.stream - Restarting
> streaming action, realDemand = 8
>
> Then, once I disconnect the connection, and there is no downstream demand
> anymore, I see:
>
> 22:05:17.744 [db-9] DEBUG slick.basic.BasicBackend.stream - Suspending
> streaming action with continuation (more data available)
> 22:05:17.749 [default-akka.actor.default-dispatcher-7] DEBUG
> akka.io.TcpIncomingConnection - Closing connection due to IO error
> java.io.IOException: Broken pipe
> 22:05:17.774 [default-akka.actor.default-dispatcher-16] DEBUG
> slick.basic.BasicBackend.stream - Scheduling stream continuation after
> transition from demand = 0
> 22:05:17.774 [db-10] DEBUG slick.basic.BasicBackend.stream - Restarting
> streaming action, realDemand = oo
>
> What is going on here? Why is realDemand oo?
>
> My code looks something like this:
>
> object Main extends App {
> implicit val system = ActorSystem()
> implicit val executor = system.dispatcher
> implicit val materializer = ActorMaterializer()
>
> implicit val csvMarshaller =
> Marshaller.withFixedContentType[CSVLine,
> ByteString](ContentTypes.`text/csv(UTF-8)`) {
> case CSVLine(line) => ByteString(line)
> }
>
> implicit val csvStreamingSupport = EntityStreamingSupport.csv()
> .withParallelMarshalling(parallelism = 8, unordered = false)
>
> val query = ??? // a Slick query
>
> val publisher: DatabasePublisher[Tick] = DB.get.stream(
> query.result.withStatementParameters(statementInit = DB.enableStream))
>
> val routes = {
> logRequestResult("webservice") {
> encodeResponse {
> pathPrefix("csv") {
> pathSingleSlash {
> get {
> complete {
> Source.fromPublisher(publisher).map(t =>
> CSVLine(t.toCSV()))
> }
> }
> }
> }
> }
> }
> }
>
> Http().bindAndHandle(routes, "127.0.0.1", 9000)
> }
>
> object DB {
> private val db = Database.forConfig("db")
>
> def get = db
>
> def enableStream(statement: java.sql.Statement): Unit = {
> statement match {
> case s if s.isWrapperFor(classOf[com.mysql.jdbc.StatementImpl]) =>
> s.unwrap(classOf[com.mysql.jdbc.StatementImpl]).enableStream
> ingResults()
> case _ => // do nothing
> }
> }
> }
>
> Any thoughts on this?
>
> Mark
>
> --
> >>>>>>>>>> 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 https://groups.google.com/group/akka-user.
> For more options, visit https://g
Re: [akka-user] Chunked Response handling with Akka Http?
This is definitely possible, just use Chunk for each element instead, that
way you can create them from a stream of strings using .map()
For the "an async request and it responds one or more times" there are a
few possible tools in Akka streams that may solve this. Take a look at
Source.unfoldAsync, Source.unfoldResourceAsync and Source.queue and see if
either of those fit your use case.
--
Johan
Akka Team
On Mon, Jan 16, 2017 at 8:03 AM, <subo...@gmail.com> wrote:
> I'd like to be able to use akka-http with chunked responses. I'd like to know
> if I can do the same thing in akka-http as with the playframework. The
> following is from playframework documentation.
>
> def index = Action {
> val source = Source.apply(List("kiki", "foo", "bar"))
> Ok.chunked(source)
> }
>
> We can inspect the HTTP response sent by the server:
>
> HTTP/1.1 200 OK
> Content-Type: text/plain; charset=utf-8
> Transfer-Encoding: chunked
>
> 4
> kiki
> 3
> foo
> 3
> bar
> 0
>
> With akka-http I was able to get close with
>
> val hi = Chunk("Hi" + "\n")
>
> val there = Chunk("there" + "\n")
>
> val last = LastChunk("boom" + "\n")
>
> val source: Source[ChunkStreamPart, NotUsed] =
> Source(scala.collection.immutable.Iterable(hi, there, last))
>
> val rootPath: Route =
>
>path("") {
>
> get {
>
>for (i <- 1 to 100) {
>
> ref ! hi
>
>}
>
>complete(Chunked(ContentTypes.`text/plain(UTF-8)`, source))
>
> }
>
>}
>
> def routes: Route = rootPath
>
> }
>
> curl returns
>
> < HTTP/1.1 200 OK
>
> < Server: akka
>
> < Date: Mon, 16 Jan 2017 14:03:10 GMT
>
> < Transfer-Encoding: chunked
>
> < Content-Type: text/plain; charset=UTF-8
>
> <
>
> Hi
>
> there
>
>
>
> Two perhaps minor issues: it does not report the last element, and it doesn't
> include the chunk size as the playframework example did.
>
> What I’d like to know is how can I do this in a dynamic way? I'm interacting
> with a embedded device (via serial) where I make an async request and it
> responds one or more times up to about 20 seconds. I'd like each response to
> be emitted as a chunk. Is this possible?
>
> Thanks,
> Andrew
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Class cast exceptions in actors that return Future[Unit]
If it in fact was a Unit you got back from the actor then neither of the things you tried should throw a class cast exception though. If the actor for example sends a Future[Unit] back, then you will have a Future[Future[Unit]] returned from ask. -- Johan Akka Team On Tue, Jan 31, 2017 at 2:41 PM, Alan Burlison <alan.burli...@gmail.com> wrote: > On 31/01/2017 21:36, Viktor Klang wrote: > > What'd be the alternative? >> > > Not sure really, don't know if there's any way of preserving a > Future[Unit] rather than converting it to a Future[Any]. As I said, it just > seemed a bit clunky, and not obvious immediately obvious as to how to solve > the issue. Perhaps just a note in the docs about how to handle Future[Unit]? > > > -- > Alan Burlison > -- > > -- > >> Read the docs: http://akka.io/docs/ >>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>>>>>>>>>> urrent/additional/faq.html >>>>>>>>>>> Search the archives: https://groups.google.com/grou >>>>>>>>>>> p/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 https://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam -- >>>>>>>>>> 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Class cast exceptions in actors that return Future[Unit]
You have a Future[Something], and a Something is an Any (everything is) but it is not Unit (only Unit is), so you can not just cast it, you must replace it with an actual Unit, which is what .map(_ => ()) does. -- Johan Akka Team On Tue, Jan 31, 2017 at 11:35 AM, Alan Burlison <alan.burli...@gmail.com> wrote: > I'm writing an Akka persistence journal that saves data in JSON format. > I've done that as normal, by subclassing AsyncWriteJournal and implementing > the necessary methods. > > The actual file IO is done by sub-Actors, one per output file where each > persistent actor has its own JSON file. The read/write methods on the IO > Actor which return a value via 'ask' return a Future[Any] as per normal. > However the asyncXXX methods in AsyncWriteJournal return Future[Unit], so > the Future[Any] returned from the IO Actor needs to be converted to a > Future[Unit]. > > I've tried a couple of ways of doing that: > > 1) Casting the return value with ".asInstanceOf[Future[Unit]]". This > generates a ClassCast exception: > > 2) Applying ".mapTo[Unit]" to the return value. This generates a ClassCast > exception: > > 3) Applying ".map(_ => ())" to the return value. This works, but sort of > smells a bit. > > Is there a cleaner way of doing this? > > -- > Alan Burlison > -- > > -- > >> Read the docs: http://akka.io/docs/ >>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>>>>>>>>>> urrent/additional/faq.html >>>>>>>>>>> Search the archives: https://groups.google.com/grou >>>>>>>>>>> p/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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka Cluster and custom code deployment (transfer the classes to the nodes to start new actors at runtime)
Hi, Nothing like that in Akka. Remote class loading has been done in the JVM before, but since it can easily allow a remote attacker full freedom to run any code they want on a server node it's not very common any more. -- Johan Akka Team On Fri, Jan 27, 2017 at 9:17 PM, Eugene Dzhurinsky <jdeve...@gmail.com> wrote: > Hi! > > Is there any way to deploy the code (jar files) along with starting the > new actors in the Akka cluster without the cluster restart? For example - I > have some "base" actors tier that provides the core logic for the data > flow. > > Now I want to create a solution, that will use these actors (using some > service discovery API) to perform certain tasks, and I will need to deploy > the new code to the nodes with certain roles, so it will be initialized - > and appropriate actors started and joined the cluster. > > I guess I can achieve something like that using some helper makeshift > shell-scripts that will download some JARs and spawn an additional JVMs, > however may be there's something I could borrow from Akka? > > Thanks > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/ > current/additional/faq.html > >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user > --- > You received this message because you are subscribed to the Google Groups > "Akka User List" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Declare Mailboxes Programmatically
The only way currently would be to programatically generate the config
entries for the mailboxes before starting the actor system.
--
Johan
Akka Team
On Sat, Jan 28, 2017 at 4:33 AM, Edoardo Vacchi <evac...@live.com> wrote:
> I know that a mailbox can be selected through
> Props.withDispatch("mailbox-id").
> Unfortunately this means "mailbox-id" needs to be a known mailbox config in
> the application.conf. Is there a way (I can't think of any right now) to
> describe the mailbox programmatically ?
>
> My idea would be to configure a mailbox via code (a DSL) instead of a
> config file. The reason I ask is that the system I'm working on configures
> most of the actor routing behavior in code, and I would like to provide a
> way to describe actor message dispatching (if needed) in a similar way.
>
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] streams: why pulls the concat combinator both sides eagerly and the merge combinator not
I agree that it would be good if it was documented, please open a ticket (and even better a ticket AND a PR adding docs) -- Johan Akka Team On Thu, Jan 19, 2017 at 3:01 AM, EXTERNAL Wachter Stefan (Keybird IT Consulting und Vetriebs GmbH, INST-ICM/BSV-BS) < external.stefan.wach...@bosch-si.com> wrote: > Hi, > > > > I stumbled accross the surprising behaviour that the concat combinator > uses so called „detached“ inputs that pull eagerly (without signalled > demand; effectively being a buffer of size 1). The comment indicates that > this is a means to avoid deadlocks in graphs with cycles. Yet, when > implementing graphs with cycles I usually use the merge combinator that > does not exhibit that behaviour. > > > > Additional proposal: Maybe the „eager pulling“ behaviour should be > documented in the API because it may be a surprising behaviour. (E.g. > expensive operations might happend without need) > > > > Mit freundlichen Grüßen / Best regards > > > > *Dr. Stefan Wachter* > > > > Bosch Software Innovations GmbH > > Communications (INST-ICM/BSV-BS) > > Stuttgarterstr. 130 > > 71332 Waiblingen > > GERMANY > > www.bosch-si.de > > www.blog.bosch-si.com > > > > Tel. +49 711 811 58 477 > > Fax +49 711 811 58 100 > > external.stefan.wach...@bosch-si.com <daniel.mau...@bosch-si.com> > > > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Varying result when running akka flow in parellel
I'd start with looking at how many elements your MinuteSource emits before
completing in the different cases.
--
Johan
Akka Team
Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM
Twitter: @akkateam
On Wed, Dec 14, 2016 at 11:10 PM, <subo...@gmail.com> wrote:
> HI,
>
> I'm working with a flow that downloads data, parses json and adds ids to a
> set (dedupe). It's working just fine however when I modify the flow to run
> in parallel, I get different results.
>
> Here's my graph:
>
> val graph: RunnableGraph[Future[HashSet[Long]]] =
> Source.fromGraph(new MinuteSource(firstMinuteYesterday,
> firstMinuteYesterday.plusDays(1)))
> .via(dsl(parallelize = 4))
> .toMat(Sink.fold(new HashSet[Long]())((accSet, set) => {
> accSet ++ set
> }))(Keep.right)
>
>
> val deduped: Set[Long] = Await.result(graph.run(), Duration.Inf)
>
> println(s"seq size is ${deduped.size} in ${new Duration(start, new
> DateTime()).toString}")
>
>
> The dsl looks like
>
> def dsl(parallelize: Int) = Flow.fromGraph(GraphDSL.create() { implicit
> builder =>
> import GraphDSL.Implicits._
>
> val dispatcher = builder.add(Balance[DateTime](parallelize))
> val merger = builder.add(Merge[Set[Long]](parallelize))
>
> for (i <- 0 to parallelize - 1) {
> dispatcher.out(i) ~> consumptionFlow.async ~> merger.in(i)
> }
>
> FlowShape(dispatcher.in, merger.out)
> })
>
>
> Here are the results for different parallelize values:
>
>
> // parallelize 1 -> seq size is 48560 in 175
> // parallelize 2 -> seq size is 48531 in 117
> // parallelize 4 -> seq size is 48481 in 107
>
>
> The resulting set size varies based on the parallelize number. What's
> interesting is the set size values are consistent, across runs. Does this
> make sense to anyone? Thanks!
>
> Andrew
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] At-least once processing with Akka Streams
On Sun, Nov 20, 2016 at 8:19 PM, Jacob Stultz <jacobastu...@gmail.com> wrote: > > 1: ensure that the message/offset or whatever other details are necessary > for later commit/ack are passed through the flow at every stage. > 2: implement the processing flow such that it is not aware of the > Kafka/SQS context, and then zip its output with the stream of messages for > acking. > I think this pretty much sums the options you have. There is a ticket in akka-stream-contrib about "focusing" on a part of an element, and an accompanying PR, that might be interesting: https://github.com/akka/akka-stream-contrib/issues/50 -- Johan Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam -- >>>>>>>>>> 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Understanding race conditions when creating actors
I think it refers to that you could by mistake pass/close over mutable internal state of the outer actor (ActorContext for example) to the inner actor which would result in race conditions. -- Johan Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam On Wed, Dec 14, 2016 at 11:13 AM, Dmb <bunadmit...@gmail.com> wrote: > As explained in the documentation > <http://doc.akka.io/docs/akka/2.4/scala/actors.html#Dangerous_Variants> it's > dangerous to create an actor within another actor using the following > variant: > val props7 = Props(new MyActor) > > The explanation for this is the following (as given in the documentation) > > This method is not recommended to be used within another actor because it >> encourages to close over the enclosing scope, resulting in non-serializable >> Props and possibly race conditions (breaking the actor encapsulation). > > > How does it cause *Props* to become non-serializable? And where is the > race condition? I cannot see how breaking encapsulation implies race > condition. > > Couldn't you expand this a little more? > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka cluster node - Marking self as Down
1. Is there any way a node which is part of Cluster can inform other
members to mark its status as Down.
Cluster().down(self) and then not terminating the actor system directly so
that the information has time to reach the other nodes:
Cluster cluster = Cluster.get(system);
cluster.down(cluster.selfAddress());
// or more gracefully
cluster.leave(cluster.selfAddress());
2. We have multiple Cluster Singleton instances in the cluster, and when
the node which is holding the instance goes unreachable, all the requests
to the Singleton instance going to Deadletter box or getting added to
Proxy's queue as configured
That's how singletons work. Be quicker with manual downing or tweak the SBR
(Split Brain Resolver
http://doc.akka.io/docs/akka/akka-commercial-addons-1.0.0/java/split-brain-resolver.html)
to down nodes quicker. Or avoid singletons if you want higher availability.
--
Johan
Akka Team
Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM
Twitter: @akkateam
On Sat, Nov 12, 2016 at 7:22 PM, Shyam kumar Akirala <
a.shyamkum...@gmail.com> wrote:
> 1. Is there any way a node which is part of Cluster can inform other
> members to mark its status as Down. We'are using Akka with Java, and we've
> registered a shutdown hook to gracefully shutdown Actor System in case of
> JVM going down. But seems after terminating Actor System on a node, that
> particular node's status is marked as Unreachable instead of Down.
>
> Runtime.getRuntime().addShutdownHook(
>
> new Thread() {
> @Override
> public void run() {
> if (actorSystem != null) {
> Future terminateFut = system.terminate();
> try {
> Await.result(terminateFut, Duration.Inf());
> } catch (Exception e) {
> LOGGER.error("Error shutting down Actor system", e);
> }
> }
> }
> }
> );
>
>
> 2. We have multiple Cluster Singleton instances in the cluster, and when
> the node which is holding the instance goes unreachable, all the requests
> to the Singleton instance going to Deadletter box or getting added to
> Proxy's queue as configured. In both cases there is message loss (in case
> of queue, if the limit 1 crosses), and downtime involved. Can you share
> any best practises to reduce the Cluster Singleton Instance downtime in
> case of node crash or network partition.
>
> Thanks,
> Shyam
>
>
>
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] priority , order events
Hi Adrian, Maybe Akka Streams would be a better fit for your problem. You can find the docs here: http://doc.akka.io/docs/akka/2.4/scala/stream/index.html -- Johan Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam On Sat, Nov 26, 2016 at 11:52 PM, <adrian.cou...@gmail.com> wrote: > I have heard a lot of good things about Akka. > > I have a stream of price , and I need to create two moving average of the > price. > > And finally I have a strategy that listens to these price, moving average > price 1 and moving average price 2 and makes some decisions. > > å > > I implemented this via actors but unfortunately I have a problem of order, > that a) I don’t know how to specify a timestamp or priority order to the > messages or events generated. > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Cluster Sharding custom routing
Hi Luis, Sharding allows for a custom ShardAllocationStrategy, which has two purposes: allocating new shards and taking decisions about rebalancing. Maybe this could cover what you want to do. (See API here: http://doc.akka.io/api/akka/2.4/#akka.cluster.sharding.ShardCoordinator$$ShardAllocationStrategy ) -- Johan Andrén Akka Team On Tue, Dec 6, 2016 at 7:13 PM, Luis Pedrosa <lpedros...@gmail.com> wrote: > Hello! > > > From what I understand, the cluster sharder GracefulShutdown > implementation, removes the node from the cluster. It also signals all the > remaining nodes to buffer any messages that target shards already allocated > on the given node. > > > I believe this is because the use case assumes that you will migrate the > entities accordingly before the node is removed from the cluster. > > > I have a different use case in mind. I would like to signal the sharder > that a node shouldn't be considered for any new allocation. However, It > should still be able to accept any message for pre-allocated shards. > > > The reason is that my entities are not ready for persistence (something I > would like to address at a later stage). Given that, migrating them is not > a desirable option. > > > I was wondering, is there a way to make this use case work with the > cluster sharder mechanics? > > Should I consider implementing a simple clustered routing solution that > satisfies our use case? > > Or is there another solution provided by the akka-cluster extensions that > I should be looking at? > > > In conclusion, on the ClusterSharder, a new shard is allocated to a node > that is currently up. I would like to control this allocation, for a given > node: > >- By still being able to accept messages for shards that are already >allocated in the node >- By not allowing any new shards to be allocated in the node > > Thank you in advance! > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] How to pass Ip/Port info of all runing services to the Akka node dynamically.
The typesafe config library allows you to override config settings by
setting the corresponding system property when you start the JVM, so
something like "java my.App -Dakka.remote.netty.tcp.hostname=example.com
-cp myjar.jar" but you can also use environment variables if defined in the
config.
See the typesafe config library docs for more info:
https://github.com/typesafehub/config#uses-of-substitutions
Note that you will likely have to define the internal ip and port
separately from the public ip and port as described here for it to work in
Docker:
http://doc.akka.io/docs/akka/2.4.14/scala/remoting.html#Remote_Configuration
--
Johan Andrén
Akka Team
On Fri, Dec 2, 2016 at 12:40 AM, Nitin Jadon <nitin.ja...@gmail.com> wrote:
> We are running Akka with Docker image and want to pass all the running
> Akka Docker's IP/Port info in case of addition on new Docker image.
>
> our application.conf file is defined as per below
>
>
>
> akka
> {
>
>
> actor
> {
> provider = "akka.cluster.ClusterActorRefProvider"
>
> debug {
> receive = on
> autoreceive = on
> lifecycle = on
> fsm = on
> event-stream = on
> unhandled = on
> router-misconfiguration = on
> }
> }
>
> remote
> {
> log-remote-lifecycle-events = off
> netty.tcp {
> hostname = "127.0.0.1"
> port = 2551
> }
> }
> }
>
>
> We have defined our Java code where we are loading config file using
>
>
> ActorSystem system = ActorSystem.create("ClusterSystem", config);
>
>
>
>
> As per current configuration, we are listening to TCP Port. please help to
> figure out how to pass new docker image IP/Port information to all the
> current running docker images.
>
>
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] How to deal with Futures in Sources/Flows
Source.unfoldAsync could be an existing stage worth looking closer at.
--
Johan Andrén
Akka Team
On Fri, Dec 9, 2016 at 7:31 PM, <subo...@gmail.com> wrote:
> I thought I needed to use a GraphStageLogic because I need to do the
> following:
>
> request url for page of data and push
> if response has next token, grab another page and push
> continue until no next token
> complete the stream
>
> I wasn't sure how to accomplish that logic in a Source.
>
> I figured out a solution by using getAsyncCallback
>
> val pushCallback = getAsyncCallback[Seq[String]] { seq =>
> push(out, seq)
> }
>
>
> Then on my future callback I call pushCallback.invoke(seq)
>
>
> It seems to work just fine but not sure that materializing a source in a
> GraphStageLogic is the best approach.
>
>
> If you have some example you can point me at I'm all ears!
>
>
> On Friday, December 9, 2016 at 1:27:31 AM UTC-7, drewhk wrote:
>>
>>
>>
>> On Thu, Dec 8, 2016 at 9:21 PM, <sub...@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> I'm creating a Source via GraphStageLogic which makes calls to another
>>> api, which happens to return a Source. However I'm unsure how to deal with
>>> Source/Futures in a GraphStageLogic. It seems that I want my shape to look
>>> like
>>>
>>> val shape: SourceShape[Seq[String]] = SourceShape(out)
>>>
>>>
>>> but I get back a Source[ByteString, NotUsed] which I can covert to
>>> Future[Seq[String]]
>>> via runWith(Sink.seq). So should I make my shape
>>>
>>> val shape: SourceShape[Future[Seq[String]]] = SourceShape(out)
>>>
>>>
>>> But then I need to materialize the Source inside the onPull(), which
>>> doesn't seem right. Is there a better way to handle this situation?
>>>
>>
>> You likely need to combine your stage with a flatMapConcat(). Remember
>> that you don't need to implement all your logic inside a GraphStage, you
>> can create Sources just by using existing combinators like mapAsync or
>> flatMapConcat. In fact, if you need some custom operation that is not
>> covered by the built-in combinators, it is usually a good idea to minimize
>> the custom part and combine it with the existing combinators to achieve
>> what you want.
>>
>>
>>>
>>> Thanks!
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c
>>> urrent/additional/faq.html
>>> >>>>>>>>>> Search the archives: https://groups.google.com/grou
>>> p/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 https://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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] akka-http proxy support
You can track the ticket to see if there is any progress or contribute such progress: https://github.com/akka/akka-http/issues/192 -- Johan Andrén Akka Team On Fri, Dec 9, 2016 at 4:45 PM, Álvaro Reguly <ubu...@reguly.com> wrote: > Any news here? This a big deal for us as Akka is locked behind the > corporate firewalls. > > thanks > > Alvaro Reguly > > On Tuesday, March 8, 2016 at 11:47:31 AM UTC-5, Konrad Malawski wrote: >> >> I hope to address this issue eventually, but we'll see how priorities >> work out. >> Keep an eye on the linked ticket for updates! >> >> -- >> Cheers, >> Konrad 'ktoso’ Malawski >> <http://akka.io>Akka <http://akka.io> @ Lightbend <http://typesafe.com> >> <http://lightbend.com> >> >> On 8 March 2016 at 17:46:29, Michi (michael...@physik.tu-muenchen.de) >> wrote: >> >> Hi Konrad, >> >> thanks for the quick answer. Fortunately it is only a small part of our >> code that needs that. I will use Apache HttpClient for now, which does >> everything we need. >> >> Best regards, >> Michael >> >> On Tuesday, March 8, 2016 at 5:37:01 PM UTC+1, Konrad Malawski wrote: >>> >>> Correct, HTTPS Proxies are not supported yet, see ticket: >>> https://github.com/akka/akka/issues/16153 >>> Sadly we're not sure when we'll get the chance to work on this yet. >>> >>> For the time being you may want to try Play's WS library, though I'm not >>> sure if they support this feature hm... >>> (They're backed by the Ning Async Http Client). >>> >>> -- >>> Cheers, >>> Konrad 'ktoso’ Malawski >>> <http://akka.io>Akka <http://akka.io> @ Lightbend <http://typesafe.com> >>> <http://lightbend.com> >>> >>> On 8 March 2016 at 15:31:28, Michi (michael...@physik.tu-muenchen.de) >>> wrote: >>> >>> Hi, >>> >>> we need to access a REST interface using HTTPS which is behind a proxy. >>> The akka-http documentation states that proxy support for HTTPS connections >>> is currently unsupported. Is there any way to use a HTTPS proxy with >>> akka-http? If not, can someone recommend a good (Java) https client library >>> that supports https proxies for accessing REST interfaces? >>> >>> Thanks, >>> Michael >>> -- >>> >>>>>>>>>> Read the docs: http://akka.io/docs/ >>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>> urrent/additional/faq.html >>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>> p/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 https://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/c >> urrent/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 https://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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka Http routes passing objects to sub routes
Hi Elliot,
look at the directives and custom directives parts of the docs, that should
help you get going:
http://doc.akka.io/docs/akka-http/current/scala/http/routing-dsl/directives/index.html
--
Johan
Akka Team
On Thu, Nov 24, 2016 at 1:22 PM, Elliot Laurendeau <tap...@gmail.com> wrote:
> Hi,
>
> I'm currently working on a Scala akka http projet and i implemented Async
> auth.
>
> Ex of my structure (these isn't my real project, its just easier to
> picture it):
>
> HttpRouter : Choose the arborescence of the URL
>
> AuthController: has a user register and auth callback from an external
> website. URL : auth/action
> CarController: has a car search function and bike register function
> BikeController: has a bike search and bike register function
>
> The HTTP Router looks like the following:
>
> val routes = {
> authRoutes ~ pathPrefix("secured") {
> authenticateBasicAsync(realm = "secure site", userAuthenticator) {
> user =>
> carRoutes ~ bikeRoutes
> }
> }
> }
>
>
> I'm adding the login security in the HttpRouter rather than in each seperate
> module.
>
>
> My question:
>
> Within my carRoutes and my bikeRoutes, i have a register function to add the
> object to my list of sold items.
>
> When a request is calling that /add method, i want to pass a reference of my
> authenticated user to mark him as the seller.
>
>
> Is it possible to pass my user object (see below) to my child/sub routes?
>
>
> authenticateBasicAsync(realm = "secure site", userAuthenticator) {
> *user *=>
> carRoutes(i want my user to be passed along here) ~ bikeRoutes(i
> want my user to be passed along here)
> }
>
>
> For testing purposes, I duplicated the authenticateBasicAsync line in my
> child routes and it works, but i'm calling my validate user twice. It works
> but i don't like this method.
>
>
> Anyone know a better solution?
>
>
> Thanks
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] When to use PersistentActor vs PersistentFSM?
It's pretty much a question about taste, if you like a PersistentFSM better
than a straight up Persistent Actor, go with it!
An actor that never transitions to a different behaviour would probably be
a bad fit though.
--
Johan
Akka Team
On Mon, Nov 21, 2016 at 1:57 AM, <t.dettr...@uq.edu.au> wrote:
> Is there a good rule-of-thumb for when to use PersistentActor vs
> PersistentFSM? I've been making heavy use of PersistentFSM for my aggregate
> entities, but I'm worried I may be over-using it as I rarely find myself
> using PersistentActor.
>
> My persistent entities almost always start in a "Not Created" state before
> being initialised by a creation message, so as a result every one of them
> will handle messages differently depending on the "state" they're in.
> However, as I've used PersistentFSM more, I've come to realise that
> PersistentFSM states are quite disconnected from persisted domain events,
> and so I'm actually storing the entity state twice (potentially leading to
> conflict).
>
> In terms of syntax, I find PersistentFSM's "when(state) {}" to be much
> cleaner. If I should really be using PersistentFSM sparingly, how can I get
> a clearer transition between behaviours in PersistentActor?
>
> Thanks,
>
> Tim
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Resumable Projection
Hi Richard, This is not handled by streams itself as it is a feature that Sources must support. I think the most common solution people use for this is Kafka. There is a Streams connector for Kafka: https://github.com/akka/reactive-kafka -Endre On Tue, Nov 15, 2016 at 3:04 AM, Richard Rodseth <rrods...@gmail.com> wrote: > Any good examples out there of resumable projections driving non-trivial > streams? > > I'm guessing I will have to keep the stream 1-1 and pass the offset all > the way downstream so I can save it at the end? > > > -- > >>>>>>>>>> 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 https://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam -- >>>>>>>>>> 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] `ActorSystem.whenTerminated` hangs indefinitely if called from an `registerOnTermination` callback
Hi Chris,
On Tue, Nov 15, 2016 at 1:46 PM, oxbow_lakes <oxbowla...@gmail.com> wrote:
> The following code:
>
> Welcome to Scala 2.12.0 (Java HotSpot(TM) 64-Bit Server VM, Java
> 1.8.0_112).
> Type in expressions for evaluation. Or try :help.
>
>
> scala> import akka.actor._; import scala.concurrent._; import duration._
> import akka.actor._
> import scala.concurrent._
> import duration._
>
> scala> :paste
> // Entering paste mode (ctrl-D to finish)
>
> val s = ActorSystem("test")
> def onTerm = {
> println("Waiting...")
> Await.result(s.whenTerminated, Duration.Inf)
> println("Fin!")
> }
> s.registerOnTermination(onTerm)
>
> // Exiting paste mode, now interpreting.
>
> s: akka.actor.ActorSystem = akka://test
> onTerm: Unit
>
> scala> s.terminate
> res1: scala.concurrent.Future[akka.actor.Terminated] = Future( completed>)
>
> scala> Waiting...
>
>
> At this point the future doesn't terminate. Looking at the documentation
> of *whenTerminated*; "Returns a Future which will be completed after the
> ActorSystem has been terminated and termination hooks have been executed".
> It's probably a good idea to indicate on the *registerOnTermination*
> method that it's dangerous to wait on the whenTerminated future from the
> callback.
>
Can you please file a ticket, or even better a PR to fix the relevant docs?
-Endre
>
>
> Chris
>
> --
> >>>>>>>>>> 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 https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Akka Team
Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM
Twitter: @akkateam
--
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Re: Akka persistence query examples
One option would be to use a persistent actor as your read side as well,
store the last offset you have seen and stream from after that offset when
the actor has recovered completely, this way you can also provide snapshots
(and possible delete the actor to replay the entire history into a view
after adding functionality for example).
Another could be to persist the read side in some other way, a relational
database for example and do queries against that.
--
Johan
Akka Team
On Wed, Nov 9, 2016 at 5:34 PM, Juan Carlos Roig <jcr...@gmail.com> wrote:
> Hi everyone. Just waking up this thread (once again) with a related
> challenge/question
>
>
> I'm used to use PersistentView for my read model before they got
> deprecated.
>
> I have a CQRS architecture where both write and read sides are cluster
> singleton. In write side I use persistent actors to validate and store the
> events and then those events get replayed to the persistent views on the
> other sharded cluster. Those persistent views process the events and keep
> the read model in memory, so queries are sent to them and they respond
> accordingly. This is very easy to do as akka persistence takes care of
> actors initialisation both at persistent actors and views so that in case
> any actor gets restarted or rebalanced by the cluster it will be started
> and initialized recovering its previous state by using snapshots and events
> replay before processing any new incoming message.
>
> I'm trying to do the same using persistent query but I'm having some
> problems.
>
> I'm creating the persistent query inside the actor view and send the
> events to itself.
>
> Example:
>
> class DefaultAccountView extends Actor with ActorLogging {
> implicit val mat = ActorMaterializer()(context.system)
> implicit val ec = context.dispatcher
>
> val aggregateId = context.self.path.name
>
> def persistenceId: String = s"AccountAggregate$aggregateId"
>
> // Configure Persistence Query
> val queries = PersistenceQuery(context.system).readJournalFor[
> LeveldbReadJournal](LeveldbReadJournal.Identifier)
>
> val events = queries.eventsByPersistenceId(persistenceId, 0L,
> Long.MaxValue)
> .map(eventEnvelope => eventEnvelope.event)
> .to(Sink.actorRef(self, StreamCompleted))
> .run()
>
>
> def handleAccountQueries: Recevice = {
> case Query1 =>
> // Process query 1
> case Query 2 =>
> // Process query 2
> }
>
>
> def handleEvents: Receive = {
> case Event1 =>
> // process event 1 update state
>
> case Event2 =>
> // process event 2 update state
>
> }
>
>
> override def receive: Receive = handleAccountQueries orElse handleEvents
> }
>
> In that case the stream will be created every time the actor gets started
> and will replay all events. The problem is that being an actor belonging to
> a sharded cluster it will be started the first time it gets a message, that
> is, a query. What happens is the first message always gets processed before
> all events are replayed and hence it fails to give correct response as the
> state is not initialised, in other words, there's no initialisation
> lyfecycle like PersistentView has. I tried to use a custom object to signal
> the end of the stream (StreamCompleted) but obviously it is never sent as
> the stream never finishes (it keeps opened sending events generated in real
> time, which is what we need to keep the view updated). On a side note, it
> will be very nice to be able to easily use snapshots here too as replay all
> events every time the view starts might take a lot of time for high load
> actors.
>
> Is there any way to do this overcoming these problems?
>
> Shall I continue using PersistentView instead although they're deprecated?
>
> How are you guys propose to implement this use case if persistent views
> are deprecated?
>
> Regards,
>
> J. Carlos
>
> --
> >>>>>>>>>> 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 https://groups.google.com/group/akka-user.
> For more options, visit https
Re: [akka-user] StreamSupervisor name
AbruptTerminationException happens when the actor running the stream (the
"graph interpreter") is terminated without the stream first completing or
failing, for example if terminating an actor system with a running stream
in it.
The paths of the supervisors are pretty much an implementation detail, but
perhaps you can provide some addition to the string "end-of-lookup" to give
you context in the log entry?
--
Johan
Akka Team
On Wed, Nov 9, 2016 at 6:33 PM, Richard Rodseth <rrods...@gmail.com> wrote:
> I'm trying to write two ScalaTest tests to compare two approaches to
> driving a Flow that sends out emails. In one approach the flow is sourced
> using eventsByTag from akka-persistence-query. In the other, I uses
> SourceQueue.offer after receiving a message via AtLeastOnceDelivery. The
> test ends with a check of mock JavaMail mailbox size.
>
> eventually {
>
> mailBoxSize("ema...@foo.com") should be(expectedEmails)
>
> }
>
> eventually {
>
> mailBoxSize("ema...@foo.com") should be(expectedEmails)
>
> }
>
> The test that uses eventsByTag to drive the flow succeeds with 1
> events. The one that uses SourceQueue.offer gets stuck at 11. In the flow
> below, the println appears 12 times but the log("end-of-lookup") only
> appears 11 times.
>
> def flowLookupEmails()(implicit lookup: EmailsByDefinition, ec:
> ExecutionContext): Flow[AlarmOccurrence, AlarmOccurrenceWithRecipients,
> NotUsed] = {
> val parallelism = 5
> val result = Flow[AlarmOccurrence].mapAsync(parallelism) { occurrence
> =>
> val emailsF: Future[List[String]] = lookup.recipientsForDefinition(
> occurrence.definitionId)
> val lookupResult: Future[AlarmOccurrenceWithRecipients] =
> emailsF.map { emails =>
> println(s"# Found emails $emails")
> AlarmOccurrenceWithRecipients(occurrence, emails)
> }
>
> lookupResult
> }
> .log("end-of-lookup").withAttributes(Attributes.logLevels(onElement =
> Logging.InfoLevel))
> .named("lookup-emails")
> result
> }
>
> When I turn on DEBUG logging, I get a lot of noise, but do see
>
> [DEBUG] [11/09/2016 08:58:02.827] [OccurrencesIntegrationSpec-
> akka.actor.default-dispatcher-12] [akka://OccurrencesIntegrationSpec/
> user/StreamSupervisor-3/flow-25-0-unknown-operation] stopped
>
> [ERROR] [11/09/2016 08:59:02.455] [OccurrencesIntegrationSpec-
> akka.actor.default-dispatcher-21] [akka.stream.Log(akka://
> OccurrencesIntegrationSpec/user/StreamSupervisor-0)] [end-of-lookup]
> Upstream failed. (akka.stream.AbruptTerminationException)
> Any ideas? What are some good techniques to find the cause of the
> AbruptTerminationException, or name the StreamSupervisors so I know which
> stream they are associated with?
>
> Thanks.
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-remote docker<->docker works fine with netty.tcp, but fails with artery - ideas?
Ah, sorry, didn't notice it in the command line. Can you verify with some other tool that you can communicate using udp between the nodes? (using netcat for example) Set logs to debug to see if there is anything helpful in there. Additionally artery contains a "flight recorder" which you can enable using the setting akka.remote.artery.advanced.flight-recorder.enabled, it will write artery related events to a binary logfile, which you can then feed to akka.remote.artery.FlightRecorderDump in akka-remote to get human readable output, it may also contain some hints about what/why/how it is not working. -- Johan Akka Team On Wed, Nov 9, 2016 at 7:14 PM, <johan.kjolhede.bonniern...@gmail.com> wrote: > Yes - i Did. Please see my original post for details. You will see that I > use the udp parameter for my docker port forwarding. Also please note that > I've tested artery successfully in docker as long as one of the > applications is running outside docker. However the problem occurs only > when both producer and consumer run in containers. See the test results > table and error log in my previous message for more details. > > /Johan > > -- > >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] Akka-remote docker<->docker works fine with netty.tcp, but fails with artery - ideas?
Hi Johan (good name!),
Artery/Aeron uses UDP, while the old remoting is TCP-based, did you
configure your docker nodes with udp port mappings?
--
Johan
Akka Team
On Wed, Nov 9, 2016 at 6:11 PM, <johan.kjolhede.bonniern...@gmail.com>
wrote:
> I'm pretty new to Akka and am currently trying out Akka's new Artery
> backend for akka-remoting (http://doc.akka.io/docs/akka/
> 2.4/scala/remoting-artery.html) while also testing the regular netty.tcp
> transport (http://doc.akka.io/docs/akka/2.4/scala/remoting.html).
>
> It looks fairly straight forward and I have two test applications running:
> * A producer application which sends a message to the consumer every 3
> seconds, and prints some stuff when the reply comes back
> * A consumer application which prints a message when it receives one, and
> then sends back a reply to sender()
>
> However I cannot get the communication to work when running both my
> applications with artery inside docker containers on the same host - it
> perfectly when I use netty.tcp though!.
> Below are the results I achieved:
>
> Results using artery
> Producer Consumer Result
> -
> IntelliJ Dockerok
> DockerIntelliJ ok
> IntelliJ IntelliJ ok
> DockerDockerfail (no message reaches the consumer, and the
> producer eventually prints handshake timeouts after 20 seconds/default
> timeout in akka-remote-artery - the same error message I get when not
> running any Consumer at all)
>
> Results using netty.tcp
> Producer Consumer Result
> -
> IntelliJ Dockerok
> DockerIntelliJ ok
> IntelliJ IntelliJ ok
> DockerDockerok
>
> As you see above, I got all combinations to work except when using artery
> to communicate between two local docker containers.
> The akka configurations for the failing artery case are as follows
>
> Consumer conf (prettified output of typesafe Config object logged):
> {
> "akka": {
> "actor": {
> "provider": "remote"
> },
> "remote": {
> "artery": {
> "bind": {
> "hostname": "0.0.0.0",
> "port": 12345
> },
> "canonical": {
> "hostname": "192.168.99.100",
> "port": 12345
> },
> "enabled": "on"
> }
> }
> }
> }
>
> Producer conf (prettified output of typesafe Config object logged):
> {
> "akka": {
> "actor": {
> "provider": "remote"
> },
> "remote": {
> "artery": {
> "bind": {
> "hostname": "0.0.0.0",
> "port": 12346
> },
> "canonical": {
> "hostname": "192.168.99.100",
> "port": 12346
> },
> "enabled": "on"
> }
> }
> }
> }
>
>
> 192.168.99.100 is the IP of my docker machine executing both containers.
> The equivalent netty.tcp config works fine (as seen in the results table
> earlier in this message)
>
> *However using artery I just get zero activity on the consumer, and
> eventually the producer starts logging errors:*
> [ERROR] [11/09/2016 16:24:17.201]
> [producer-system-akka.remote.default-remote-dispatcher-6]
> [akka.remote.artery.Association(akka://producer-system)] Outbound message
> stream to [akka://consumer-system@192.168.99.100:12345] failed.
> Restarting it. Handshake with [akka://consumer-system@192.168.99.100:12345]
> did not complete within 2 ms
> (akka.remote.artery.OutboundHandshake$HandshakeTimeoutException)
> after about 20 seconds, indicating what I assume is a timeout.
>
> When I use netty.tcp it works just fine between the containers.
> What is also strange is that it works with artery when I move one of the
> applications outside of a docker container. Both
>
> When running each application outside docker (e.g. from within intelliJ
> during development) they can communicate with each other fine, but as soon
> as I try to make them talk to each other with artery from docker container
> to docker container, no messages get through to the consumer
> Each application can be configured (cmd line args) to communicate with
> either netty.tcp or arter
Re: [akka-user] Website Correction / http://akka.io/docs/
For future reference the docs page on akka.io referred to (and the rest of the akka.io site) is in the akka.github.com repo, here: https://github.com/akka/akka.github.com/blob/master/_includes/download.html The actual doc contents are in the akka and akka-http repos here: https://github.com/akka/akka/tree/master/akka-docs https://github.com/akka/akka-http/tree/master/docs -- Johan Akka Team On Tue, Nov 8, 2016 at 6:35 AM, Gavin Baumanis <gavinbauma...@gmail.com> wrote: > I would have happily created a PR but i couldnt find the documentation in > the github repo. > > If you wouldn't mind pointing me in the right direction I will gladly > contribute to doc corrections as I find issues/ have suggestions. > > On 7 Nov 2016 3:56 PM, "Konrad Malawski" <konrad.malaw...@lightbend.com> > wrote: > >> Thanks for the PR Ivan, it's so nice to see someone actively participate >> in fixing things :-) >> Yep, it's as Ivan said, small typo - now fixed. >> >> Cheers! >> >> -- >> Konrad `ktoso` Malawski >> Akka <http://akka.io> @ Lightbend <http://lightbend.com> >> >> On 5 November 2016 at 11:03:36, Ivan Vyshnevskyi (sain...@gmail.com) >> wrote: >> >> Hi, >> >> Thanks for reporting! >> >> I think that the issue is that `akka-http-testkit` wasn't moved to the >> separate list with the rest of `akka-http` dependencies that are on their >> own versioning scheme now (2.4.11 at the moment, 3.0/10.0 soon-ish). >> I've opened a pull request [1] to fix that. >> >> But I suppose this won't be addressed until Monday. >> >> [1]: https://github.com/akka/akka.github.com/pull/348 >> >> On 5 November 2016 at 18:32, Gavin Baumanis <gavinbauma...@gmail.com> >> wrote: >> >>> Hi there everyone, >>> >>> Just a quick-fix for the information displayed at; >>> http://akka.io/docs/ >>> >>> "com.typesafe.akka" %% "akka-http-testkit" % "2.4.12" >>> :s/2.4.12/2.4.11/g >>> -- >>> >>>>>>>>>> Read the docs: http://akka.io/docs/ >>> >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c >>> urrent/additional/faq.html >>> >>>>>>>>>> Search the archives: https://groups.google.com/grou >>> p/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 https://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/c >> urrent/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 https://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/c >> urrent/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/to >> pic/akka-user/zgzi-svVY70/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 https://groups.google.com/group/akka-user. >> For more options, visit https://groups.google.com/d/optout. >> > -- > >>>>>>>>>>
Re: [akka-user] How can I check that MemberUp(m:Member) does not equal self actor?
Hi,
Cluster.selfAddress will give you the address of the actor system, and then
you can compare it to Member.address
--
Johan
Akka Team
On Mon, Nov 7, 2016 at 6:18 AM, Unknown Unknown <zx80l...@gmail.com> wrote:
> There is the following actor which belongs to cluster and is subscribed to
> cluster events:
>
>
> class Worker extends Actor {
>val cluster = Cluster(context.system)
>var router = Router(BroadcastRoutingLogic(), Vector.empty[
> ActorRefRoutee])
>
>override def preStart(): Unit =
> cluster.subscribe(self, classOf[MemberEvent])
>
>override def postStop(): Unit = cluster.unsubscribe(self)
>
>
>def receive = {
> ...
> case MemberUp(m: Member) =>
> val a = context.actorSelection(
>RootActorPath(m.address) / "user" / "worker")
>
> val isNotSelf:Boolean = ??? //TODO how to check m != self
> if(isNotSelf)
>router = router.removeRoutee(a).addRoutee(a)
>...
>}
> }
>
>
>
>
> In this case the Worker actor belongs to cluster, therefore it add to
> router itself.
>
> How can I check that `MemberUp(m:Member)` does not equal `self` actor (see
> TODO in above code)?
>
> I use akka 2.4.12.
>
> The same question at http://stackoverflow.com/questions/40453413
>
> --
> >>>>>>>>>> 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 https://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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
Re: [akka-user] [akka-cluster]How is the Terminate message issued to its watcher if wachee is closed during the unreachable period?
Hi, On Mon, Oct 24, 2016 at 6:42 PM, <fh...@mails.tsinghua.edu.cn> wrote: > Hi guys, > I'm recently studying the source code of akka-cluster and there's > something that I really cannot understand. > Actually, I was trying to reproduce what akka-cluster document said "If > system messages cannot be delivered to a node it will be quarantined and > then it cannot come back from unreachable". So I've down these steps: > >1. Create Actor A on node1, Actor B on node2; >2. Let A watch B; >3. Cut the wire between node1 and node2; >4. Let B send a user message, namely UMSG, to A; >5. Close B, and I've seen that Actor B tried to send a message with >type DeathWatchNotification; >6. Reconnect node1 and node2, and the cluster "revived" with both >nodes had changed state of each other from UNREACHABLE to UP; >7. ActorA immediately received Terminated message from ActorB, while >it didn't receive the user message (UMSG, see Step4) from B. > > Yes, this is exactly how it should work. User messages have no guaranteed delivery, and hence they are simply dropped during connection problems. System messages (like Watch and DeathWatchNotification) on the other hand are buffered until they are acknowledged by remote system. Once the connection comes back, they are attempted to be redelivered (as you see at step 7). So what happens 2. Watch sent from A to B and acknowledged as received by B 4. UMSG is dropped as connection is severed 5. DeathWatchNotification is attempted to be sent by B, but ack does not arrive 7. above DWN is resent, processed by actor A and acknowledged the reception of the message > Here is my questions > >1. DeathWatchNotification is also a kind of system message, and, which >might not be true, it failed issuing to its watcher. According to the >document mentioned above, the state of the node containing the watcher >actor should have been put to DOWN, but it didn't. Why is this? Where can I >find the code related to this behavior? > > This is a misunderstanding. System message delivery is only considered failed once all the buffer space designated for containing system messages for redelivery are depleted. Otherwise these messages are kept around for a long time. If a remote system is not responding for a long time then it is considered dead. This can happen in two ways: - In ordinary remoting DeathWatch is sending heartbeats between systems of the watcher and watchee. If a limit is reached (watch-failure-detector) then the remote system is considered dead, is quarantined and all system messages are dropped, and watched actors declared dead. - if cluster is used (your case), cluster nodes monitor each other via heartbeats, and if there are missed heartbeats from a system, members designate that as unreachable. This is not the final state and recoverable, it is only if a system is DOWN-ed when it is deemed dead, and watcher actors are informed of this fact. DOWNing can be manual, programmatic, or by using auto-downing: http://doc.akka.io/docs/akka/2.4/scala/cluster-usage.html#Auto-downing__DO_NOT_USE_ > >1. Who is the real actor that sends this Terminated system message? > > It is converted to a user message from the DWN system message. > >1. I've noted that in akka-cluster both system message and user >message are delivered in a same way using akka-remote. So why is UMSG >dropped but Terminated received? > > See above. > > Thanks for helping in advance > Hangyu > > -- > >>>>>>>>>> Read the docs: http://akka.io/docs/ > >>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/c > urrent/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 https://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Akka Team Lightbend <http://www.lightbend.com/> - Reactive apps on the JVM Twitter: @akkateam -- >>>>>>>>>> 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 https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
