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
Hello @channel, can anyone please help me
I am using these two sbt artifact "com.typesafe.akka" %% "akka-actor" % "2.6.5" and "com.typesafe.akka" %% "akka-remote" % "2.6.5" and the Scala latest version.
I am trying to create a simple remote application in akka, I have given the below configuration
MembersService {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2552
}
}
}
}
MemberServiceLookup {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2553
}
}
}
}
with scala code:
object MembersService extends App {
val config = ConfigFactory.load.getConfig("MembersService")
val system = ActorSystem("MembersService", config)
val worker = system.actorOf(Props[Worker], "remote-worker")
println(s"Worker actor path is ${worker.path}")
}
object MemberServiceLookup extends App {
val config = ConfigFactory.load.getConfig("MemberServiceLookup")
val system = ActorSystem("MemberServiceLookup", config)
val worker = system.actorSelection("akka.tcp://MembersService@127.0.0.1:2552/user/remote-worker")
worker ! Worker.Work("Hi Remote Actor")
}
I need some help regarding akka http. It's not timing out when connecting, and the connection attempt is taking 1-2 minutes despite me setting the akka.http.client.connecting-time to 3 second
Failed Request in 90090 ms Tcp command [Connect(host:10191,None,List(),Some(3 seconds),true)] failed because of java.net.ConnectException: Connection refusedAnybody can advise on how to set a timeout for the connection? Using val akka
2.5.31and akka http10.1.11
Really appreciate if someone can help on this. Has anybody successfully configured the connection timeout?
context.watch(actorRef), context.children() always returns empty ... what could cause this... kindly help thanks
How can I prove that FileIO is picking up my dispatcher?
I'm trying to change the underlying dispatcher of FileIO. The docs say that it is using akka.stream.
blocking-io-dispatcher
akka.stream {
blocking-io-dispatcher = "my-dispatcher"
}
my-dispatcher {
type = Dispatcher
executor = "thread-pool-executor"
thread-pool-executor {
fixed-pool-size = 10
}
throughput = 1
}But just to be really sure, I also set it explicitly in code
FileIO
.fromPath(source.value)
.withAttributes(ActorAttributes.dispatcher("my-dispatcher"))
.log(s"reading ${source.value}")
.fold(ByteString.empty)(_ ++ _)
.map(_.toArray)
.map(ImmutableImage.loader().fromBytes)
}But the log statements are still being written using the default dispatcher. Why?
12:01:47.624UTC DEBUG[default-akka.actor.default-dispatcher-4] [akka.stream.Log(akka://default/system/Materializers/StreamSupervisor-0)] - [reading /Users/oliver/Downloads/e7lp1y1tuwt41.jpg] Element: [...]Is the underlying file operation still done on my thread pool? Is logging just working on another threadpool?
How can I prove that the reading is done by my own threadpool?
Hi, I have a question regarding an akka-streams Graph, I built around ZipLatest:
I was trying to build a graph for processing data from a SourceWithContext. The idea is to split the context from the actual data, process all data chunks and then zip the transformed data with the context of the last element on the stream (each with the same context):
val broadcast = b.add(Broadcast[(Out, Ctx)](2))
val zip = b.add(ZipLatest[Out2, Ctx])
val getLastContextFlow = Flow[(Out, Ctx)].reduce((_, cur) => cur).map(_._2)
broadcast.out(0).map(_._1) ~> processingFlow ~> zip.in0
broadcast.out(1) ~> getLastContextFlow ~> zip.in1
stream.FlowShape(broadcast.in, zip.out)However, this does only work, if the processingFlowonly emits a single message.
I presume, this is because reducewill only emit when the upstream is completed, so the ZipLatestwill simply time-out.
Is there a more elegant way to achieve this?
I can't really funnel the context through the processingFlow because that's a library and it supports neither FlowWithContext, nor tuple-elements on the stream?
Any suggestions are welcome - thanks!
1 reply
@thomasschoeftner I don't think it's possible as streams are infinite in general. So, in order to yield elements you need to traverse the stream to the end once, to get the last element's context, then traverse it again, attaching the found context to every element:
val source: SourceWithContext[Int, String, _] = Source(1 to 5).asSourceWithContext(x => s"$x ctx")
val resSource: Future[SourceWithContext[Int, String, _]] =
source.runWith(Sink.last).map(_._2).map { lastCtx =>
source
.map(_ + 10)
.mapContext(_ => lastCtx)
}
println(Await.result(resSource.flatMap(_.runWith(Sink.seq)), 5.seconds))
out: Vector((11,5 ctx), (12,5 ctx), (13,5 ctx), (14,5 ctx), (15,5 ctx))But, honestly, it'd be easier just convert the stream to a Seq and travers it twice.
2 replies
long story short, it proved that watchTermination does not provide strong enough guarantees since the graph stage preceding it may still process events after watch termination matVal future completes. these events may either be postStop, on upstream/downstream finished/failed or even a push/pull on another port.
thinking about it seems as if what I rally need is a way to 'scope' parts of the graph which seems impossible with the current API, a compromise might be watching the actors interpreting my graph (even better if I could somehow map async islands to actors) - is this somehow possible via the materializer API?
QueryParser.parse(query) match {
case Success(ast) =>
variables.map(parse) match {
case Some(Left(error)) => complete(BadRequest, formatError(error))
case Some(Right(json)) => executeGraphQL(executeFn, ast, operationName, json, tracing.isDefined, authToken)
case None => executeGraphQL(executeFn, ast, operationName, Json.obj(), tracing.isDefined, authToken)
}
case Failure(error) => complete(BadRequest, formatError(error))
}
}
ApiVersion being compiled with JDK 13 and I'm trying to run it in a JDK 11 container. I'm not sure where this comes from since I don't have JDK 13 installed on my machine at all. Any clue what's going on here?
akka.stream.SubscriptionWithCancelException$NoMoreElementsNeeded$ was thrown. without any further information. I had no luck searching for this Exception on the net.
NoMoreElementsNeeded internally, that shouldn’t bubble up to userland however, unless you have some homebuilt custom operators/graphstages.