Rxjs: Chain differents observable implementations without loose terminal implementation

[hhfrancois]

Feature Request

new flatMap or similar operator than keep terminal observable type

Is your feature request related to a problem? Please describe.
When I use websocket, I need to build url from observable process. (get sessionId from server)
But, IO cant to chain observable and WebSocketSubject without loose the WebSocketSubject implementaion

(typescript syntax)

initWs() {
   this.wsSubject: WebSocketSubject<string> = this.getSomething().pipe(
      flatMap((value: any) => {
          // .... 
          return websocket<string>(url); // return WebSocketSubject<string> 
      }
   )
)

getSomething(): Observable<any>{
// ...
}
// this is not correct wsSubject is Observable<string>.
// so impossible to call this.wsSubject.next(messageToServer);

Describe the solution you'd like
When you chain flatMap, I expect to get new implementation of observable

Describe alternatives you've considered
Use a native ws impl

(If this is new operator request) describe reason it should be core operator
Until I meet the problem I expected that flatMap works like it

Additional context
It is usefull for websocket case, but maybe in other case

[luillyfe]

Since wsSubject is a Subject we can take advantage of it and use his asObservable method; Recall, asObservable creates a new Observable with this Subject as the source. Then, in order to send messages you can still using your wsSubject.

Observables are not made to send data, instead you need to use an Observer/Subject.

import { webSocket } from 'rxjs/webSocket';

const wsSubject = webSocket('ws://localhost:8081');
const wsAsObservable$ = wsSubject.asObservable();

this.getSomething().pipe(
    flatMap((value: any) => {
        // ....
        return wsAsObservable$;
    }
);

// Keep using your `wsSubject` Send messages to the Server
wsSubject.next(messageToServer);

[hhfrancois]

I'm not sure you understood my problem, I need the result of the first call for register websocket. The first call result needed for compute the webSocket url.

Call server, pipe and flatmap return webSocket (from rxjs).
Sorry for short syntax, I write this from my Mobile.

[luillyfe]

@hhfrancois I am here to help. Let's make things clear for both of us. You said,

I need the result of the first call for register websocket

But since here we are just make some declarations:

const wsSubject = webSocket('ws://localhost:8081');
const wsAsObservable$ = wsSubject.asObservable();

const source$ = this.getSomething().pipe(
    flatMap((value: any) => {
        // ....
        return wsAsObservable$;
    }
);

Then you can still using wsSubject, wsAsObservable$ as you want. Let's say,

initWs() {
  //Register the webSocketSubject
  this.wsSubject: WebSocketSubject = wsSubject;
  //And the Observable
  this.wsAsObservable$: Observable = source$;
}

Or better:

initWs() {
  //Register the webSocketSubject
  this.wsSubject: WebSocketSubject = webSocket('ws://localhost:8081');
  //And the Observable
  this.wsAsObservable$: Observable = this.getSomething().pipe(
    flatMap((value: any) => {
        // ....
        return wsAsObservable$;
    }
);
}

Let me know how this works for you, 🤪!

[hhfrancois]

Thank for your help. Sorry, maybe I didn't explain correctly. Cause my bad English...

In fact I need to request server to get information for compute webSocket url. In your example, you define webSocket url before the famous call.

I need the result of the first call for compute the webSocket subscription.

Let ws = this.service.getSomething().pipe(
flatMap( (val: string) =>{
return webSocket(computeUrl(val));
})
)
ws.subscribe((wsMsg: any) => {
console.log('receive msg from backend', wsMsg);
});

....

ws.next('payload'); // send message to backend

But I loose the wsSubject implementation during flatMap ...
So I can't use it for send message.

[luillyfe]

OK I came out with this approach (use async fucntions) but remenber, async/await does not work well with Observables then, use it with caution. However, there is more than one way to subscribe to an Observable!

const initWs = async () => {
    let ws = () => null;
    await this.service.getSomething().forEach(url => ws = webSocket(url));

    ws.subscribe(response => {
        console.log(response);
    });
};

initWs();

The downside here is you have to wrap your webSocketSubject inside of an async func. @hhfrancois, let me know if this meet your requirements.

[hhfrancois]

Hi, thank every body for responses.
In fact I don't ask a solution, I have already one. this is rather a feature request
Just think that the flatMap function should be works like map function... Maybe it's not possible, it's just a request ...

For the async/ await solution... I dont like too the async/await... It's uggly, but thank a lot.

So an other example :

when you use map :

getAs(): A[] {...}
getBFromA(a: A): B {...}

getBs(): B[] {
  return this.getAs().map(( a: A) => {
     return this.getBFromA(a);
  })
}

In this case the map function change the type. right
Why flatMap not ? It's map function no ?

getObsA(): Observable<A> {...}

getSubjectBFromA(a: A): Subject<B> {...}

getSubjectBs(): Subject<B> { // error don't return Subject but observable instead the map function
  return this.getObsA().pipe(
     flatMap(( a: A) => {
       return this.getSubjectBFromA(a);
     })
  )
}

In this case the flatMap function return every time an observable. however I flatMap it in Subject...

The solution for my case is :

wsSubject: WebSocketSubject;

initWs() {
  this.getSomething().pipe(
    flatMap((value: any) => {
       wsSubject = websocket(this.computeUrl(value));
       return wsSubject;
    })
  ).subscribe((msg: any) => {
    // receive message
  })
}
// ....

sendMsg() {
  if(!!this.wsSubject) {
    this.wsSubject.next(msg);
  }
}

[luillyfe]

https://github.com/ReactiveX/rxjs/issues/4355
@hhfrancois what do you think ? it is the same thing you are asking for, isn't it ?

[callmez]

I think it's the same issue. @hhfrancois solution is not best way.
The way I think of is dynamic get url when WebsocketSubject has first subscribe.
here is example

const url = async () => { return someService.getWebsocketUrl(); };
const websocket$ = new WebsocketSubject(url);
websocket$.subscribe(...); // get url when  call websocket$.subscribe()

[luillyfe]

Since we do not have support for dynamic urls already, any proposal solution feels like a hack. But the important thing here is to made just one issue/feature request for this topic.

[cartant]

What this issue seeks is not possible. See this comment for an explanation.