Publishing to a lifted Subject is type-unsafe

[masaeedu]

The lift method in Subject.ts looks like this:

  lift<T, R>(operator: Operator<T, R>): Observable<T> {
    const subject = new Subject(this, this.destination || this);
    subject.operator = operator;
    return <any>subject;
  }

The T parameter there is redundant, and causes a loss of type safety. The lift method on a Subject<Foo> should accept an Operator<Foo, ?>, not an Operator<?, ?>. A similar change for Observable.lift and Operator.call has been proposed in #1214.

A few other issues to consider:

• the signature specifies that it returns an Observable<T>. The new Subject it is returning has as its operator an Operator<T, R>, and will therefore behave as an Observable<R>
• the Observer functionality of subject is hidden and cannot be accessed, since the return type is Observable<T>
• the destination of the new Subject is set to this.destination. Assuming you took a Subject<number>, lifted it through .map(n => n.toString()), and tried nexting strings to the resulting Subject<string>, the destination of the original subject, which expects numbers, would be handed strings

Overall, I don't think this override is necessary. The behavior for lift provided in the Observable class (which is to return a new observable, set its source to this, and attach an operator), is the same as what is implemented here. The additional behavior of wrapping in a Subject that posts to this.destination is type-unsafe and unused (at least within the codebase).

[trxcllnt]

@masaeedu when Subject's lift returns a Subject instance, we can maintain the two-way pipeline through operator chaining for things like web sockets.

[masaeedu]

@trxcllnt Could you direct me to the code where this is done?

[trxcllnt]

@masaeedu everywhere in the Subject class that references destination.

[masaeedu]

Yes, I realize that, but my point was that you never (as far as I know), actually call next on the value produced by lifting Subject. Are there any places in the code where you lift a Subject<Something>, cast the returned Observable<SomethingElse> back to a Subject<SomethingElse>, and next values to it?

[masaeedu]

Note that I'm not talking about situations where you next values to the original Subject instance itself after producing a different lifted Subject. These scenarios would still work exactly the same way after removing this override.

[trxcllnt]

@masaeedu yes that does happen. I don't use TypeScript so the compiler doesn't complain at me, but I do next messages to the Subject instance returned from an operator.

[benlesh]

I think the complaint here is really that TypeScript doesn't support higher-kinded typing when it really should if its target is JavaScript. We could fix this if we were building our .d.ts files manually I suppose. Perhaps @david-driscoll knows a better way?

[masaeedu]

@trxcllnt So let's look at this scenario:

import {Subject} from 'rxjs/rx';

// Setup subject and a subscription
let numbers = new Subject<number>();
let rounded = numbers.map(Math.round);
rounded.subscribe(n => console.log(n));

// lift into something that produces, (and accepts?) strings
let strings = numbers.map(n => n.toString());
strings.subscribe(s => console.log(s));

// Push in some values
numbers.next(1.5);
strings.next("Hello");

As a user, especially a TypeScript user, it is confusing to see this result in:

2
1.5
NaN
Hello

This is not how Subjects behave in Rx.NET at least, and I would imagine the case is similar in RxJava or other implementations. Could you please point me to the code where this is being used so I can understand what use cases this enables?

[masaeedu]

@Blesh I don't think higher kinded types would make this better. In the example above, the subscription to rounded could be in one module, and numbers exported. That would compile nicely. Once my consuming code does:

import {numbers} from 'someModule'
numbers.map(n => n.toString()).next("Hello");

the existing module somehow needs to be recompiled with the awareness that downstream code is passing in strings. There is no way to represent this in a type system.

I guess this goes back to #541; maybe I'm just trying to project my static typing biases onto a project that just isn't suited for it 😄

[david-driscoll]

@Blesh without writing the types manually there would be no good way to do this.

As for the behavior, it seems fairly confusing. That is specifically why would a subscription to a subject be bound to another subject much deeper in the chain. I can understand it being useful for WebSockets, but it could also cause additional confusion later on.

I put @masaeedu's changes on requirebin and was confused, basically it's just forwarding all next calls up to the base numbers subject, which makes sense from the implementation, but as a consumer I'm really confused by that. Moreover how can I stop that from happening and strictly return just an Observable instead of a lifted Subject?

http://requirebin.com/?gist=01aa8e7ece1218b88318

[trxcllnt]

@masaeedu

var socket = Rx.Observable.websocket(socketURL);
var doesSomeStuff = socket.flatMap(...).filter(...);
someFunction(doesSomeStuff);

function someFunction(socket) {
  socket.subscribe((x) => {
    console.log('hello!' + x);
    socket.next(x + ' acknowledged');
  });
}

@david-driscoll Moreover how can I stop that from happening and strictly return just an Observable instead of a lifted Subject?

const obs = subject.asObservable();

[david-driscoll]

asObservable() isn't available today, which was what I was trying to point out. :)

[masaeedu]

@trxcllnt Wouldn't this work?

var socket = Rx.Observable.websocket(socketURL); // This is a subject
var doesSomeStuff = socket.flatMap(...).filter(...); // From this point on, these are NOT subjects
someFunction(doesSomeStuff);

function someFunction(observable) {
  observable.subscribe((x) => {
    console.log('hello!' + x);
    socket.next(x + ' acknowledged');
  });
}

Why do you need to post back to the mapped, filtered etc. subject? It's not like the values you're nexting in the subscribed handler are being passed through a "reversed" chain of inverse operators; they're just being naively propagated from one subject to the next until they get back to the original subject.

[masaeedu]

@trxcllnt Note that the above scenario would be enabled by simply removing the lift override. I had a question about the code you posted though; how are you preventing infinite loops?

[trxcllnt]

@masaeedu yes, but you don't have to keep a reference to the original socket. if that chain can be maintained through the Subject implementation, I haven't heard a good why we wouldn't. Infinite loops don't occur for the same reason they don't occur today in RxJS 4 when you Subject.create(observer, observable);

[david-driscoll]

Infinite loop is easy, not with a WebSocket because it's async nature.

var infiniteLoop = new Rx.Subject();

var self = infiniteLoop.map(function(x) { return x+1 });
self.subscribe(function(i) {
  infiniteLoop.next(i++);
});
self.next(0);

http://jsbin.com/wewadug/1/edit?js,console

[masaeedu]

@trxcllnt Well the reason is that it breaks type safety for us picky TypeScript users 😄. I don't think it's so bad to have to keep a reference to a websocket.

Regarding the infinite loop, wouldn't posting to the websocket result in the subscribed handler being called, which would post a value to the websocket, which would call the handler and so forth? I'm just asking to check if there's any special message routing machinery that would be broken by the change I'm proposing.

[trxcllnt]

@david-driscoll that's always been possible with Subjects (minus the Subject out of the lift override, but no existing code currently calls onNext on the object returned from map). The only way to guard against re-entrancy is by introducing potentially unbounded buffers.

@masaeedu I don't think it's so bad to have to keep a reference to a websocket.

Agree to disagree.

Regarding the infinite loop, wouldn't posting to the websocket result in the subscribed handler being called, which would post a value to the websocket, which would call the handler and so forth? I'm just asking to check if there's any special message routing machinery that would be broken by the change I'm proposing.

Perhaps in a naive implementation. My use-case is in classes that extend Subject, and I need my sub-types available so I can keep calling my custom operators. If we have to change the type signature of Subject from Subject<T> to Subject<T, R>, we can do that.

[masaeedu]

Agree to disagree.

Could you elaborate on why it's worse to have a direct reference to the websocket than to have the chain of this.destination references through the instances of Subject that are created by applying operators?

Perhaps in a naive implementation

This is what I was trying to check. Would it not be possible to change whatever the implementation does to avoid the infinite loop encountered by a naive implementation, so that it still works without the this.destination chain?

If we have to change the type signature of Subject from Subject<T> to Subject<T, R>, we can do that

This would be a useful change for Subscriber and Subject for several reasons, but I don't see how it would help here.

[masaeedu]

that's always been possible with Subjects (minus the Subject out of the lift override, but no existing code currently calls onNext on the object returned from map)

@trxcllnt But this is the use case we're talking about, no? The situation where next is called on the return value of an operator. Or do you mean map specifically?

[trxcllnt]

Could you elaborate on why it's worse to have a direct reference to the websocket

Because the WebSocket may or may not actually exist if, for example, it is lazily created and destroyed based on ref-counting.

than to have the chain of this.destination references

This isn't a linked-list of references to destination. The intermediate Subjects share the same reference to destination because lift calls the Subject constructor with this.destination first:

  lift<T, R>(operator: Operator<T, R>): Observable<T> {
    const subject = new Subject(this, this.destination || this); // <---
    subject.operator = operator;
    return <any>subject;
  }

I can sympathize with the argument that destination shouldn't fallback to this.

[masaeedu]

Ah yes, that's a good point regarding this.destination being preferred. I see. Nevertheless, I don't see that a direct reference to the original subject in the subscriber of a transformed observable keeps it alive any longer than the reference created via passing along this.destination.

[masaeedu]

Could you provide an example where using a direct reference to the original subject behaves differently from using the last result in an operator chain?

[trxcllnt]

@masaeedu there is no difference between the two aside from convenience.

[masaeedu]

Ok, so just to be clear, the reason the direct reference is worse is because it is inconvenient, not because of a difference in ref counting, correct?

As a compromise, maybe you could implement the lift override at the websocket subject level only, and let us remove the override from Subject?

[trxcllnt]

@masaeedu the advantage of implementing two-way communication in the Subject base class is that any Subject implementation or subclass is two-way compatible. I suppose we could remove the lift override and keep the next, error, complete, and subscribe support for Subject-chaining, though I'd much rather fix the type signature than remove the override because it'll fix the type signature for everything that extends Subject as well.

[masaeedu]

@trxcllnt What would the fixed type signature look like? The cognitive problem for me right now is that a Subject can receive values of any type whatsoever, given the appropriate transformations are applied before nexting, so I really can't see a way to make this type safe. There's also no precedent for this in other implementations of Rx, so I can draw no inspiration there.

[masaeedu]

Perhaps we could make a TwoWaySubject extends Subject or something like that, for classes where we want to base the implementation in this foundation.

[david-driscoll]

A fixed type signature, is any method that lifts from a Subject would really return a Subject.

With doing any major work on this yet we could do something like....

interface CoreOperators<T, O extends Observable<T>> {
  map(selector: any): O;
}

...

export class Observable<T> implements CoreOperators<T, Observable<T>> {...}

export class Subject<T> implements CoreOperators<T, Subject<T>> {...}

Problem is Subject inherits from Observable so how do we get Subject to inherit from Observable without having some intermediary class that doesn't implement CoreOperators I have no idea.

[trxcllnt]

@david-driscoll Observable is just a wrapper for the subscribe method, Subjects could mixin Observable's subscribe implementation the way it mixes in Subscriber's add, remove and unsubscribe implementations.

[masaeedu]

@david-driscoll IMO that's making the problem worse, because it's now exposing the API that causes type unsafety to TypeScript developers, who care about type safety the most.

Let's say the signature was lift<R>(op: Operator<T, R>): Subject<R>. We can actually change it to that without any problems, by doing the appropriate any casts in the body of the function to make this signature possible. Assuming we did, however, it would encourage TS users to write code like that first snippet you put in requirebin, and get confusing results where a string is passed to a callback that static typing indicates will only be passed numbers.

Right now they get a compile error on the last line, which I think mitigates the problem somewhat.

[masaeedu]

Summarizing the suggestions in this thread:

• Remove the override of lift (breaks code that nexts to the lifted Subject)
• Create a TwoWaySubject extends Subject<any> and override lift to return Subjects there (still breaks code that nexts to instances not deriving from this new class)
• Change signature to lift<R>(op: Operator<T, R>): Subject<R> and make corresponding change to CoreOperators (requires any casts in the implementation of lift, since the compiler recognizes you can't safely forward values from a Subject<R> to a Subject<T>)
• ???

Do any of these sound like a reasonable plan of action @trxcllnt, @Blesh?

[trxcllnt]

@masaeedu @david-driscoll does this look any better? I got it to where the compiler isn't yelling at me anymore:

class Subject<T, T_Back> extends Observable<T> implements Observer<T_Back>, Subscription {

    protected source: Observable<any>;
    protected destination: Observer<T_Back>;
    protected observers: Observer<T>[];

    constructor(destination?: Observer<T_Back>, source?: Observable<any>) {
        super();
        this.source = source;
        this.destination = destination;
    }

    lift<R>(operator: Operator<T, R>): Subject<R, T_Back> {
        const subject = new Subject<R, T_Back>(this.destination || this, this);
        subject.operator = operator;
        return subject;
    }

    next(value: T | T_Back): void {
        if (this.destination) {
            return this.destination.next(<T_Back> value);
        }
        const observers: Observer<T>[] = this.observers.concat();
        let index = -1;
        const len = observers.length;
        while (++index < len) {
            observers[index].next(<T> value);
        }
    }

    error(error: any): void {
        if (this.destination) {
            return this.destination.error(error);
        }
        const observers: Observer<T>[] = this.observers;
        this.observers = null;
        let index = -1;
        const len = observers.length;
        while (++index < len) {
            observers[index].error(error);
        }
    }

    complete(): void {
        if (this.destination) {
            return this.destination.complete();
        }
        const observers: Observer<T>[] = this.observers;
        this.observers = null;
        let index = -1;
        const len = observers.length;
        while (++index < len) {
            observers[index].complete();
        }
    }

    unsubscribe(): void {}
    add(subscription: Subscription| Function | void): void {}
    remove(subscription: Subscription): void {}
}

const numbers = new Subject<number, number>();
const strings = <Subject<string, number>> numbers.map<string>((num) => {
    return num.toString(2);
});

numbers.subscribe((num: number) {
    console.log(num);
});

strings.subscribe((str: string) {
    console.log(str);
});

numbers.next(5);
// prints
// 5
// "101"

strings.next(10);
// prints 
// 10
// "1010"

[david-driscoll]

I know I'm slow to respond, but that looks better assuming it compiles, it would be a breaking change for making subjects though (from the views of TS user anyway), so probably not very desirable.

In TypeScript it will probably still return Observable, but that should be okay. For this convince case it's probably simpler for the consumer to just typecast to Observer or Subject themselves.

[benlesh]

Closing this issue as stale.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.