Wie schreibt man Observable <Buffer> in Datei mit einer Vert.x Pumpe?

Question

Ich möchte alle emittierten Puffer der rs Observable in eine Datei mit einem Vert.x 3.5.0 Pump (RxJava2 Variante) streamen. Das folgende Beispiel funktioniert irgendwie nicht und ich weiß nicht warum.

import static org.junit.Assert.assertEquals; 

import java.io.File; 
import java.io.IOException; 

import org.apache.commons.io.FileUtils; 
import org.junit.Test; 

import io.reactivex.Observable; 
import io.vertx.core.file.OpenOptions; 
import io.vertx.reactivex.core.Vertx; 
import io.vertx.reactivex.core.buffer.Buffer; 
import io.vertx.reactivex.core.file.AsyncFile; 
import io.vertx.reactivex.core.streams.Pump; 

public class PumpTest { 

    @Test 
    public void testPump() throws IOException { 
     Vertx vertx = Vertx.vertx(); 
     File file = new File("target/test"); 
     file.delete(); 

     Observable<Buffer> rs = Observable.just(Buffer.buffer("test123")); 
     vertx.fileSystem().rxOpen(file.getAbsolutePath(), new OpenOptions()).map(f -> { 
      Pump pump = Pump.pump(rs, f); 
      pump.start(); 
      return f; 
     }).map(AsyncFile::flush).subscribe(AsyncFile::close); 
     assertEquals("test123", FileUtils.readFileToString(file)); 
    } 
} 

Answer 1

bei der Dokumentation der Suche von ReadStream<T> (ersten Parameter von Pump.pump) verweist auf die ReadStreamSubscriber Klasse. Was aus einem Observablen einen Readstream machen kann.

Anstelle von rs sollte ReadStreamSubscriber.asReadStream(rs, Function.identity()) in dem bereitgestellten Code funktionieren.