Von RxJava 1 bis RxJava 2

Question

Ich versuche, dieses RxJava1 Code 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); 
     } 
    }); 
} 

Die Sache, die ich nicht bekommen ein Teilnehmer ein neues Abonnement, um es hinzuzufügen in Rxjava2 ist zu konvertieren.

Answer 1

Genießen RxJava 2 Prägnanz (Flowable ist die Stützklasse Rückstau jetzt):

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

Wenn Sie nicht wollen, Rückstau Flowable dann mit Observable ersetzen.