Fixes to the operators.

[akarnokd]

Fixes for issue #1136

• OperatorMulticast is straightforward from concurrency perspective. The only consideration is that if the current subscription gets unsubscribed before the connect reaches the unsafeSubscribe, it really depends on the source what it will do with an unsubscribed client. It is possible to put extra effort to make sure a newly established connection won't get unsubscribed before it is actually connected or if it gets unsubscribed immediately, no subscription is attempted at all.
• OperatorSampleWithTime: didn't want to push too many PRs so I just included it here. There was a missing unsubscribe in the onCompleted() that makes sure the worker is stopped.
• Subscribers.empty() was implemented by returning the same Subscriber to everyone, which Subscriber is stateful so if someone unsubscribes it, it will appear everywhere as unsubscribed and can have unwanted effects. There is no such problem with Observers.empty() as it is stateless. The change just uses Subscribers.from() to wrap Observers.empty() and every caller gets its own independent instance.
• OperatorRefCount was a bit more tricky. Since it has a connection counter, one has to serialize subscriptions with unsubscriptions. However, it is possible a subscription gets unsubscribed before code reaches the connect check which may disrupt the connection counter. The solution is to keep track of the unsubscriptions that happen before the connection attempts and not change the counter in case of out-of-order behavior. The final aim was to avoid leaking the connection statuses if the unsafeSubscribe throws concurrently with a client unsubscribing by using weak tokens (integers wouldn't have worked as the first 0-127 are cached in the JVM and would never GC).

[cloudbees-pull-request-builder]

RxJava-pull-requests #1050 SUCCESS
This pull request looks good

[benjchristensen]

Thanks for tackling these complicated issues.