RxJS 检测 observable 何时被订阅

Question

我需要检测一个可观察对象 (observedEvents) 何时被订阅，然后再订阅另一个可观察对象 (triggerEvent)。我不想手动订阅 triggerEvent，但只希望在 observedEvents 有订阅时订阅一次。

这里有一些代码解释了我在寻找什么：

// This just emits events
let emitting = new EventEmitter();

// This is the main Observable which someone else might
// have access to
let observedEvents = Rx.Observable.merge(
  Rx.Observable.fromEvent(emitting, 'aba'),
  Rx.Observable.fromEvent(emitting, 'bob')
)

// This trigger should get a subscription if observedEvents
// has one, i.e. when I subscribe to observedEvents
// that subscription activates this trigger

// I have made an attempt at this by calling skipUntil
// this however skips one event, but I don't want that
let triggerEvent = Rx.Observable.merge(
  // these actions are things that can
  // happen when the trigger is active
  Rx.Observable.of('a').delay(200),
  Rx.Observable.of('b').delay(400),
  Rx.Observable.of('c').delay(600)
)
.skipUntil(observedEvents);

// Something else should be used to activate trigger
// I don't want to do this part manually
triggerEvent.subscribe(val => {
    console.log(`Do something fancy with ${val}`);
});

//----------------------------------------------------
// Somewhere else in the code...
//----------------------------------------------------
observedEvents.subscribe(evt => {
  console.log(`Some event: ${evt}`);
});
// At this point I want triggerEvent to become active
// because observedEvents has a subscription

setTimeout(() => {
  emitting.emit('bob', 'world');
  setTimeout(() => emitting.emit('aba', 'Whosebug!'), 500);
}, 200);

<!DOCTYPE html>
<html>
<head>
  <script src="https://npmcdn.com/@reactivex/rxjs@5.0.0-beta.7/dist/global/Rx.umd.js"></script>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/EventEmitter/5.1.0/EventEmitter.min.js"></script>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width">
  <title>JS Bin</title>
</head>
<body>

</body>
</html>

这可能吗？

我希望这能解释我要找的东西。

在我写这篇文章时，我在想 Subjects 的解决方案可能是我需要的。我不确定，但我只需要朝着正确的方向推动或可能的话找到解决方案。

Answer 1

果然我使用主题是正确的。关键是 Subject 的观察者列表。这是我最后做的：

let emitting = new EventEmitter();
let sub = new Rx.Subject();

// return this to users
let myGlobalSub = sub.merge(Rx.Observable.of(1, 2, 3));

// For internal use
let myObservers = Rx.Observable.fromEvent(emitting, 'evt');

console.log(`The number of subscribers is ${sub.observers.length}`);

// Only do something if myGlobalSub has subscribers
myObservers.subscribe(l => {
  if (sub.observers.length) { // here we check observers
    console.log(l);
  }
});

// Somewhere in the code...
emitting.emit('evt', "I don't want to see this"); // No output because no subscribers

myGlobalSub.subscribe(l => console.log(l)); // One sub

emitting.emit('evt', 'I want to see this'); // Output because of one sub

console.log(`The number of subscribers is ${sub.observers.length}`);

<!DOCTYPE html>
<html lang=en>

<head>
  <script src="https://unpkg.com/rxjs@5.5.11/bundles/Rx.min.js"></script>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/EventEmitter/5.2.5/EventEmitter.min.js"></script>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width">
  <title>JS Bin</title>
</head>

<body>
</body>

</html>

Answer 2

@smac89 的回答使用了 observers属性，它已在 RxJS 7 中弃用，并计划在版本 8 中删除。

如果您只想知道一个 Observable 是否已被订阅，您可以使用 observed 布尔值属性。

目前没有与观察者数组真正等效的东西。

let emitting = new EventEmitter();
let sub = new rxjs.Subject();

// return this to users
let myGlobalSub = rxjs.merge(sub,rxjs.of(1, 2, 3));

// For internal use
let myObservers = rxjs.fromEvent(emitting, 'evt');

console.log(`Has the subject been subscribed to ? ${sub.observed}`);

// Only do something if myGlobalSub has subscribers
myObservers.subscribe(l => {
  if (sub.observed) { // here we check observers
    console.log(l);
  }
});

// Somewhere in the code...
emitting.emit('evt', "I don't want to see this"); // No output because no subscribers

myGlobalSub.subscribe(l => console.log(l)); // One sub

emitting.emit('evt', 'I want to see this'); // Output because of one sub

console.log(`Has the subject been subscribed to ? ${sub.observed}`);

<!DOCTYPE html>
<html lang=en>

<head>
  <script src="https://unpkg.com/rxjs@%5E7/dist/bundles/rxjs.umd.min.js"></script>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/EventEmitter/5.2.5/EventEmitter.min.js"></script>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width">
  <title>JS Bin</title>
</head>

<body>
</body>

</html>

RxJS 检测 observable 何时被订阅

RxJS detect when observable has been subscribed to

javascript

observable

ecmascript-6

reactivex