score:0
Accepted answer
the problem is in your sink. that sink.head will return one element from your materialized stream. so the question is, why it´s received more than one value when mapasync(>1) is used in stream materialization?. maybe it´s because it uses more than one actor pushing values downstream.
in any case, change your sink to something like:
val sink: sink[string, future[string]] = sink.fold("")((a, b) => b ++ a)
and it will work.
Source: stackoverflow.com
Related Query
- akka streams stops after parallelism
- Akka streams Source.repeat stops after 100 requests
- Akka streams pass through flow limiting Parallelism / throughput of processing flow
- Akka streams - resuming graph with broadcast and zip after failure
- Akka Streams stop stream after process n elements
- How to work Supervision strategy after recover at non-linear graphs in Akka Streams
- Akka streams kafka commit offset after filter
- Akka server stops immediately after running
- Akka Streams takeWhile processing next element even after condition fails
- Get reference of original element after Akka Streams Sink?
- Parsing stops with Akka Streams mapAsync
- Akka Stream Kafka vs Kafka Streams
- Creating a flow from actor in Akka Streams
- akka http: Akka streams vs actors to build a rest service
- How to use an Akka Streams SourceQueue with PlayFramework
- Why is Akka Streams swallowing my exceptions?
- Proper way to stop Akka Streams on condition
- Reading a CSV files using Akka Streams
- How to split an inbound stream on a delimiter character using Akka Streams
- How to download a HTTP resource to a file with Akka Streams and HTTP?
- Akka Streams custom merge
- How to properly stop Akka streams from the outside
- Akka Streams - How to keep materialized value of an auxiliary Sink in a Graph
- How to clean up substreams in continuous Akka streams
- How to assemble an Akka Streams sink from multiple file writes?
- Akka Streams Websocket Wiring
- Count number of elements in Akka Streams
- Websocket Proxy using Play 2.6 and akka streams
- Akka Streams running on cluster nodes
- Akka Streams split stream by type
More Query from same tag
- Dynamic languages - which one should I choose?
- Scala - Using DF filter on multiple fields
- Set similarity join using Spark
- How to send a 404 (or another specific) HTTP error in Lift?
- Why does the Spark application code need to be an object rather than a class?
- Akka Remoting for a grid server application
- Figure out a way to convert a var implementation to val
- Scaladoc: Do I really need to give the full path everytime I use a link?
- How to convert following flatMap/map snippet to for-comprehension in Scala?
- how is (myActor ? q) async in following scalatra example?
- Is there a library which provides utility functions for views based on twitter bootstrap for play framework
- Convert akka ByteString to Java InputStream?
- Nothing inferred for type parameter
- Load data with where clause in spark dataframe
- open file via rest api in intellij 2016
- scala 3 opaque type: cannot call methods defined in extension
- scala priority queue not ordering properly?
- Retrieve spark dataframe array column value and reuse it as a column name in a UDF
- scala implicit conversion for case object
- Play-json: Error on updating an object attribute inside an array
- Scala implicits exercises
- using a partial function for catch block with scala
- Error reading s3 bucket it spark
- Akka with SBT and IntelliJ
- Http Get request is not working
- How do I add new value to already existing JSON file?
- Using case class to display Expressions
- Scala parallel execution
- Split field and create multi rows from one row Spark-Scala
- Scala SortedMap : Get all keys greater than a given key