How to Observe a Custom Event using RXJS in Angular 2?

Question

I have a third party library that I am intending to integrate with RxJS. This is a messaging library called Tiger Text. According to them I can listen to an event called messages and when the stream has a message I can use it to further utilize it. The code snippet for the same is as follows:-

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

Now please have a look at the place where you have the below mentioned piece of code.

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

I wanted to know how to use RxJS so as to create an observable out of this piece of code so as to subscribe to the stream and whenever we have a change I take the new data and process it as I wish.

Please Advice.

Answer 1

For this use-cases you typically don't need to write a custom Observable and you can use just Observable.create(). Then it depends on whether you want to write a cold or a hot observable.

For cold Observables you create the producer of values when subscribing and close it when unsubscribing:

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

Or if you want to write a hot Observable the producer will exist independently on any subscriptions and just add/remove the listener:

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

Sometimes you can see this solved by using a Subject but this is usually more complicated than using Observable.create() because then you need to handle the creation and tear down logic yourself and also Subjects have internal state.

Here's a very similar question as yours:

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

Articles on the topics related to your question by the lead developer of 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