/** * Copyright 2016 Netflix, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package rx.internal.operators; import rx.Observable.*; import rx.Subscriber; import rx.exceptions.Exceptions; import rx.plugins.RxJavaHooks; /** * Transforms the downstream Subscriber into a Subscriber via an operator * callback and calls the parent OnSubscribe.call() method with it. * @param the source value type * @param the result value type */ public final class OnSubscribeLift implements OnSubscribe { final OnSubscribe parent; final Operator operator; public OnSubscribeLift(OnSubscribe parent, Operator operator) { this.parent = parent; this.operator = operator; } @Override public void call(Subscriber o) { try { Subscriber st = RxJavaHooks.onObservableLift(operator).call(o); try { // new Subscriber created and being subscribed with so 'onStart' it st.onStart(); parent.call(st); } catch (Throwable e) { // localized capture of errors rather than it skipping all operators // and ending up in the try/catch of the subscribe method which then // prevents onErrorResumeNext and other similar approaches to error handling Exceptions.throwIfFatal(e); st.onError(e); } } catch (Throwable e) { Exceptions.throwIfFatal(e); // if the lift function failed all we can do is pass the error to the final Subscriber // as we don't have the operator available to us o.onError(e); } } }