如何與InputStreams的合併可流動的繼續，如果一個是封閉的，RxJava 2.X

我是新來RxJava2，如果我正確地使用它我不知道。如何與InputStreams的合併可流動的繼續，如果一個是封閉的，RxJava 2.X

我有一對夫婦是從TCP流中讀取數據InputStreams的，和我的數據讀入一個byte[]。我則走的byte[]和創造每一個InputStream和Flowable合併個人Flowable爲一體。

我希望能夠關閉1 InputStream但仍然合併Flowable繼續從非關閉InputStream讀取。

目前，我已經從2 InputStreams正確讀取，但是當我打電話給close時，它會導致一個例外，java.net.SocketException: Socket closed被拋出。

我的問題是，如何正確關閉InputStream或可流動的，這樣我可以繼續從其他的InputStream的平均時間讀取。

InputStream inputStream = response.body().asInputStream(); 
InputStream inputStream2 = response2.body().asInputStream(); 

Flowable<byte[]> flowable = Bytes.from(inputStream) 
           .distinct(); 

Flowable<byte[]> flowable2 = Bytes.from(inputStream2) 
            .distinct(); 

Flowable.merge(flowable.subscribeOn(Schedulers.newThread()), 
       flowable2.subscribeOn(Schedulers.newThread())) 
     .subscribe(s -> System.out.println("Data: " + new String(s)), 
      e -> System.out.println("Error: " + e.getLocalizedMessage() + " : " + e.getMessage()), 
      () -> System.out.println("Complete!")); 

//Calling this results in an exception being thrown. 
inputSteam2.close();

堆棧跟蹤：

io.reactivex.exceptions.UndeliverableException: java.net.SocketException: Socket closed 
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349) 
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.innerError(FlowableFlatMap.java:604) 
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onError(FlowableFlatMap.java:665) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onError(FlowableSubscribeOn.java:102) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.onError(FlowableGenerate.java:189) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:114) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.requestUpstream(FlowableSubscribeOn.java:133) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onSubscribe(FlowableSubscribeOn.java:90) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate.subscribeActual(FlowableGenerate.java:52) 
    at io.reactivex.Flowable.subscribe(Flowable.java:12986) 
    at io.reactivex.Flowable.subscribe(Flowable.java:12932) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:748) 
Caused by: java.net.SocketException: Socket closed 
    at java.net.SocketInputStream.socketRead0(Native Method) 
    at java.net.SocketInputStream.socketRead(SocketInputStream.java:116) 
    at java.net.SocketInputStream.read(SocketInputStream.java:171) 
    at java.net.SocketInputStream.read(SocketInputStream.java:141) 
    at sun.security.ssl.InputRecord.readFully(InputRecord.java:465) 
    at sun.security.ssl.InputRecord.read(InputRecord.java:503) 
    at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:973) 
    at sun.security.ssl.SSLSocketImpl.readDataRecord(SSLSocketImpl.java:930) 
    at sun.security.ssl.AppInputStream.read(AppInputStream.java:105) 
    at okio.Okio$2.read(Okio.java:140) 
    at okio.AsyncTimeout$2.read(AsyncTimeout.java:238) 
    at okio.RealBufferedSource.read(RealBufferedSource.java:45) 
    at okhttp3.internal.http.Http1xStream$UnknownLengthSource.read(Http1xStream.java:476) 
    at okio.RealBufferedSource$1.read(RealBufferedSource.java:386) 
    at java.io.InputStream.read(InputStream.java:101) 
    at com.github.davidmoten.rx2.Bytes$1.accept(Bytes.java:47) 
    at com.github.davidmoten.rx2.Bytes$1.accept(Bytes.java:43) 
    at io.reactivex.internal.operators.flowable.FlowableInternalHelper$SimpleGenerator.apply(FlowableInternalHelper.java:44) 
    at io.reactivex.internal.operators.flowable.FlowableInternalHelper$SimpleGenerator.apply(FlowableInternalHelper.java:35) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:109) 
    ... 14 more 
Exception in thread "RxNewThreadScheduler-3" io.reactivex.exceptions.UndeliverableException: java.net.SocketException: Socket closed 
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349) 
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$MergeSubscriber.innerError(FlowableFlatMap.java:604) 
    at io.reactivex.internal.operators.flowable.FlowableFlatMap$InnerSubscriber.onError(FlowableFlatMap.java:665) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onError(FlowableSubscribeOn.java:102) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.onError(FlowableGenerate.java:189) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate$GeneratorSubscription.request(FlowableGenerate.java:114) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.requestUpstream(FlowableSubscribeOn.java:133) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onSubscribe(FlowableSubscribeOn.java:90) 
    at io.reactivex.internal.operators.flowable.FlowableGenerate.subscribeActual(FlowableGenerate.java:52) 
    at io.reactivex.Flowable.subscribe(Flowable.java:12986) 
    at io.reactivex.Flowable.subscribe(Flowable.java:12932) 
    at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:748)

來源

2017-07-11 dchau

'Bytes.from'是如何實現的？你用'mergeDelayError'試過了嗎？ – akarnokd

我還沒有看到mergeDelayError，但我會研究這一點。 'Bytes.from'來自另一個[library]（https://github.com/davidmoten/rxjava2-extras/blob/master/src/main/java/com/github/davidmoten/rx2/Bytes.java#L42 ），它通過檢查流是否已經結束而生成流程圖。 – dchau

閱讀後'mergeDelayError'，我不認爲我想因爲當我關閉的InputStream，該異常會繼續存在，直到時間的一些未知的時期，其他InputStreams是開放的。 – dchau

我能夠通過創建我自己的操作和創造從它Flowable成功關閉流。

這個類是一個新的RxJava運營商實現並記錄例外，我從關閉一個InputStream，這對於我的使用情況是好的得到。

public class MyFlowableOperator implements FlowableOperator { 
    @Override 
    public Subscriber apply(Subscriber subscriber) throws Exception { 
    return new Subscriber() { 
     @Override 
     public void onSubscribe(Subscription s) { 
     subscriber.onSubscribe(s); 
     } 

     @Override 
     public void onNext(Object o) { 
     subscriber.onNext(o); 
     } 

     @Override 
     public void onError(Throwable t) { 
     if(t instanceof SocketException) { 
      log.debug("Input Stream Closed"); 
     } else { 
      subscriber.onError(t); 
     } 
     } 

     @Override 
     public void onComplete() { 
     subscriber.onComplete(); 
     } 
    }; 
    } 
}

下面是我如何申報RxJava運營商的一個新的實例，然後從InputStream

// Private method to utilize the new RxJava Operator 
private Subscriber getFlows(Subscriber subscriber) { 
try { 
    return new MyFlowableOperator().apply(subscriber); 
    } catch (Exception e) { 
    log.error(e.getMessage()); 
    } 
    return null; 
}

創建Flowable時，當創建Flowable本身利用它，我用lift方法叫我方法並返回一個帶有錯誤處理的Flowable。

// Declaration of the Flowables 
Flowable<byte[]> flowable = Bytes.from(inputStream) 
           .subscribeOn(Schedulers.newThread()) 
           .distinct() 
           .lift(this::getFlows); 

Flowable<byte[]> flowable2 = Bytes.from(inputStream2) 
            .subscribeOn(Schedulers.newThread()) 
            .distinct() 
            .lift(this::getFlows);

來源

2017-07-14 14:52:54 dchau

如何與InputStreams的合併可流動的繼續，如果一個是封閉的，RxJava 2.X

回答

相關問題