rewrite OnSubscribeRefCount to handle synchronous source

Resubmitting this one having stuffed up the rebase.

Rewrite of ```OnSubscribeRefCount``` because original didn't handle synchronous source (#1688) and as far as I could see the original was fundamentally broken by this use case. Please note also includes a fix in ```OperatorMulticast``` where a null check was required. I've done the rewrite using locks rather than submit a lock-free version just to ensure that the approach is valid with the experts first.
No because connect is called with an action that allows the subscription to
be read and in our case recorded.  The action also unlocks.
On 3 Oct 2014 11:38, "Ben Christensen" <notifications@github.com> wrote:

> In
> rxjava-core/src/main/java/rx/internal/operators/OnSubscribeRefCount.java:
>
> > -                queue = null;
> > -                if (localQueue == null) {
> > -                    emitting = false;
> > -                    return;
> > +    public void call(final Subscriber<? super T> subscriber) {
> > +
> > +        lock.lock();
> > +        if (subscriptionCount.incrementAndGet() == 1) {
> > +
> > +            final AtomicBoolean writeLocked = new AtomicBoolean(true);
> > +
> > +            try {
> > +                // need to use this overload of connect to ensure that
> > +                // baseSubscription is set in the case that source is a
> > +                // synchronous Observable
> > +                source.connect(onSubscribe(subscriber, writeLocked));
>
> Won't this block right here on a synchronous source and then deadlock any
> other subscriber trying to subscribe since the lock is now held?
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/ReactiveX/RxJava/pull/1695/files#r18372521>.
>