OperatorZipIterable does not unsubscribe from source

[samuelgruetter]

OperatorZipIterable does not unsubscribe from its source, as illustrated by this example:

    Action1<String> printer = new Action1<String>() {
        @Override
        public void call(String t1) {
            System.out.println(t1);
        }
    };

    Func1<Integer, String> squareStr = new Func1<Integer, String>() {
        @Override
        public String call(Integer t1) {
            System.out.println("Omg I'm calculating so hard: " + t1 + "*" + t1 + "=" + (t1*t1));
            return " " + (t1*t1);
        }
    };

    @Test public void test() {
        Observable<Integer> o = Observable.from(1, 2, 3, 4, 5);
        Iterable<String> it = Arrays.asList("a", "b", "c", "d", "e");
        o.map(squareStr).zip(it, concat2Strings).take(2).subscribe(printer);
    }

which outputs

Omg I'm calculating so hard: 1*1=1
 1-a
Omg I'm calculating so hard: 2*2=4
 4-b
Omg I'm calculating so hard: 3*3=9
Omg I'm calculating so hard: 4*4=16
Omg I'm calculating so hard: 5*5=25

The last 3 lines should not be there.

[akarnokd]

The inner subscriber is not chained with the outer. I'll post a PR to fix the issue shortly.

[benjchristensen]

That Pr is merged, so is this fixed?

[samuelgruetter]

Yes, closing.