This channel is available for all Akka enthusiasts—newbies as well as gurus—for the exchange of knowledge and the coordination of efforts around Akka. For more info see: http://bit.ly/akka-gitter
Btw. I opened an issue on typelevel/cats#3729 for a use case to do with Akka. I am not sure that is the right way to do things anymore, but I think it describes an not often spoken about use of Free Monads: one Free Monad, multiple completely different implementations as the Script is sent around to different actors.
Hi there, I am new to alpakka-streams-cassandra connector, is it possible to insert an UDT via CassandraFlow? Using the bound statement I could insert data as a raw string but not directly as a case class that models the UDT. Before using the datastax java driver, i was wondering if any of you managed to do this via the alpakka connector ? :)
I refer to this use case https://doc.akka.io/docs/akka/current/stream/stream-io.html where I'm only interested by the flow output but not the input.
If this is possible then I can use flatMapMerge after a decoding phase.
Or I'm totally wrong ?
you could attach a Source.empty, but sometimes that will cause the connection to be torn down before you want to stop consuming its output. In such cases using Source.maybe might help.
Hi, excuse me, I wanted to ask if Streams is a concept isolated from actors. I understand that through the materializer I can create actors that help to process the Stream but what happens in the case that I already have an actor created (example in the chat, a Chat Room, where the Chat Room Actor would be created with a status as users list connected to it), through the materializer, could an actor already created (ActorRef) be passed to the Stream? This thought for example for the exposure of a Rest API or a websocket where it would need the Stream to process the requests.
3 replies
@bblfish: that distinction never seemed very relevant to me in the Akka context - do you have an example?
My use case is writing an implementation of Tim Berners-Lee's Solid Web Server, with each resource given as an Akka actor, and a guard to determine if the request is allowed access. I'd like it to be used for streaming videos and later enterprise integration.
Hi,
i generated a key for production environment using playGenerateSecret , however i am still getting an exception when making a post request
io.jsonwebtoken.security.WeakKeyException: The signing key's size is 168 bits which is not secure enough for the HS256 algorithm. The JWT JWA Specification (RFC 7518, Section 3.2) states that keys used with HS256 MUST have a size >= 256 bits (the key size must be greater than or equal to the hash output size). Consider using the io.jsonwebtoken.security.Keys class's 'secretKeyFor(SignatureAlgorithm.HS256)' method to create a key guaranteed to be secure enough for HS256. See https://tools.ietf.org/html/rfc7518#section-3.2 for more information.
at io.jsonwebtoken.SignatureAlgorithm.assertValid(SignatureAlgorithm.java:387)
at io.jsonwebtoken.SignatureAlgorithm.assertValidSigningKey(SignatureAlgorithm.java:315)
at io.jsonwebtoken.impl.DefaultJwtBuilder.signWith(DefaultJwtBuilder.java:122)
at io.jsonwebtoken.impl.DefaultJwtBuilder.signWith(DefaultJwtBuilder.java:134)
at io.jsonwebtoken.impl.DefaultJwtBuilder.signWith(DefaultJwtBuilder.java:142)
at play.api.mvc.JWTCookieDataCodec$JWTFormatter.format(Cookie.scala:776)
at play.api.mvc.JWTCookieDataCodec.encode(Cookie.scala:631)
at play.api.mvc.JWTCookieDataCodec.encode$(Cookie.scala:629)
at play.api.mvc.DefaultJWTCookieDataCodec.encode(Cookie.scala:816)
at play.api.mvc.FallbackCookieDataCodec.encode(Cookie.scala:798)
at play.api.mvc.FallbackCookieDataCodec.encode$(Cookie.scala:798)
at play.api.mvc.DefaultSessionCookieBaker.encode(Session.scala:123)
at play.api.mvc.CookieBaker.encodeAsCookie(Cookie.scala:466)
at play.api.mvc.CookieBaker.encodeAsCookie$(Cookie.scala:465)
at play.api.mvc.DefaultSessionCookieBaker.encodeAsCookie(Session.scala:123)
at play.api.mvc.Result.$anonfun$bakeCookies$2(Results.scala:354)
at scala.Option.map(Option.scala:242)
at play.api.mvc.Result.bakeCookies(Results.scala:353)
at play.core.server.common.ServerResultUtils.prepareCookies(ServerResultUtils.scala:291)
at play.core.server.AkkaHttpServer.$anonfun$runAction$5(AkkaHttpServer.scala:432)
at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:434)
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:48)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)I also generated a key using
SecretKey key = Keys.secretKeyFor(SignatureAlgorithm.HS256);
String secret = Encoders.BASE64.encode(key.getEncoded());But still getting the same exception
Hi, I've been looking at the documentation and in the examples I always find that the root actor (ActorSystem) exposes the entire protocol. This means that it exposes all the possible commands that can be done in the entire hierarchy, is this not redundant? I mean, if I want to consume a specific command from a child actor, can't I bring a child actor from a non-actor class and start the protocol for that child? Example (chat): a gabbler has the command to be able to send a message to a room, then for another class to consume it, not an actor, the command is raised to the root Actor causing the root to send a message to the gabbler which then sends it to the room.
5 replies
Hello, i have a question
I am now developing using cluster akka. The reference example is this link (https://github.com/elleFlorio/akka-cluster-playground)
I am now developing using cluster akka. The reference example is this link (https://github.com/elleFlorio/akka-cluster-playground)
Basically, the code structure is almost the same.
I have seen that actors are created in a round robin fashion to cluster members according to the akka actor deployment configuration.
object ProcessorActor {
case object Hello
case object PingBroad
def props(roomID: Int, userID: Int) = Props(new ProcessorActor(roomID, userID))
}
class ProcessorActor(roomID: Int, userID: Int) extends Actor with ActorLogging {
override def preStart(): Unit = {
log.info("Room {} User {}", roomID, userID)
}
override def receive: Receive = {
case Hello => {
log.info("@@ PING ACTORS My actor path {} @@", context.self.path)
}
case PingBroad => {
log.info("@@ PING BROADCAST My actor path {} @@", context.self.path)
}
case _ =>
}
}class Node(ip: String, nodeId: String) extends Actor with ActorLogging {
var host: String = ip
val processor: ActorRef = context.actorOf(Processor.props(nodeId), "processor")
val processorRouter: ActorRef = context.actorOf(FromConfig.props(Props.empty), "processorRouter")
val clusterManager: ActorRef = context.actorOf(ClusterManager.props(nodeId), "clusterManager")
private var actors: Set[ActorRef] = Set.empty
override def receive: Receive = {
case GetClusterMembers => clusterManager forward GetMembers
case GetFibonacci(value) => processorRouter forward ComputeFibonacci(value)
case GenActor(roomID: Int, userID: Int) => processorRouter forward CreateActor(roomID, userID)
case CreatedActor(result: ActorRef, replyTo: ActorRef) => {
log.info("{} created", result.path)
actors += result
replyTo ! ProcessorResponse(nodeId, actors.size)
}
case GetActorCount => {
val replyTo = sender()
replyTo ! ProcessorResponse(nodeId, actors.size)
}
case PingActors => {
actors.foreach(_ ! Hello)
}
case PingRoom(roomID: Int) => {
actors.foreach(_ ! PingBroad)
context.actorSelection(s"akka.tcp://cluster-playground@*/user/node/processor/$roomID-*") ! PingBroad
}
}
}
Hi guys one question : How can we shutdown ActorSystem base on exceptions thrown in the main App :
object Boot extends App{
implicit val system: ActorSystem = ActorSystem("Boot")
implicit val materializer: ActorMaterializer = ActorMaterializer()
throw new Exception("Boom!!!")
}
in this case ActorSystem will be still alive and App will not stop anymore
is there any way to catch jvm exceptions indirectly , maybe some crazy idea
the main problem is i want to shutdown system base on exception thrown , for example this snippet :
object Boot extends App{
implicit val system: ActorSystem = ActorSystem("Boot")
implicit val materializer: ActorMaterializer = ActorMaterializer()
sys.addShutdownHook(system.terminate())
throw new Exception("Boom!!!")
}it still not work for my situation , the problem is we have so many logic out of ActorSystem , and for some reason exceptions stack reaches to Boot.scala , but as ActorSystem will not shutdown , so App will not terminated,
we want to terminate the App if any exception escalated to the
Boot object (our main singleton)
4 replies
Hello guys, I can't find a solution using existing stream operators for the following problem:
I have a source of elements (here Strings) which are sent to an actor for processing.
I want this actor to return Done and not the string itself to be sure it can't modify the string.
But I need keep the string in the scope to publish it downstream, how can I do that ?
val actorFlow: Flow[String, Done, NotUsed] = ActorFlow.ask ...
val statefulFlowOfString: Flow[String, String, NotUsed] = ???
Source(List("A", "B", "C")
.via(statefulFlowOfString)It could look like statefulMapConcat except that this method must return an IterableOnce where I would like to return a Flow[String, String, NotUsed]
Thanks
@Kyocfe: indeed actorSelection doesn't work across the cluster without knowing IP addresses, and usually other mechanisms to obtain ActorRef's (like cluster sharding) are nicer in that case
1 reply
Hi all,
I'm trying to upload a file with the Akka Http client using:
def createEntity(file: File): Future[RequestEntity] = {
require(file.exists())
val formData =
Multipart.FormData(
Source.single(Multipart.FormData.BodyPart.fromFile("file", MediaTypes.`application/octet-stream`, file)))
Marshal(formData).to[RequestEntity]
}However:
[2021-01-18 11:53:02,377] [ERROR] [akka.actor.ActorSystemImpl] [MediaGatewayHttpServer-akka.actor.default-dispatcher-240] [akka.actor.ActorSystemImpl(MediaGatewayHttpServer)] - Error during processing of request: 'EntityStreamSizeException: incoming entity size (while streaming) exceeded size limit (8388608 bytes)! This may have been a parser limit (set via `akka.http.[server|client].parsing.max-content-length`), a decoder limit (set via `akka.http.routing.decode-max-size`), or a custom limit set with `withSizeLimit`.'. Completing with 413 Payload Too Large response. To change default exception handling behavior, provide a custom ExceptionHandler.Anyone able to help me out?
6 replies
Hi! Question on an implementation detail of the SBR.
What is the reason for interpreting exited (exiting confirmed?) and downed nodes as unreachable? During a rolling update this ends up triggering the SBR with a noop and logs
What is the reason for interpreting exited (exiting confirmed?) and downed nodes as unreachable? During a rolling update this ends up triggering the SBR with a noop and logs
SBR found unreachable members... which is not true. def receive: Receive = {
// ...
case UnreachableMember(m) => unreachableMember(m)
case MemberDowned(m) => unreachableMember(m)
case MemberExited(m) => unreachableMember(m)
// ...
}
Hi,
We are using Akka cluster & DistributedPubSub on kubernetes,
We have a problem, in some cases when we are calling "mediator.ask" the target actor is not receiving the message and the message goes to dead letters.
It happens even if the target actor is on the same kubernetes pod as the sender.
Does anyone have an idea why it can happen?
We are using Akka cluster & DistributedPubSub on kubernetes,
We have a problem, in some cases when we are calling "mediator.ask" the target actor is not receiving the message and the message goes to dead letters.
It happens even if the target actor is on the same kubernetes pod as the sender.
Does anyone have an idea why it can happen?
Hi there! I have a question about the standard Kinesis connector. Does it support aggregation of messages into a single record (as KPL does, see https://docs.aws.amazon.com/streams/latest/dev/kinesis-kpl-concepts.html )? I mean, besides supporting batching of records and retries of failed PUTs.
Thanks in advance!