/**
  * Invoked when Observable.subscribe is called.
  */
 public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
     // cover for generics insanity
 }

/**
 * A one-argument action.
 */
public interface Action1<T> extends Action {
    void call(T t);
}

 /**
  * Creates an Observable with a Function to execute when it is subscribed to.
  * <p>
  * <em>Note:</em> Use {@link #create(OnSubscribe)} to create an Observable, instead of this constructor,
  * unless you specifically have a need for inheritance.
  * 
  * @param f
  *            {@link OnSubscribe} to be executed when {@link #subscribe(Subscriber)} is called
  */
 protected Observable(OnSubscribe<T> f) {
     this.onSubscribe = f;
 }

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
// validate and proceed
  if (subscriber == null) {
      throw new IllegalArgumentException("observer can not be null");
  }
  if (observable.onSubscribe == null) {
      throw new IllegalStateException("onSubscribe function can not be null.");
      /*
       * the subscribe function can also be overridden but generally that's not the appropriate approach
       * so I won't mention that in the exception
       */
  }

  // new Subscriber so onStart it
  subscriber.onStart();

  /*
   * See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls
   * to user code from within an Observer"
   */
  // if not already wrapped
  if (!(subscriber instanceof SafeSubscriber)) {
      // assign to `observer` so we return the protected version
      subscriber = new SafeSubscriber<T>(subscriber);
  }

  // The code below is exactly the same an unsafeSubscribe but not used because it would 
  // add a significant depth to already huge call stacks.
  try {
      // allow the hook to intercept and/or decorate
      hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
      return hook.onSubscribeReturn(subscriber);
  } catch (Throwable e) {
      // special handling for certain Throwable/Error/Exception types
      Exceptions.throwIfFatal(e);
      // if an unhandled error occurs executing the onSubscribe we will propagate it
      try {
          subscriber.onError(hook.onSubscribeError(e));
      } catch (Throwable e2) {
          Exceptions.throwIfFatal(e2);
          // if this happens it means the onError itself failed (perhaps an invalid function implementation)
          // so we are unable to propagate the error correctly and will just throw
          RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
          // TODO could the hook be the cause of the error in the on error handling.
          hook.onSubscribeError(r);
          // TODO why aren't we throwing the hook's return value.
          throw r;
      }
      return Subscriptions.unsubscribed();
  }
}