I made a lot of changes in the following section as I realized it wasn't capturing the `Future` from submitting to the pool and allowing Subscriptions to cancel the future if unsubscribed.

This in turn led me to need a `CompositeSubscribtion` since there are multiple things that can be unsubscribed in this flow.

Please review through this logic closely as the composite Subscriptions add complication as well as the fact that I have different logic for `Executor`, `ExecutorService` or `ScheduledExecutorService` implementations being passed in.