title: rxjava and terminating streams
body: <p>I'm new to reactive programming using rxjava and after going through the simpler examples I'm now trying to figure out how to work with continuous streams. The problem I have with the example below is that the program doesn't terminate after I've taken the 3 elements. My assumption is that I somehow need to unsubscribe to my observable but I don't fully grasp how to terminate the while loop and make the program exit. </p>
  
  <p>I've come across the following post <a href="https://stackoverflow.com/questions/20663255/rxjava-terminating-infinite-streams">RxJava -- Terminating Infinite Streams</a> but I still can't figure out what I'm missing.</p>
  
  <pre><code>class MyTwitterDataProvider {
  /*
  This example is written in Groovy
  
  Instance variables and constructor omitted
  */
  
  public Observable&lt;String&gt; getTweets() {
      BufferedReader reader = new BufferedReader(new InputStreamReader(getTwitterStream()))
  
      Observable.create({ observer -&gt;
          executor.execute(new Runnable() {
              def void run() {
                  String newLine
                  while ((newLine = reader.readLine()) != null) {
                      System.out.println("printing tweet: $newLine")
                      observer.onNext(newLine)
                  }
  
                  observer.onCompleted()
              }
          })
      })
  }
  
  def InputStream getTwitterStream() {
  // code omitted
  }
  
  public static void main (String [] args) {
      MyTwitterDataProvider provider = new MyTwitterDataProvider()
      Observable&lt;String&gt; myTweetsObservable = provider.getTweets().take(3)
  
      Subscription myTweetSubscription = myTweetsObservable.subscribe({tweet-&gt; println("client prints: $tweet")})
     // myTweetSubscription.unsubscribe()
  }
  }
  </code></pre>