Wie lesen richtig Flux <DataBuffer> und es zu einem einzelnen input konvertieren

Question

I WebClient und benutzerdefinierten BodyExtractor Klasse für meine Feder-Boot-Anwendung

WebClient webLCient = WebClient.create(); 
webClient.get() 
    .uri(url, params) 
    .accept(MediaType.APPLICATION.XML) 
    .exchange() 
    .flatMap(response -> { 
    return response.body(new BodyExtractor()); 
    }) 

BodyExtractor.java

@Override 
public Mono<T> extract(ClientHttpResponse response, BodyExtractor.Context context) { 
    Flux<DataBuffer> body = response.getBody(); 
    body.map(dataBuffer -> { 
    try { 
     JaxBContext jc = JaxBContext.newInstance(SomeClass.class); 
     Unmarshaller unmarshaller = jc.createUnmarshaller(); 

     return (T) unmarshaller.unmarshal(dataBuffer.asInputStream()) 
    } catch(Exception e){ 
     return null; 
    } 
    }).next(); 
} 

Above Code bin mit funktioniert mit geringer Nutzlast, aber nicht auf ein großes Problem, ich denke es liegt daran, dass ich nur einen einzigen Fluss Wert mit next lesen und ich bin mir nicht sicher, wie zu kombinieren und alle dataBuffer lesen.

Ich bin neu in Reaktor, so dass ich nicht viele Tricks mit Flussmittel/Mono.

Answer 1

konnte ich es durch die Verwendung Flux#collect und SequenceInputStream

@Override 
public Mono<T> extract(ClientHttpResponse response, BodyExtractor.Context context) { 
    Flux<DataBuffer> body = response.getBody(); 
    return body.collect(InputStreamCollector::new, (t, dataBuffer)-> t.collectInputStream(dataBuffer.asInputStream)) 
    .map(inputStream -> { 
     try { 
     JaxBContext jc = JaxBContext.newInstance(SomeClass.class); 
     Unmarshaller unmarshaller = jc.createUnmarshaller(); 

     return (T) unmarshaller.unmarshal(inputStream); 
     } catch(Exception e){ 
     return null; 
     } 
    }).next(); 
} 

InputStreamCollector.java

public class InputStreamCollector { 
    private InputStream is; 

    public void collectInputStream(InputStream is) { 
    if (this.is == null) this.is = is; 
    this.is = new SequenceInputStream(this.is, is); 
    } 

    public InputStream getInputStream() { 
    return this.is; 
    } 
} 

Answer 2

Rekonstruieren der InputStream Niederlagen den Zweck der WebClient in erster Linie mit machen zu arbeiten, weil nichts, bis emittiert wird, Die Operation collect ist abgeschlossen. Für einen großen Strom kann das eine sehr lange Zeit sein. Das reaktive Modell behandelt nicht einzelne Bytes, sondern Blöcke von Bytes (wie Spring DataBuffer). Siehe meine Antwort hier für eine elegantere Lösung: https://stackoverflow.com/a/48054615/839733