CODEWITHSUNDEEP
javarx-javareactive-programming
newprint
newprint

Reputation: 7134

rxJava - How Observable can emit from the body of .create method?

I am looking at the following example taken from GitHub, that clearly works, but I don't understand why:

Observable<String> fsObs = CreateObservable.listFolder(
                Paths.get("src", "com", "alex", "experiment"),
                "*.java")
                .flatMap(path -> CreateObservable.from(path));

// CreateObservable.listFolder
public static Observable<Path> listFolder(Path dir, String glob) {
        return Observable.<Path>create(subscriber -> {
            try {
                DirectoryStream<Path> stream =
                        Files.newDirectoryStream(dir, glob);

                subscriber.add(Subscriptions.create(() -> {
                    try {
                        stream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }));
                Observable.<Path>from(stream).subscribe(subscriber);
            } catch (DirectoryIteratorException ex) {
                subscriber.onError(ex);
            } catch (IOException ioe) {
                subscriber.onError(ioe);
            }
        });
    }

Notice that Observable.<Path>from(stream).subscribe(subscriber) is being created. How does this line emits messages into flatMap(path -> CreateObservable.from(path)) ?

Upvotes: 0

Views: 284

Answers (1)

Aaron He
Aaron He

Reputation: 5559

Because DirectoryStream implements Iterable which would iterate over the entries in a directory, and Observable.from(Iterable) converts the sequence into an Observable emitting the items in the sequence, which is the entries in a directory. Then, when a subscriber subscribes to the Observable, it will be receive items being emitted.

A simplified version of Observable.<Path>from(stream).subscribe(subscriber) is just:

for (Path path : stream) {
    subscriber.onNext(path);
}

Calling subscriber.onNext is how the data being emitted.

Upvotes: 1

Related Questions