Отказаться от RxJS Observables

Question

У меня есть эти два объекта, и я хочу прекратить слушать их события. Я совершенно новичок в наблюдаемых и RxJS и просто пытаюсь работать с библиотекой Inquirer.

Вот RxJS API для справки: http://reactivex.io/rxjs/class/es6/Observable.js~Observable.html

Как я могу отказаться от этих типов наблюдаемых?

ConnectableObservable:

FilterObservable:

FilterObservable { 
    source: 
    RefCountObservable { 
    source: 
     ConnectableObservable { 
     source: [Object], 
     _connection: [Object], 
     _source: [Object], 
     _subject: [Object] }, 
    _count: 1, 
    _connectableSubscription: ConnectDisposable { _p: [Object], _s: [Object] } }, 
    predicate: [Function] } 

мне нужно отказаться от этих объектов:

'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' && 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' && e.key.ctrl); 
    }).share(), 

    normalizedDownKey: keypress.filter(function (e) { 
     return e.key.name === 'down' || e.key.name === 'j' || (e.key.name === 'n' && e.key.ctrl); 
    }).share(), 

    numberKey: keypress.filter(function (e) { 
     return e.value && '123456789'.indexOf(e.value) >= 0; 
    }).map(function (e) { 
     return Number(e.value); 
    }).share(), 

    spaceKey: keypress.filter(function (e) { 
     return e.key && e.key.name === 'space'; 
    }).share(), 

    aKey: keypress.filter(function (e) { 
     return e.key && e.key.name === 'a'; 
    }).share(), 

    iKey: keypress.filter(function (e) { 
     return e.key && e.key.name === 'i'; 
    }).share() 
    }; 
}; 

Мой текущий догадка является то, что нет Явный вызов подписываться происходит так:

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'); 
    }); 

, но вместо этого, есть эти вызовы:

events.normalizedUpKey.takeUntil(validation.success).forEach(this.onUpKey.bind(this)); 
events.normalizedDownKey.takeUntil(validation.success).forEach(this.onDownKey.bind(this)); 

так что моя догадка, что я нужен способ, чтобы свести на нет/отменить вызов takeUntil.

Answer 1

Если вы хотите отказаться от подписки, вам необходимо иметь объект Subscription. Это объект, возвращаемый с каждого звонка Observable.subscribe(). Например:

let subscriber = Observable.subscribe(...); 
... 
subscriber.unsubscribe(); 

Для получения дополнительной информации см: https://github.com/ReactiveX/rxjs/blob/master/doc/subscription.md

Answer 2

я второй, что первый комментатор говорил. Однако он чувствует, что должен быть вызов, как этот где-то в коде:

let subscription = normalizedUpKey.subscribe(data => console.log('data')); 

, что вы могли бы сделать

subscription.unsubscribe() 

на. Как еще вы узнаете, когда что-то случится, или это часть сторонней библиотеки?

Подробнее о Rxjs. Я предлагаю вам ознакомиться с этой бесплатной книгой, https://www.gitbook.com/book/chrisnoring/rxjs-5-ultimate/details