Netflix/RxJava — pull request #1515 by benjchristensen (2014-07-25 00:04:55+00:00)
Title: Support Subject Factory with Multicast

This is a proposed change to allow using a `Subject` factory with `multicast` and for methods such as `publish()` and `replay()` to use the factory so that each time `ConnectableObservable.connect()` is invoked, a new `Subject` is created and used. Otherwise, an `Observable` sequence can not be reused.

#### Use Case

```java
final AtomicInteger count = new AtomicInteger();
Observable subject1 = Observable.just(10);
Observable subject2 = Observable.just(20);
Observable combined = Observable.combineLatest(subject1, subject2, (fst, snd) -> {
int i = count.incrementAndGet();
System.out.println((fst + i) + " + " + (snd + i));
return fst + snd;
}).publish().refCount();

combined.subscribe(
System.out::println,
t -> t.printStackTrace(),
() -> System.out.println("Completed"));

combined.subscribe(
System.out::println,
t -> t.printStackTrace(),
() -> System.out.println("Completed"));
```

Output without these changes:

```
11 + 21
30
Completed
Completed
```

Output with changes:

```
11 + 21
30
Completed
12 + 22
30
Completed
```

#### Test Case

```java
@Test
public void testConnectDisconnectConnectAndSubjectState() {
Observable o1 = Observable.just(10);
Observable o2 = Observable.just(20);
Observable combined = Observable.combineLatest(o1, o2, new Func2() {

@Override
public Integer call(Integer t1, Integer t2) {
return t1 + t2;
}

}).publish().refCount();

TestSubscriber ts1 = new TestSubscriber();
TestSubscriber ts2 = new TestSubscriber();

combined.subscribe(ts1);
combined.subscribe(ts2);

ts1.assertTerminalEvent();
ts1.assertNoErrors();
ts1.assertReceivedOnNext(Arrays.asList(30));

ts2.assertTerminalEvent();
ts2.assertNoErrors();
ts2.assertReceivedOnNext(Arrays.asList(30));
}
```