rxJava - Как Observable может испускать из тела метода .create?

Question

Я смотрю на следующем примере, взятом из GitHub, который четко работает, но я не понимаю, почему:

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); 
      } 
     }); 
    } 

Обратите внимание, что Observable.<Path>from(stream).subscribe(subscriber) создается. Как эта линия выдает сообщения в flatMap(path -> CreateObservable.from(path)) ?

Answer 1

Поскольку DirectoryStream реализует Iterable который будет перебирать записи в каталоге, и Observable.from(Iterable) преобразует последовательность в наблюдаемый излучающие элементы в последовательности, которая является записью в каталоге. Затем, когда абонент подписывается на Observable, он будет получать элементы, которые будут испускаться.

упрощенная версия Observable.<Path>from(stream).subscribe(subscriber) просто:

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

Вызов subscriber.onNext как данные, излучаемых.