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('user@mail.com', 's3cr3t', { udid: 'unique-device-id' }).then(function (session) {
  onSignedIn(session)
})

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

  client.messages.sendToUser(
    'someone@mail.com',
    '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.

It seems like you're using promises to check the data. Promise will only check the data once. You need to use Observable. If you need more details I can write as an answer. Let me know. — JP. K., Mar 29 '17 at 21:01
Its actually a third party library whose code I have written or pasted. So within the `client.on` method I wait for the data to come in an async manner and once there I just use the data. I dont know the internal mechanism of the library but will request if you can write and Observable. I am not sure how to :( — Shiv Kumar Ganesh, Mar 29 '17 at 21:04
I created a basic one for you with get request, I also wrote some comments for you because I don't know the size of your application , and provided a link because there are some stuffs you need to do on the component which they show it exactly how I would. — JP. K., Mar 29 '17 at 21:24

cartant · Answer 1 · 2019-05-01T19:32:53.540

You can use fromEventPattern to create an observable from a custom event:

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

You pass to fromEventPattern functions that add and remove the event handler using the custom API's add and remove mechanism. You've not included it in your question, but I've assumed the API you're using implements an off method.

score 5 · Answer 2 · edited Sep 23 '17 at 17:07

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('user@mail.com', '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('user@mail.com', '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:

I am so glad that you replied with links. And Yes the links are really useful to understand the same. Since the third party does not give me a handle to subscribe, writing the code in Angular 2 was a pain. Thanks — Shiv Kumar Ganesh, Mar 30 '17 at 20:23

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

2 Answers2