Add OnSubscribeFunction and refactor to support it

This is related to ongoing work related to covariant support at https://github.com/Netflix/RxJava/pull/331