如何在Angular 2中使用RXJS观察自定义事件？

Question

我有一个第三方库，我打算与RxJS集成。这是一个名为Tiger Text的消息库。据他们说，我可以听一个叫做消息的事件，当流有消息时我可以用它来进一步利用它。相同的代码片段如下： -

var client = new TigerConnect.Client({ defaultOrganizationId: 'some-org-id' })

client.signIn('[email protected]', 's3cr3t', { udid: 'unique-device-id' }).then(function (session) {
  onSignedIn(session)
})

function onSignedIn(session) {
  console.log('Signed in as', session.user.displayName)

  client.messages.sendToUser(
    '[email protected]',
    'hello!'
  ).then(function (message) {
    console.log('sent', message.body, 'to', message.recipient.displayName)
  })

  client.events.connect()

  client.on('message', function (message) {
    console.log(
      'message event',
      message.sender.displayName,
      'to',
      message.recipient.displayName,
      ':',
      message.body
    )
  })
}

现在请看看你有下面提到的代码的地方。

client.on('message', function (message) {
    console.log(
      'message event',
      message.sender.displayName,
      'to',
      message.recipient.displayName,
      ':',
      message.body
    )
  })

我想知道如何使用RxJS以便从这段代码中创建一个可观察的代码，以便订阅流，每当我们进行更改时，我都会采用新数据并按照我的意愿处理它。

请指教。

Answer 1

对于这种用例，您通常不需要编写自定义Observable，只能使用Observable.create()。那么这取决于你是想要写冷或热观察。

对于冷Observable，您在订阅时创建值的生成器，并在取消订阅时关闭它：

Observable.create(obs => {
  var client = new TigerConnect.Client({ defaultOrganizationId: 'some-org-id' });
  client.signIn('[email protected]', 's3cr3t', { udid: 'unique-device-id' }).then(function (session) {
    onSignedIn(session);
  });

  client.on('message', function (message) {
    obs.next(...);
  });

  return () => {
    client.close(); // or whatever...
  };
});

或者如果你想编写一个热的Observable，生产者将在任何订阅上独立存在，只需添加/删除监听器：

var client = new TigerConnect.Client({ defaultOrganizationId: 'some-org-id' });
client.signIn('[email protected]', 's3cr3t', { udid: 'unique-device-id' }).then(function (session) {
  onSignedIn(session);
});

Observable.create(obs => {
  let listener = client.on('message', function (message) {
    obs.next(...);
  });

  () => {
    // remove the event listener somehow
    listener.remove();
  };
});

有时候你可以通过使用Subject来解决这个问题，但这通常比使用Observable.create()更复杂，因为那时你需要自己处理创建和拆除逻辑，并且主体也有内部状态。

这是一个与你非常相似的问题：

• Subscribe to a stream with RxJS and twitter-stream-api module

有关RxJS主要开发人员与您的问题相关主题的文章：

• https://medium.com/@benlesh/hot-vs-cold-observables-f8094ed53339
• https://medium.com/@benlesh/on-the-subject-of-subjects-in-rxjs-2b08b7198b93
• https://medium.com/@benlesh/rxjs-dont-unsubscribe-6753ed4fda87
• https://medium.com/@benlesh/learning-observable-by-building-observable-d5da57405d87

Answer 2

您可以使用fromEventPattern从自定义事件创建一个observable：

import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/fromEventPattern';

const messages = Observable.fromEventPattern(
  handler => client.on('message', handler),
  handler => client.off('message', handler)
);
messages.subscribe(message => console.log(message));

您传递给fromEventPattern函数，使用自定义API的添加和删除机制添加和删除事件处理程序。您没有将它包含在您的问题中，但我假设您使用的API实现了off方法。