Outer completable switch maps

[knighty]

I've been struggling with the best way to handle switchmaps where the ideal is to complete the switchmap when the outer completes. The behavior in rxjs is such that the switchmap only completes when both the inner and outer have. This makes a lot of sense where the outer observable is basically just a proxy around a single item, for example a fetch that then maps to another fetch based on some data from the first. Clearly it'd be very undesirable if the switchmap (or any of the other variants I should note) completed here because then you wouldn't be able to do anything downstream.

In the case of long running streams, however, I struggle to find many cases when I wouldn't want the inner observable to complete when the outer one does. I found some discussion of this here #2714 but the discussion didn't really seem to go anywhere. The quintessential way of dealing with this kind of problem is a takeUntil that comes after all of the possible switchMaps, but this seems very unideal to me for many situations. A pattern I've been using a lot is for something like a WebSocket having an Observable of the client which things subscribe to then have switchMaps do things like subscribing to messages from the socket. Then when the socket closes, the complete event fires and downstream events can unsubscribe. This, however, will not complete anything that uses switchMaps downstream and doing this with takeUntils requires a proxy subject that gets a value poked into it when the socket completes. This feels very messy to me, and not to mention if you use any of these observables from the socket further downstream, you need to pass around this disconnection subject so that can takeUntil as well.

Example of usage:

socket.ts

const disconnected$ = new Subject<void>();
const client$ = new Observable<WebSocket>(subscriber => {
    const ws = new WebSocket(url);
    function error(e: Error) {
        subscriber.error(e);
    };
    function open() {
        subscriber.next(ws);
    }
    function close(code: number, reason: Buffer) {
        subscriber.complete();
        disconnected$.next();
    }
    ws.addListener("error", error);
    ws.addListener("open", open);
    ws.addListener("close", close);
    return () => {
        ws.removeListener("error", error);
        ws.removeListener("open", open);
        ws.removeListener("close", close);
    }
});

some-other-file.ts

client$.pipe(
    switchMap(client => fromEvent(client, "message")) // This will stayed subscribed when the client is closed
    takeUntil(disconnected$) // This is needed with vanilla switchmaps, which requires passing around this subject everywhere
)

In this particular example we could also have a takeUntil on the fromEvent that hooks into the close event on the client but that's just more boilerplate and only works here because we happen to be doing a switchMap off the client so we have that object.

This seems like such a common problem to me with streams that it makes me feel like I must be doing something inherently wrong with streams since it isn't talked about more and doesn't have any first class support for. I've made my own switchMapComplete operator which does the behavior I want but that feels very irksome to me because I don't want to mess around with some core behavior.

I would love some feedback on whether any of this makes sense or not.