2.x replication refactoring

We should use doOnUnsubscribe here. When a stream from a forInterest() call is unsubscribed by the calling entity, we should decrement the removed interest's refCount until it is at zero, at which time we unsubscribe to it from the channel level (do the actual removeInterest at InterestChannelImpl).