Очень похоже на Вадима

Кого?

что именно понятно? а то мне чот непонятно

Коллеги, подскажите каким путем пойти:
в akka streams нужна функция типа groupedWithin, но только чтобы на выходе брался последний элемент
(буфферизация всех данных не нужна, нужен только последний элемент)

последний элемент за каждый временной промежуток?

за временной промежуток, либо за размер пачки (что первее пробьется)

А делать groupWithin().map(_.last) — по памяти не устраивает?

да

а что взять под задачу есть: service A -> http -> service B, и нужно в сервисе A сделать cancel но чтобы cancel закрывал также процес и на service B ?

Сейчас попробую написать

😍

akka http умеет в такое?

а причем тут akka http?

К сожалению, не вышло написать нормально.
Тут я попробовал разделить стрим на два набора - один аггрегирует всё, что не попало в throttle и ловит последний элемент в группах из n.
Но т.к. throttle начинает бэкпрешшу только после того, как получил хотя бы один элемент, его элемент может быть старше. Таким образом, порядок ломается
https://scastie.scala-lang.org/Odomontois/aPjgt5t1SLGpsCEXNqwdfw/2


Думаю, если вам нужен чистый троттл по времени без ограничения на размер можно просто взять цепочку
.conflate .throttle

А то, что вы описали, похоже только кастомным стейджем

Благодарю за попытку! У меня тоже не вышло. Сейчас я пользуюсь измененной версией groupedWithin. (Взял к себе имплементацию и вместо аггрегации сделал сохранение последнего значения)

Ну вот и хорошо

полухорошо)) имплементация относится к @InternalApi

ну вы же скопипастили

Т.е. написали просто новый формально код с использованием апи для стейжей