Реактивный потребитель, наблюдаемый в Java

Question

Допустим, у меня очень длинная строка:

    trillions of chunks
            |
            v
    /asdf/........./bar/baz/foo
                ^
                |
    what I try to find is closer to the right:
        the data after 9999999th '/'

Мне нужны все порции данных до этой косой черты, но не косые черты. Я вижу это как поток и хочу сделать следующее:

1. Я начинаю читать символы со спины и считать слеш.
2. Все что угодно, кроме косой черты, которую я вставил в структуру данных «Последний пришел первым».
3. Чтобы не ждать завершения всей операции, я начинаю читать данные из структуры данных lifo, когда она становится доступной.
4. Я прекращаю работу после 9999999-го '/'

Может ли нечто подобное быть достигнуто с помощью реактивных потоков и как?

Answer 1

Я думаю, следующий код даст то, что вы хотите

@Test
public void reactiveParser() throws InterruptedException {
    ConnectableFlux<String> letters = Flux.create((Consumer<? super FluxSink<String>>) t -> {
        char[] chars = "sfdsfsdf/sdf/sdfs/dfsdfsd/fsd/fsd/fs/df/sdf".toCharArray();
        for (char c : chars) {
            t.next(String.valueOf(c));
        }
    }).publish();

    letters
            .window(
                    letters.filter( t -> t.equals("/"))
            )
            .flatMap( t -> t.collectList())
            .map( t -> t.stream().collect(Collectors.joining()))
            .subscribe(t -> {
                System.out.println(t);
            });

    letters.connect();
}

В приведенном выше примере используется проектный реактор.Это довольно крутой способ делать реактивные вещи внутри Java.

В следующем коде можно выполнить множество оптимизаций.Не использовать строки для представления одной буквы будет одним из них.

Но основная идея есть.Вы создаете поток / наблюдаемый, который испускает буквы по мере их поступления, и делаете его видимым для совместного использования (вам нужно показывать поверх излучаемых значений), а затем просто собираете их в единую строку.Код ниже должен дать следующий вывод:

sfdsfsdf
/sdf
/sdfs
/dfsdfsd
/fsd
/fsd
/fs
/df

Конечно, вы должны использовать неблокирующее соединение, чтобы байты могли считываться асинхронно.