Questions tagged [akka-stream]

Akka implementation for handling streaming data on the JVM

Filter by
Sorted by
Tagged with
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 ...
Sushant Somani's user avatar
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 ...
Sushant Somani's user avatar
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 ...
Sushant Somani's user avatar
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 ...
Uko's user avatar
  • 13.1k
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 ...
David Santiago Gantiva Castro's user avatar
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.
Chris Stewart's user avatar
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 ...
Allan's user avatar
  • 81
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 ...
Arunav Sanyal's user avatar
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 ...
CannyMiao's user avatar
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....
Skeeve's user avatar
  • 1,205
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 ...
user3477337's user avatar
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 ...
user3468054's user avatar
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/...
blue-sky's user avatar
  • 52.1k
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. ...
user6722996's user avatar
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 ...
pawel.panasewicz's user avatar
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 ...
Arun's user avatar
  • 67
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 ...
Arya's user avatar
  • 2,839
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 = ...
rimmi anand's user avatar
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(&...
Sky's user avatar
  • 2,509
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(...
Bill Goldberg's user avatar
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 ...
Sigurd's user avatar
  • 51
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[...
vasigorc's user avatar
  • 882
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(...
Nitesh's user avatar
  • 1
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 ...
Zaryab Ali's user avatar
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. ...
Zaryab Ali's user avatar
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, ...
ArunasR's user avatar
  • 1,917
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 ...
user1185076's user avatar
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, ...
Md Shahnewaz's user avatar
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 => ...
Md Shahnewaz's user avatar
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/...
Always_a_learner's user avatar
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. ...
martinnovoty's user avatar
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 ...
Alexandru Toma's user avatar
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) ...
Ahmed Wasim's user avatar
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 ...
Daniel's user avatar
  • 1
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 ...
Siddharth Shankar's user avatar
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 ...
silverberry's user avatar
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 ...
Nandu's user avatar
  • 151
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[...
Siddharth Shankar's user avatar
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 ...
Daniel82's user avatar
  • 101
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 ...
sadegh's user avatar
  • 21
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 ...
ach2ashes's user avatar
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[_]]...
sachin's user avatar
  • 1
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 ...
nicholsonjf's user avatar
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 (...
Rlupo07's user avatar
  • 51
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 ...
Siddharth Shankar's user avatar
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 ...
landau's user avatar
  • 41
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 ...
Always_a_learner's user avatar
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
Always_a_learner's user avatar
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 ...
Mathivanan's user avatar
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 ...
Siddharth Shankar's user avatar

1
2 3 4 5
35