title: RxJava -- Terminating Infinite Streams
body: <p>I am exploring reactive programming and RxJava. It is fun, but I am stuck on a problem for which I cannot find an answer. My basic question: what is a reactive-appropriate way to terminate an otherwise infinitely-running Observable? I also welcome critiques and reactive best practices regarding my code.</p>
  
  <p>As an exercise, I am writing a log file tail utility. The stream of lines in the log file is represented by an <code>Observable&lt;String&gt;</code>. To get the <code>BufferedReader</code> to continue reading text that is added to the file, I ignore the usual <code>reader.readLine() == null</code> termination check and instead interpret it to mean that my thread should sleep and wait for more logger text.</p>
  
  <p>But while I can terminate the Observer using <code>takeUntil</code>, I need to find a clean way to terminate the otherwise infinitely-running file watcher. I can write my own <code>terminateWatcher</code> method/field, but that breaks the Observable/Observer encapsulation -- and I'd like to stay as strict to the reactive paradigm as possible.</p>
  
  <p>Here is the <code>Observable&lt;String&gt;</code> code:</p>
  
  <pre><code>public class FileWatcher implements OnSubscribeFunc&lt;String&gt; {
      private Path path = . . .;
  
      @Override
      // The &lt;? super String&gt; generic is pointless but required by the compiler
      public Subscription onSubscribe(Observer&lt;? super String&gt; observer) {
          try (BufferedReader reader = new BufferedReader(new FileReader(path.toFile()))) {
              String newLine = "";
              while (!Thread.interrupted()) {  // How do I terminate this reactively?
                  if ((newLine = reader.readLine()) != null)
                      observer.onNext(newLine);
                  else
                      try {
                          // Wait for more text
                          Thread.sleep(250);
                      } catch (InterruptedException e) {
                          Thread.currentThread().interrupt();
                      }
              }
              observer.onCompleted();
          } catch (Exception e) {
              observer.onError(e);
          }
  
          return null;  // Not sure what Subscription I should return
      }
  }
  </code></pre>
  
  <p>Here is the Observer code that prints the new lines as they come:</p>
  
  <pre><code>public static void main(String... args) {
      . . .
      Observable&lt;String&gt; lines = Observable.create(createWatcher(file));
      lines = lines.takeWhile(new Func1&lt;String, Boolean&gt;() {
          @Override
          public Boolean call(String line) {
              // Predicate for which to continue processing
              return !line.contains("shutdown");
          }
      }).subscribeOn(Schedulers.threadPoolForIO())
              .observeOn(Schedulers.currentThread());
      // Seems like I should use subscribeOn() and observeOn(), but they
      // make my tailer terminate without reading any text.
  
      Subscription subscription = lines.subscribe(new Action1&lt;String&gt;() {
          @Override
          public void call(String line) {
               System.out.printf("%20s\t%s\n", file, line);
          }
      });
  }
  </code></pre>
  
  <p>My two questions are:</p>
  
  <ol>
  <li>What is a reactive-consistent way to terminate an otherwise infinitely-running stream?</li>
  <li>What other mistakes in my code make you cry? :)</li>
  </ol>