title: Unsubscribe from RxJS Observables
body: <p>I have these two objects, and I want to stop listening to their events. I am totally new to observables and RxJS and just trying to work with the <a href="https://github.com/SBoudrias/Inquirer.js" rel="noreferrer">Inquirer</a> library.</p>
  
  <p>Here is the RxJS API for reference:
  <a href="http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html" rel="noreferrer">http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html</a></p>
  
  <p>How can I unsubscribe from these types of observables?</p>
  
  <p><strong>ConnectableObservable:</strong></p>
  
  <pre><code>   ConnectableObservable {
       source: EventPatternObservable { _add: [Function], _del: [Function], _fn: undefined },
       _connection: ConnectDisposable { _p: [Circular], _s: [Object] },
       _source: AnonymousObservable { source: [Object], __subscribe: [Function: subscribe] },
       _subject: 
        Subject {
          isDisposed: false,
          isStopped: false,
          observers: [Object],
          hasError: false } },
    _count: 1,
    _connectableSubscription: 
     ConnectDisposable {
       _p: 
        ConnectableObservable {
          source: [Object],
          _connection: [Circular],
          _source: [Object],
          _subject: [Object] },
       _s: AutoDetachObserver { isStopped: false, observer: [Object], m: [Object] } } }
  </code></pre>
  
  <p><strong>FilterObservable:</strong></p>
  
  <pre><code>FilterObservable {
    source: 
     RefCountObservable {
       source: 
        ConnectableObservable {
          source: [Object],
          _connection: [Object],
          _source: [Object],
          _subject: [Object] },
       _count: 1,
       _connectableSubscription: ConnectDisposable { _p: [Object], _s: [Object] } },
    predicate: [Function] }
  </code></pre>
  
  <p>I need to unsubscribe from these objects:</p>
  
  <pre><code>'use strict';
  var rx = require('rx');
  
  function normalizeKeypressEvents(value, key) {
    return {value: value, key: key || {}};
  }
  
  module.exports = function (rl) {
  
    var keypress = rx.Observable.fromEvent(rl.input, 'keypress', normalizeKeypressEvents)
      .filter(function (e) {
        // Ignore `enter` key. On the readline, we only care about the `line` event.
        return e.key.name !== 'enter' &amp;&amp; e.key.name !== 'return';
      });
  
    return {
      line: rx.Observable.fromEvent(rl, 'line'),
  
      keypress: keypress,
  
      normalizedLeftKey: keypress.filter(function (e) {
        return e.key.name === 'left';
      }).share(),
  
      normalizedRightKey: keypress.filter(function (e) {
        return e.key.name === 'right';
      }).share(),
  
      normalizedUpKey: keypress.filter(function (e) {
        return e.key.name === 'up' || e.key.name === 'k' || (e.key.name === 'p' &amp;&amp; e.key.ctrl);
      }).share(),
  
      normalizedDownKey: keypress.filter(function (e) {
        return e.key.name === 'down' || e.key.name === 'j' || (e.key.name === 'n' &amp;&amp; e.key.ctrl);
      }).share(),
  
      numberKey: keypress.filter(function (e) {
        return e.value &amp;&amp; '123456789'.indexOf(e.value) &gt;= 0;
      }).map(function (e) {
        return Number(e.value);
      }).share(),
  
      spaceKey: keypress.filter(function (e) {
        return e.key &amp;&amp; e.key.name === 'space';
      }).share(),
  
      aKey: keypress.filter(function (e) {
        return e.key &amp;&amp; e.key.name === 'a';
      }).share(),
  
      iKey: keypress.filter(function (e) {
        return e.key &amp;&amp; e.key.name === 'i';
      }).share()
    };
  };
  </code></pre>
  
  <p>My current best guess is that no explicit call to subscribe is happening like this:</p>
  
  <pre><code>var source = Rx.Observable.fromEvent(input, 'click');
  
  var subscription = source.subscribe(
    function (x) {
      console.log('Next: Clicked!');
    },
    function (err) {
      console.log('Error: %s', err);
    },
    function () {
      console.log('Completed');
    });
  </code></pre>
  
  <p>but instead, there are these calls:</p>
  
  <pre><code>events.normalizedUpKey.takeUntil(validation.success).forEach(this.onUpKey.bind(this));
  events.normalizedDownKey.takeUntil(validation.success).forEach(this.onDownKey.bind(this));
  </code></pre>
  
  <p>so my best guess is that I need a way to nullify/cancel the takeUntil call.</p>