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<Integer> subject1 = Observable.just(10);
        Observable<Integer> subject2 = Observable.just(20);
        Observable<Integer> 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<Integer> o1 = Observable.just(10);
        Observable<Integer> o2 = Observable.just(20);
        Observable<Integer> combined = Observable.combineLatest(o1, o2, new Func2<Integer, Integer, Integer>() {

            @Override
            public Integer call(Integer t1, Integer t2) {
                return t1 + t2;
            }
            
        }).publish().refCount();
        
        TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>();
        TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>();

        combined.subscribe(ts1);
        combined.subscribe(ts2);
        
        ts1.assertTerminalEvent();
        ts1.assertNoErrors();
        ts1.assertReceivedOnNext(Arrays.asList(30));
        
        ts2.assertTerminalEvent();
        ts2.assertNoErrors();
        ts2.assertReceivedOnNext(Arrays.asList(30));
    }
```