[akka-user] flatMapMerge and SupervisionStrategy

[Richard Rodseth]

I'm using akka-kafka with committablePartitionedSource, meaning I have a
source of (partition, innersource) tuples.

I'm currently using the flatMapMerge recipe from

https://doc.akka.io/docs/akka-stream-kafka/current/consumer.html#source-per-partition

val done = Consumer.committablePartitionedSource(consumerSettings,
Subscriptions.topics("topic1"))
  .flatMapMerge(maxPartitions, _._2)
  .via(business)
  .batch(max = 20, first =>
CommittableOffsetBatch.empty.updated(first.committableOffset)) {
(batch, elem) =>
batch.updated(elem.committableOffset)
  }
  .mapAsync(3)(_.commitScaladsl())
  .runWith(Sink.ignore)

If I have a default supervision strategy which resumes, set at the
materializer level, but want to stop the stream (which in my case restarts
a host actor with backoff) at the source level, do I need to do a
withAttributes(attributesForStoppingDecider) at both the outer source and
inner source leves?
i.e.

val done = Consumer.committablePartitionedSource(consumerSettings,
Subscriptions.topics("topic1"))
  .withAttributes(attributesForStoppingDecider).flatMapMerge(maxPartitions,
_._2.withAttributes(attributesForStoppingDecider))
.via(business)

or can I do it after the flatMapMerge? Question motivated by this open
ticket which I'm not sure I fully understand yet.

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

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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] Short lived persistent actors and deletion

[Justin du coeur]

On Fri, Oct 6, 2017 at 10:40 AM, Ole Hjalmar Herje 
wrote:

> Ok. Do you know if it is it supported by other journal plugins? And is
> leveldb stil not considered "production ready"?
>

LevelDB wasn't really *designed* to ever be "production ready", as far as I
know.  It doesn't scale (by design), and it's reportedly prone to data
corruption when things go wrong.

Personally, given the existence of ccm , I
don't think it's worth bothering with LevelDB even for development.  There
are better options...

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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-user] Re: Using Actors to send data to websockets

[Rob Crawford]

Have you read this:

https://doc.akka.io/docs/akka-http/current/java/http/server-side/websocket-support.html

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

[akka-user] Using Actors to send data to websockets

[Anmol Singh Jaggi]

I have been experimenting with Akka websockets and was able to send some 
static data to a websocket using the following code:

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;


import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.http.javadsl.ConnectHttp;
import akka.http.javadsl.Http;
import akka.http.javadsl.ServerBinding;
import akka.http.javadsl.model.HttpRequest;
import akka.http.javadsl.model.HttpResponse;
import akka.http.javadsl.model.ws.Message;
import akka.http.javadsl.model.ws.WebSocket;
import akka.japi.Function;
import akka.stream.ActorMaterializer;
import akka.stream.Materializer;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;


public class Server {


  public static HttpResponse handleRequest(HttpRequest request) {
System.out.println("Handling request to " + request.getUri());
if (request.getUri().path().equals("/greeter")) {
  final Flow greeterFlow = greeterHello
();
  return WebSocket.handleWebSocketRequestWith(request, greeterFlow);
} else {
  return HttpResponse.create().withStatus(404);
}
  }


  public static void main(String[] args) throws Exception {
ActorSystem system = ActorSystem.create();


try {
  final Materializer materializer = ActorMaterializer.create(system
);


  final Function handler = request -> 
handleRequest(request);
  CompletionStage serverBindingFuture = Http.get(
system).bindAndHandleSync(handler,
  ConnectHttp.toHost("localhost", 8080), materializer);


  // will throw if binding fails
  serverBindingFuture.toCompletableFuture().get(1, TimeUnit.SECONDS
);
  System.out.println("Press ENTER to stop.");
  new BufferedReader(new InputStreamReader(System.in)).readLine();
} finally {
  system.terminate();
}
  }


  public static Flow greeterHello() {
return Flow.fromSinkAndSource(Sink.ignore(),
Source.single(new akka.http.scaladsl.model.ws.TextMessage.Strict
("Hello!")));
  }
}


I now want to send some data to a websocket from an Actor, something like 
this:  

import akka.actor.ActorRef;
import akka.actor.UntypedActor;


public class PushActor extends UntypedActor {
  @Override
  public void onReceive(Object message) {
if (message instanceof String) {
  String statusChangeMessage = (String) message;
  // How to push this message to a socket ??
} else {
  System.out.println(String.format("'%s':\nReceived unknown message 
'%s'!", selfActorPath, message));
}
  }
  
}

I am unable to find any example of how to do this online.
Could somebody please guide me on this?

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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] Short lived persistent actors and deletion

[Patrik Nordwall]

That is currently not supported but you could contribute that feature to
the leveldb plugin. We could use Long.MaxValue as the signal to remove
everything.

/Patrik
fre 6 okt. 2017 kl. 10:44 skrev Ole Hjalmar Herje :

> Hi, is it possible to delete persistent actors completely from the
> persistent storage?
> I have many short lived persistent actors with auto-generated
> unique persistenceId where I perform a deleteMessages(lastSequenceNr())
> when they are done with their work. If the application restarts I perform
> recovery by using LeveldbReadJournal and a currentPersistenceIds() folowed
> by a flatMapMerge on currentEventsByPersistenceId so i only get the ids
> that actually have messages, i.e is not done. But the result of
> currentPersistenceIds query will grow over time since the persistenceId
> mapping is still in storage even if all messages are deleted. (This issue
> was originally about the same?, but ended up with something else:
> https://github.com/akka/akka/issues/21677)
> So my question is if there is a way to delete the persistenceid from
> storage as part/instead of deleteMessages(lastSequenceNr())?
>
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to 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.

[akka-user] Short lived persistent actors and deletion

[Ole Hjalmar Herje]

Hi, is it possible to delete persistent actors completely from the 
persistent storage? 
I have many short lived persistent actors with auto-generated 
unique persistenceId where I perform a deleteMessages(lastSequenceNr()) 
when they are done with their work. If the application restarts I perform 
recovery by using LeveldbReadJournal and a currentPersistenceIds() folowed 
by a flatMapMerge on currentEventsByPersistenceId so i only get the ids 
that actually have messages, i.e is not done. But the result of 
currentPersistenceIds query will grow over time since the persistenceId 
mapping is still in storage even if all messages are deleted. (This issue 
was originally about the same?, but ended up with something 
else: https://github.com/akka/akka/issues/21677)
So my question is if there is a way to delete the persistenceid from 
storage as part/instead of deleteMessages(lastSequenceNr())?

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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-user] remote actor actorSystem.actorSelection("Actorpath").resolveOne(20.seconds) times out after several request processed.

[Kotresh Chikkabidari]

Hi All,

We are using Scala and Akka(version 2.4.0) in out application. We used 
remote actors and after processing several requests, actor resolve is 
timing out. 
here is the snippet of actor resolve. And we are running our application in 
Virtual machines.

  val a1 = Address("akka.tcp", "actorservice", localHostAddress, "9015")
  val a2 = a1.toString + "/user/supervisor/coordinator"

try {
val r = system.actorSelection(a2).resolveOne(20.seconds)
Right(Await.result(r, 20.seconds))
} catch {
  case e: TimeoutException => log.info(msg + e.getMessage); throw e
}

Any of you faced the similar problem and found any solution for the same? 
Thanks in advance.

Regards,
Kotresh

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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-user] Re: Akka with Java: remote actor deploy with SpringExtension

[Deep Gandhi]

I am also facing the same issue. Anything you got on this?

On Thursday, 13 April 2017 02:10:00 UTC+5:30, Alexander Lukyanchikov wrote:
>
> Hi,
>
> We are using Akka with Java API, integrated with Spring Framework (each 
> actor is Prototype scope Spring bean)
> There is no problem to create local actors, using 
> Akka IndirectActorProducer (we use approach, described here 
> )
>
> But now we are trying to use Cluster aware router with Pool of Remote 
> Deployed Routees, and there is a problem - as far as I understand Akka 
> sends serialized Props to the remote node when we define a router like this:
>
> @Bean
> public ActorRef processorRouter() {
> return system.actorOf(
> new ClusterRouterPool(new RoundRobinPool(100),
> new ClusterRouterPoolSettings(1000, 100,
> true, 
> "processor")).props(SpringExtension.SpringExtProvider.get(system).props(ProcessorActor.class)));
> }
>
>
> Here is a serialization exception:
>
> akka.remote.MessageSerializer$SerializationException: Failed to serialize 
> remote message [class akka.remote.DaemonMsgCreate] using serializer [class 
> akka.remote.serialization.DaemonMsgCreateSerializer].
> at akka.remote.MessageSerializer$.serialize(MessageSerializer.scala:61) 
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at 
> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:895)
>  
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at 
> akka.remote.EndpointWriter$$anonfun$serializeMessage$1.apply(Endpoint.scala:895)
>  
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58) 
> ~[scala-library-2.11.8.jar!/:na]
> at akka.remote.EndpointWriter.serializeMessage(Endpoint.scala:894) 
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at akka.remote.EndpointWriter.writeSend(Endpoint.scala:786) 
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at akka.remote.EndpointWriter$$anonfun$4.applyOrElse(Endpoint.scala:761) 
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at akka.actor.Actor$class.aroundReceive(Actor.scala:497) 
> ~[akka-actor_2.11-2.4.17.jar!/:na]
> at akka.remote.EndpointActor.aroundReceive(Endpoint.scala:452) 
> ~[akka-remote_2.11-2.4.17.jar!/:na]
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526) 
> [akka-actor_2.11-2.4.17.jar!/:na]
> at akka.actor.ActorCell.invoke(ActorCell.scala:495) 
> [akka-actor_2.11-2.4.17.jar!/:na]
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257) 
> [akka-actor_2.11-2.4.17.jar!/:na]
> at akka.dispatch.Mailbox.run(Mailbox.scala:224) 
> [akka-actor_2.11-2.4.17.jar!/:na]
> at akka.dispatch.Mailbox.exec(Mailbox.scala:234) 
> [akka-actor_2.11-2.4.17.jar!/:na]
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 
> [scala-library-2.11.8.jar!/:na]
> at 
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
>  
> [scala-library-2.11.8.jar!/:na]
> at 
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 
> [scala-library-2.11.8.jar!/:na]
> at 
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>  
> [scala-library-2.11.8.jar!/:na]
>  Caused by: java.io.NotSerializableException: No configured 
> serialization-bindings for class 
> [org.springframework.boot.context.embedded.AnnotationConfigEmbeddedWebApplicationContext]
> at akka.serialization.Serialization.serializerFor(Serialization.scala:235) 
> ~[akka-actor_2.11-2.4.17.jar!/:na]
> at 
> akka.serialization.Serialization.findSerializerFor(Serialization.scala:211) 
> ~[akka-actor_2.11-2.4.17.jar!/:na]
> ...
>
>
> Actualy I'm not sure that it'll work even if there were no serialization 
> error, looks like the problem is a bit more complex.
> Is there a graceful way to tell remote Akka to create a pool of actors via 
> IndirectActorProducer mechanism, as it works for local actors?
>
> Thank you,
> Alex
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to 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.