class 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;
    }

    public  K8LogReader(String podName, LogLineAccumulator logLineAccumulator, ConnectionTracker connectionTracker);

}