Questions tagged [akka-stream]
Akka implementation for handling streaming data on the JVM
1,740
questions
0
votes
1
answer
41
views
Getting overloaded method ~> with alternatives Akka broadcast
I am trying to broadcast incoming Source[ByteString, Any] to 2 different flows and then fan-in(zip) the output. However I'm getting error "overloaded method ~> with alternatives".
val ...
0
votes
0
answers
36
views
withSizeLimit Akka directive not working for nested Future method calls
I am trying to upload a file to S3 & then to Google Cloud. However I also want to check the file size before upload. It should be <=3Mb.
I have created a Route and it has a withSizeLimit ...
0
votes
2
answers
41
views
Fetch Source[ByteString, Any] out of stream into 2 sinks
I am trying to put incoming Source[ByteString, Any] into 2 sinks & trying to duplicate the incoming stream in akka streaming graphs. I am getting desired 'is' as input stream but bs in not of type ...
1
vote
2
answers
61
views
Limit a Source ticking frequency in Akka streams and always act on the latest message
Here is a context of the problem:
There is a Source, it's constantly ticking, there is no guarantee about the ticking frequency
We want to limit the maximum ticking rate of the source (e.g., we are ...
0
votes
1
answer
30
views
How to set headers in RabbitMq with Scala
Hi I'm trying to set headers in the rabbit messages with the library akka-stream-alpakka-amqp in scala but I can't be able to make it work and reading the repository I can't see any reference or ...
1
vote
1
answer
33
views
Inverse of `Flow.batch` in akka streams
Is there an inverse to Flow.batch
I want the signature of the method to be Flow[Vector[T],T,NotUsed]
It should unwrap a Vector and emit every element of the Vector downstream as a single element.
0
votes
1
answer
48
views
Scala-3.3.0: dotty.tools.FatalError: cannot resolve reference to
I'm seeing this error with scala-3.3.0 and akka (or pekko):
dotty.tools.FatalError: cannot resolve reference to type akka.stream.scaladsl.type.Source
dotty.tools.FatalError: cannot resolve reference ...
1
vote
1
answer
58
views
Regarding clean shutdown for actor system and akka stream
We wish to accomplish the following on akka stream shutdown
The source is stopped,
The elements in the stream are completed upto a timeout point
then the stream/actor system is shut down
We have the ...
0
votes
1
answer
30
views
Why does this java.lang.IllegalStateException happen, when I use playframework to response a chunked akka source?
When I use playframework to response a mp3 audio. I want to use the chunked transceiver. Then I make a akka source in akka stream to response. And use the play Ok.chunked resule. But it has a ...
1
vote
1
answer
62
views
Excetption thrown while executing Akka Streams doesn't contain any references to the source code
There is an akka streams scala application, it consumes messages from kafka topics, converts them to Dynamo DB requests, and executes them. Sometimes the app throws exceptions these exceptions:
java....
1
vote
2
answers
150
views
Akka 2.7.0 memory leak?
I'm using Akka actor system with version 2.7.0 and we are facing memory leak issue as below:
When the process gets completed, memory is not getting free and below are two potential causes:
enter ...
2
votes
1
answer
46
views
Akka streams, conditional divertLeft
I have a divertLeft function, based on the code here: https://bszwej.medium.com/akka-streams-error-handling-7ff9cc01bc12, which will divert Lefts to the specified sink, and pass Rights onward to the ...
0
votes
1
answer
31
views
Akka streams over web socket fails to process/.receive data after successful authorise (Java)
I'm attempting to learn about Akka web sockets using a publicly available finance stream of data, the socket authenticates but does not receive the stream of data.
Reading https://doc.akka.io/docs/...
0
votes
0
answers
31
views
Need help reusing ScalikeJDBC caller's transaction in method returning Akka stream
I am not able to reuse ScalikeJDBC caller's session in a method that returns Akka stream.
I am trying to write a class for database access that returns Source for use in Akka streaming application.
...
0
votes
0
answers
33
views
Xml Framing for akka streams
I'm consuming stream of ByteStrings from TCP endpoint using akka streams. Chunks of data should form one or many XML documents. I'd like my stream to output ByteStrings with full Xmls.
For example ...
0
votes
1
answer
132
views
Akka - Actor Cluster Inconsistent Shard rebalance while Rolling restart of nodes
Issue
When more no of actors in a shard the rebalancing is not happening properly when rolling restart of nodes.
When the nodes restarted one by one sequentially some shard not rebalancing to other ...
1
vote
1
answer
118
views
Akka Streams: StreamDetachedException when offering to a QueueSource
We have a DownloadFileFlow class which uses Akka Streams & Akka Http for downloading files from difference domains; Sometimes (~50%) in offering to the QueueSource we encounter ...
0
votes
1
answer
99
views
Supervision.decider strategy in akka 2.6.0
I have to pass supervision.decider but in akka 2.6.0+ Materializer does not have similar settings.
Can anyone share how to write below code in higher version:
val mat: ActorMaterializer = ...
0
votes
1
answer
131
views
Akka http- Request timed out after 10 seconds while waiting for entity data: Consider increasing the timeout for toStrict
I am using akka http server for file upload as:
(path("file" / "upload") & post) {
formFields("formParams") { formParams =>
fileUpload(&...
1
vote
1
answer
23
views
Compile error while returning chunked response from play 2.8.18
I am trying out chunked response in play framework.
The below controller code works fine.
def chunkedSample = Action {
val source = Source.apply(List("Hello","Hai"))
Ok.chunked(...
0
votes
0
answers
25
views
Handling Errors in Akka Streams: Using Exception and current element of the stream
I'm using Akka Stream 2.6 and I want to handle a failed stream.
Source.from(...).via(business()).recover(MyException.class, elem -> errorMessage())
I'm aware, that I can map an Exception based on ...
0
votes
0
answers
70
views
Need help testing a RestartFlow using Akka Stream Testkit
I have a following flow using Akka Stream scaladsl:
@testOnly
def restartableFlow: Flow[MyEvent, AmqpSendResult, NotUsed] = RestartFlow
.withBackoff(restartSettings) { () =>
Flow[...
0
votes
1
answer
62
views
Retry mechanism for GraphStage akka
How can we retry/restart a graphStage in case of failures in postStop() ?
I am relatively new to akka and would really appreciate your help.
Here is what I am trying to do:
source
.via(...
0
votes
1
answer
38
views
Don't want provider to create new instance each time while using Google Guice Dependency Injection
I.m new to Google Guice Dependency Injection. I'm implementing (Provider), but it's making new instance each time. I want it to use same instance of StudentRepository Class which it was created first ...
0
votes
0
answers
99
views
How to send JSON Response with Status Code in akka-http (Scala)?
I'm new to akka-http and trying to send JSON Response with Status Codes in scala, but I'm able to do that. Help me out. Basically I'm getting error while I send HttpResponse with ContentTypes as JSON.
...
1
vote
0
answers
44
views
Dividing Akka source into batches
I have a file parser, something like
def parser(file: File): Source[Record, NotUsed] = ...
and a function that consumes Records:
def send(source: Source[Record, _]: Future[Result]
The files are big, ...
0
votes
1
answer
56
views
Get an object out of Akka future response
I'm using Akka streams to rate limit the calls to an external rest application and use its result to enrich the data.
I can't get to map the Future HttpResponse to the data in the stream.
What I've ...
1
vote
2
answers
144
views
How to Implement Actor.Ask to Control the flow of data at a certain rate
I used below Method but it is using large amount of CPU so i want to use Actor.Ask instead of below method can anyone help me how to use Actor.Ask
KafkaConsumer.PlainSource(
consumerSettings, ...
1
vote
1
answer
113
views
KafkaConsumer.PlainSource Method (in C#.Net) Using large amount of CPU
below method using large amount of CPU can anyone help me to minimize the CPU usage with proper solution.
KafkaConsumer.PlainSource(
consumerSettings, subscription)
.RunForeach(result =>
...
1
vote
1
answer
91
views
scala-akka:- Passing command line arguments to sbt run to locate .csv files
I have a command line program that calculates statistics from humidity sensor data.
I also have .csv files inside src/main/scala/data
When i do sbt "run data" or sbt "run src/main/scala/...
0
votes
1
answer
33
views
akka stream behaviour based on outside variable
I have a generic question on Akka Streams,
I need to change stream behavior based on variable outside of akka. The variable is static and is changed by other piece of code.
How would you achieve this. ...
0
votes
0
answers
67
views
Client send message on tcp socket connection using akka streams
After successfully connected to a tcp socket server, how can i sent the messages to that server?
i have created this client, and is working fine as long as is a two way communication with the server ...
0
votes
1
answer
59
views
IOResult methods deprecated
I am new to Scala and am using the method def process however getting build errors like "method wasSuccessful in class IOResult is deprecated (since 2.6.0): status is always set to Success(Done)
...
0
votes
1
answer
43
views
Akka.Net Stream Ordering (Stream Ordering Over the Network)
Does Akka.net Streams preserve input order of elements? If so, does this hold true when working with reactive streams over the network (when using stream refs)?
Current Version of Akka.Streams being ...
0
votes
2
answers
103
views
Adding values from a Map to an existing Source in Scala
I am using akka streams. Can we add values from a map to an existing source in scala? I want to merge all the values from a Future[Map[String, A]] and a source of A Source[A,_] which are not already ...
0
votes
1
answer
163
views
Problems with Incompletely Consumed Akka Streams
We have a service that uses Alpakka 3.0.4 (Scala 2.13) to stream multiple large files from S3, zip them, and send the zipped stream out as HTTP responses. The idea is to start sending out even before ...
1
vote
2
answers
60
views
Not able to execute Akka Work pulling java example provided in the akka document
I am interested in understanding the work pull feature in akka. I am trying to create a main method for the example provided that will create and start the producer(ImageWorkManager class in the ...
0
votes
1
answer
105
views
Compare two tuples into a source of touples in scala
I am trying to compare two tuples of objects with each other and return a Source of latest tuples. For example, I have two objects ObjectA and ObjectB which are in form of tuple like (ObjectA, Some[...
0
votes
0
answers
53
views
BroadcastHub instances on Websocket connection are not cleaned up after disconnect
I have a web applications that is only available on an intranet.
It serves data to the frontend asynchronously via Websocket.
However, running it for a while will create multiple instances of ...
2
votes
0
answers
56
views
Alpakka search in elasticsearch without scrolling
can someone help me how can I search without using the scroll feature inside the elastic with alpakka.
I want to use alpakka or akka-stream to reactively send a query to elastic without scrolling, But ...
0
votes
1
answer
72
views
Code not working for FileTailSource in akka Stream
I have a log file that is fed data about every visit for some websites ( it is actually just a python programm that simaulates this ) , i want to count the number of visits per websites .
So i m ...
0
votes
1
answer
177
views
Akka to Scala Cats FS2 migration
"Source" and "OverflowStrategy" are parts of Akka streams. How can I reimplement below code in FS2?
object Topic {
def apply: Topic = {
val (a1, a2) = Source.queue[Message[_]]...
0
votes
1
answer
50
views
Akka stream fails when throttled
I'm using Akka Streams to stream lines of text from a file. All I'm trying to do currently is print each line to stdout. I'm able to get this working without any throttling, i.e. all the lines print ...
0
votes
1
answer
239
views
Akka Stream application using more memory than the jvm's heap
Summary:
I have a Java application that uses akka streams that's using more memory than I have specified the jvm to use. The below values are what I have set through the JAVA_OPTS.
maximum heap size (...
0
votes
1
answer
46
views
Iterating over two Source and filter using a property in Scala
I am trying to filter out common elements in terms of latest versions of the object and return another Source. My object looks like:
case class Record(id: String, version: Long)
My method's input are ...
0
votes
1
answer
72
views
Benefits of Alpakka Kafka Producer over plain Kafka Producer
I have an application that receives an api request and relays its to a Kafka API Producer. Each request calls the producer to send a message to Kafka. The producer exists throughout the application ...
3
votes
1
answer
240
views
scala-akka-streams difference between run() and runWIth()
What are the key difference between run and runWith:-
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Keep, Sink, Source}
object RunAndRunWith extends ...
2
votes
1
answer
81
views
How to run futures in their own execution context and not on the Actor System dispatcher. [Scala|Akka]
How do we use src/main/resources/application.conf for the same where the dedicated dispatcher in our code would automatically lookup in application.conf for the dispatcher
0
votes
0
answers
44
views
How to pull records from bigquery using Akka Streams Alpakka Big query connector
I have a query for big query native table, currently I'm able to run and get the data and write to sync, but I need to run the same query periodically and fetch the results.
My implementation looks ...
0
votes
1
answer
151
views
Convert a Source to Flow in Scala
How to convert a Source to Flow?
Input: Source[ByteString,NotUsed] a
Intermediary Step: Call an API which returns an InputStream
Output: Flow[ByteString,ByteString,NotUsed]
I am doing it as:
Type of ...