@Test public void testIOClose() throws Exception { K8LogReader k8LogReader = new K8LogReader("podNameFoo", logLineAccumulator, null); LogWatch logWatch = mock(LogWatch.class); ByteArrayOutputStream originalByteArrayOutputStream = new ByteArrayOutputStream(1000); PipedInputStream in = new PipedInputStream(); final PipedOutputStream out = new PipedOutputStream(in); when(logWatch.getOutput()).thenReturn(in); k8LogReader.observeRuntimeLog(logWatch); originalByteArrayOutputStream.write("Hey!\n".getBytes()); new Thread(new Runnable() { public void run () { try { originalByteArrayOutputStream.writeTo(out); } catch (IOException e) { throw new RuntimeException(e); } } }).start(); originalByteArrayOutputStream.writeTo(out); Thread.sleep(1000); }
public Observable<String> observeRuntimeLog(LogWatch logWatch) { logger.debug("Starting observable for {}", podName); if(subscriptionScheduler == null && observerScheduler == null) { subscriptionScheduler = Schedulers.newThread(); observerScheduler = Schedulers.newThread(); } this.logWatch = logWatch; Action1<Throwable> onError = new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("IO issue in stream [{}] as {}", throwable.getMessage(), throwable.getClass()); subscription.unsubscribe(); } }; stringObservable = StringObservable.byLine( StringObservable.from( new InputStreamReader(logWatch.getOutput())).doOnError(onError) ).observeOn(observerScheduler, false); subscription = stringObservable.subscribeOn(subscriptionScheduler) .subscribe(logLineAccumulator::appendLine); return stringObservable; }
K8LogReader { public Observable<String> observeRuntimeLog(LogWatch logWatch) { logger.debug("Starting observable for {}", podName); if(subscriptionScheduler == null && observerScheduler == null) { subscriptionScheduler = Schedulers.newThread(); observerScheduler = Schedulers.newThread(); } this.logWatch = logWatch; Action1<Throwable> onError = new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("IO issue in stream [{}] as {}", throwable.getMessage(), throwable.getClass()); subscription.unsubscribe(); } }; stringObservable = StringObservable.byLine( StringObservable.from( new InputStreamReader(logWatch.getOutput())).doOnError(onError) ).observeOn(observerScheduler, false); subscription = stringObservable.subscribeOn(subscriptionScheduler) .subscribe(logLineAccumulator::appendLine); return stringObservable; } }
K8LogReader { public Observable<String> observeRuntimeLog(LogWatch logWatch) { logger.debug("Starting observable for {}", podName); if(subscriptionScheduler == null && observerScheduler == null) { subscriptionScheduler = Schedulers.newThread(); observerScheduler = Schedulers.newThread(); } this.logWatch = logWatch; Action1<Throwable> onError = new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("IO issue in stream [{}] as {}", throwable.getMessage(), throwable.getClass()); subscription.unsubscribe(); } }; stringObservable = StringObservable.byLine( StringObservable.from( new InputStreamReader(logWatch.getOutput())).doOnError(onError) ).observeOn(observerScheduler, false); subscription = stringObservable.subscribeOn(subscriptionScheduler) .subscribe(logLineAccumulator::appendLine); return stringObservable; } K8LogReader(String podName, LogLineAccumulator logLineAccumulator, ConnectionTracker connectionTracker); }
K8LogReader { public Observable<String> observeRuntimeLog(LogWatch logWatch) { logger.debug("Starting observable for {}", podName); if(subscriptionScheduler == null && observerScheduler == null) { subscriptionScheduler = Schedulers.newThread(); observerScheduler = Schedulers.newThread(); } this.logWatch = logWatch; Action1<Throwable> onError = new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("IO issue in stream [{}] as {}", throwable.getMessage(), throwable.getClass()); subscription.unsubscribe(); } }; stringObservable = StringObservable.byLine( StringObservable.from( new InputStreamReader(logWatch.getOutput())).doOnError(onError) ).observeOn(observerScheduler, false); subscription = stringObservable.subscribeOn(subscriptionScheduler) .subscribe(logLineAccumulator::appendLine); return stringObservable; } K8LogReader(String podName, LogLineAccumulator logLineAccumulator, ConnectionTracker connectionTracker); Observable<String> observeRuntimeLog(LogWatch logWatch); void shutdown(); String getPodName(); void setPodName(String podName); }
K8LogReader { public Observable<String> observeRuntimeLog(LogWatch logWatch) { logger.debug("Starting observable for {}", podName); if(subscriptionScheduler == null && observerScheduler == null) { subscriptionScheduler = Schedulers.newThread(); observerScheduler = Schedulers.newThread(); } this.logWatch = logWatch; Action1<Throwable> onError = new Action1<Throwable>() { @Override public void call(Throwable throwable) { logger.warn("IO issue in stream [{}] as {}", throwable.getMessage(), throwable.getClass()); subscription.unsubscribe(); } }; stringObservable = StringObservable.byLine( StringObservable.from( new InputStreamReader(logWatch.getOutput())).doOnError(onError) ).observeOn(observerScheduler, false); subscription = stringObservable.subscribeOn(subscriptionScheduler) .subscribe(logLineAccumulator::appendLine); return stringObservable; } K8LogReader(String podName, LogLineAccumulator logLineAccumulator, ConnectionTracker connectionTracker); Observable<String> observeRuntimeLog(LogWatch logWatch); void shutdown(); String getPodName(); void setPodName(String podName);  }