RxJava buffer/window until element suffers condition

Question

I am trying to read lines from a reader and group them to blocks that belong together.

Source text:

bla1
bla2
### block separator ###
bla3
bla4
### block separator ###
...

I need to get the two blocks (bla1, bla2) and (bla3, bla4).

Code:

import org.apache.commons.lang3.StringUtils;
import rx.Observable;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.Reader;
import java.util.Iterator;

public class BlockBuilder {

  public static void main(String[] args) {

    try {
      FileReader fileReader = new FileReader("/path/to/some/file");
      LineIterable lineIterable = new LineIterable(fileReader);

      Observable.from(lineIterable)
          .buffer(100)
          // Needed instead of time/count: until line matches condition
          // something like .buffer(line -> line.equals("### block separator ###")
          .forEach(gatheredLines -> {
            String gatheredBlock = StringUtils.join(gatheredLines, '\n');
            System.out.println(gatheredBlock);
            System.out.println("###### ###### ###### ######");
          });
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }

  private static class LineIterable implements Iterable<String> {
    private final Iterator<String> iterator;
    public LineIterable(Reader reader) {
      iterator = new BufferedReader(reader).lines().iterator();
    }
    @Override
    public Iterator<String> iterator() {
      return iterator;
    }
  } 
}

It doesn't matter if buffer or window is used or if I'm completely wrong for thinking of those two.

I thought it must be possible with the bufferClosingSelector for the buffer or the closingSelector for the window. Both are functions that create an Observer which can trigger the closing of the current buffer or window but I can't see where I can get hold of the current line here.

akarnokd · Accepted Answer · 2015-07-09 12:03:32Z

3

You can publish your source and use it for both buffering and buffer boundary:

Observable<String> source = Observable.just(
        "a", "b", "#", 
        "c", "d", "e", "#", 
        "f", "g");

source.publish(p -> 
        p.filter(v -> !"#".equals(v))
        .buffer(() -> p.filter(v -> "#".equals(v))))
.subscribe(System.out::println);

answered Jul 9, 2015 at 12:03

akarnokd

70.1k14 gold badges161 silver badges199 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

snieke Over a year ago

That would be great and it compiles with no error. But as soon as I try this code I get a huge error: Error:(16, 25) java: no suitable method found for buffer(()->p.filt[...]s(v))) method rx.Observable.<TClosing>buffer(rx.functions.Func0<? extends rx.Observable<? extends TClosing>>) is not applicable (cannot infer type-variable(s) TClosing (argument mismatch; bad return type in lambda expression rx.Observable<java.lang.String> cannot be converted to ? extends rx.Observable<? extends TClosing>)) [...]

akarnokd Over a year ago

Works for me from Eclipse 4.5. You may need to add explicit <String> before the second p.filter() call.

Matthias Over a year ago

Can someone explain what is the use of publish() in this scenario?

akarnokd Over a year ago

The signals of the source must be multicast to multiple sub-flows without subscribing to the source multiple times (which is not guaranteed to emit the same data).

Collectives™ on Stack Overflow

RxJava buffer/window until element suffers condition

1 Answer 1

4 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related