Myujth

Lets say I have a very long string:

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

I need all the chunks of data up to this slash, but not any slashes. I see this as a stream and want to do the following:

1. I start to read symbols from the back and count slashes.


2. Anything but slash I put into Last-In-First-Out data structure.


3. In order not to wait for the whole operation to finish, I start reading data from the lifo datastructure as it becomes available.


4. I terminate after the 9999999th '/'

Can something like this be accomplished with reactive streams and how?

I think the following code will achve what you want

@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();

The example above utilizes the project reactor. Which is pretty cool way of doing the reactive stuff inside of java.

There is plenty of optimization that can be done in the following code. Not using Strings to represent a single letter would be one of them.

But the basic idea is there. You create flux/observable that emits a letters as they come in and make that observable shareable (you have to window over emitting values) and then just collect them in to a single sting. The code bellow should give the following output:

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

Of course you have to utilize non-blocking connection so the bytes could be read asynchronously.