It is exactly what I have tried to achieve! You guys did amazing work with all akka streams. Thanks! :)
четверг, 22 января 2015 г., 14:43:52 UTC+6 пользователь drewhk написал: > > Hi Alexey, > > > > On Thu, Jan 22, 2015 at 4:15 AM, Alexey Romanchuk <alexey.r...@gmail.com > <javascript:>> wrote: > >> Hey! >> >> I have a stream that process incoming messages, assemble big message >> "pack" and send it to other system via network. Incoming messages are >> relatively small and I use big buffers to improve throughput for all stages >> related to these small messages. At the very end of stream I have "message >> collector" which aggregates messages and periodically produces message >> "packs" which must be sent via network. >> >> System looks like this: >> >> (Input) -> (Complex flow for small messages) -> (Aggregator) -> (Network >> sender) >> >> I increased buffer size to 128 to improve throughput, but the problem is >> in network sender. Network sender can resend data several times in case of >> network failure. I want to force only 1 element buffer before network >> sender to perform effective backpressure to input. In case of same size >> buffer size stream can be flooded with lot of huge aggregators packages. >> >> I know that there is "buffer" flow combinator, but it looks like it work >> only if I want to increase buffer size before some stage, but does not work >> for decreasing buffer size. >> > > You can control the internal buffer size for stages as it is explained > here: > http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-M2/scala/stream-rate.html#Internal_buffers_and_their_effect > > > 1. val flow = > 2. Flow[Int] > 3. .section(OperationAttributes.inputBuffer(initial = 1, max = 1)) { > sectionFlow => > 4. // the buffer size of this map is 1 > 5. sectionFlow.map(_ * 2) > 6. } > 7. .map(_ / 2) // the buffer size of this map is the default > > > -Endre > > >> >> Could you advise right way to achieve this kind of backpressure? >> >> Thanks! >> >> -- >> >>>>>>>>>> Read the docs: http://akka.io/docs/ >> >>>>>>>>>> Check the FAQ: >> http://doc.akka.io/docs/akka/current/additional/faq.html >> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user >> --- >> You received this message because you are subscribed to the Google Groups >> "Akka User List" group. >> To unsubscribe from this group and stop receiving emails from it, send an >> email to akka-user+...@googlegroups.com <javascript:>. >> To post to this group, send email to akka...@googlegroups.com >> <javascript:>. >> Visit this group at http://groups.google.com/group/akka-user. >> For more options, visit https://groups.google.com/d/optout. >> > > -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
