От RxJava 1 до RxJava 2

Question

Я пытаюсь преобразовать этот RxJava1 код RxJava2

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

Дело в том, что в Rxjava2 я не получаю подписчик, чтобы добавить новую подписку на него.

Answer 1

Наслаждайтесь RxJava 2 лаконичность (Flowable это противодавление поддерживает класс сейчас):

public static Flowable<Path> listFolder(Path dir, String glob) { 
    return Flowable.using(
     () -> Files.newDirectoryStream(dir, glob), 
     stream -> Flowable.fromIterable(stream), 
     stream -> stream.close()); 
} 

Если вы не хотите, противодавление затем заменить Flowable с Observable.