Rxjs: subscribe ignores second argument of event emitter

[benjaminhon]

I am using redis, which exposes the event emitter as follows with 2 arguments, without rxjs, i can get both arguments, however with rxjs, the second argument is missing.

r.on 'message', (channel, message) ->
  console.log channel, message

# => Greet, Hello

Rx.Observable.fromEvent(r, 'message').subscribe (channel, message) ->
  console.log channel, message

# => Greet, undefined

is this supposed to be the case?

[mpodlasin]

That is how it works at the moment. You can get access to the second parameter by using project function:

Rx.Observable.fromEvent(r, 'message', (channel, message) => [channel, message])
.subscribe(arr => {
  console.log(arr[0]); // logs value of channel
  console.log(arr[1]); // logs value of message
});

That being said it seems like a strong case to make fromEvent behave like bindCallback which emits an array if there are many arguments in callback.

[mpodlasin]

Does redis have an off function for canceling subscriptions? How do you stop listening to events?

[benjaminhon]

yea it has,

sub.unsubscribe()
On Wed, Mar 1, 2017 at 11:39 PM, Mateusz Podlasin [e-mail]
wrote:

Does redis have an off function for canceling subscriptions? How do you
stop listening to events?

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
https://github.com/ReactiveX/rxjs/issues/2431#issuecomment-283375053,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AHzqCEyEqj6pW3IYWbeXzoOjn2xhO0jbks5rhZEcgaJpZM4MPszu
.

[mpodlasin]

That's weird. If there is no off (I literally mean method that is named "off") function on redis, then fromEvent should throw an Error, since it is not supported...

[trxcllnt]

We should probably update fromEvent to work with EventEmitters, as I constantly have this problem too. From what I remember, EventEmitters have on and removeListener methods. Does that sound right?

[kwonoj]

, EventEmitters have on and removeListener methods.

doesn't node evemtEmitter works already via addListener (https://nodejs.org/api/events.html#events_emitter_addlistener_eventname_listener) /removeListener ?(https://github.com/ReactiveX/rxjs/blob/7c41c08119cecb0c039ed40b2e1f3a0e6ac0d56e/src/observable/FromEventObservable.ts#L10-L13)

[benjaminhon]

@Podlas29 that does the job

[kwonoj]

I'm not sure where this issue has landed - is there specific we need to resolve in this issue?

[felixfbecker]

It would probably be a more ergonomic API to default to an array if multiple args are provided, but it would also be a breaking change. At the same time, the workaround is very concise:

Observable.fromEvent(r, 'message', Array.of)
  .subscribe(([channel, message]) => {
    // ...
  })

[felixfbecker]

or you could even create an object with lodash:
```ts
Observable.from(r, 'message', partial(zipObject, ['channel', 'message']))
.subscribe(event => {
console.log(event.channel)
console.log(event.message)
})

[benjaminhon]

zeromq for node alse use multipart replies which requires the above

[raycarter]

A simple use case about this issue and more about #3048 and PR #3049:

const Observable = require('rxjs').Observable;
const Oboe = require('oboe');

// create oboe instance
const oboe = Oboe({
  url: 'http://test-streaming.shengsoft.net',
  method: 'GET'
});

// original usage of oboe node event
oboe.on('node', '!.*', function() {
  console.log('from original oboe node event callback');
  console.log(arguments); // here 3 arguments are passed in
  console.log('');
});

// create another oboe instance
const oboe1 = Oboe({
  url: 'http://test-streaming.shengsoft.net',
  method: 'GET'
});

// use fromEventPattern to convert oboe event in Observable
const oboeObs = Observable.fromEventPattern((handler) => {
  oboe1.on('node', '!.*', handler);
});

oboeObs.subscribe(function() {
  console.log('from Observable handler');
  console.log(arguments); // !!! only the first argument is passed in
  console.log('');
});

In this simple use case Observable combines Oboe to consume a json-streaming server. The server returns only a part of a whole json object every second. Observable is introduced to process the data returned from the json-streaming server. But it got only the first argument while the original Oboe event passed 3 arguments in event handler. I have created a simple repo for you to retry what I mean.

[benlesh]

This is no longer an issue. fromEvent will emit an array of arguments if it gets more than one argument back from the handler.