1.x: deanonymize Observable inner classes (#3848)

* 1.x: deanonymize Observable inner classes

* Further simplification of types

* Fix indentation, rename Lambda to Action

Author: akarnokd

src/main/java/rx/Observable.java

@@ -432,4 +432,27 @@ public R call(Object... args) {
             }
         };
     }
+    
+    /**
+     * Wraps an Action0 instance into an Action1 instance where the latter calls
+     * the former.
+     * @param action the action to call
+     * @return the new Action1 instance
+     */
+    public static <T> Action1<T> toAction1(Action0 action) {
+        return new Action1CallsAction0<T>(action);
+    }
+    
+    static final class Action1CallsAction0<T> implements Action1<T> {
+        final Action0 action;
+        
+        public Action1CallsAction0(Action0 action) {
+            this.action = action;
+        }
+        
+        @Override
+        public void call(T t) {
+            action.call();
+        }
+    }
 }

src/main/java/rx/functions/Actions.java

@@ -0,0 +1,46 @@
+/**
+ * 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.*;
+import rx.Observable.OnSubscribe;
+
+/**
+ * Holds a singleton instance of an empty Observable which is stateless and completes
+ * the child subscriber immediately.
+ */
+public enum EmptyObservableHolder implements OnSubscribe<Object> {
+    INSTANCE
+    ;
+    
+    /**
+     * Returns a type-corrected singleton instance of the empty Observable.
+     * @return a type-corrected singleton instance of the empty Observable.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> Observable<T> instance() {
+        return (Observable<T>)EMPTY;
+    }
+    
+    /** The singleton instance. */
+    static final Observable<Object> EMPTY = Observable.create(INSTANCE);
+    
+    @Override
+    public void call(Subscriber<? super Object> child) {
+        child.onCompleted();
+    }
+}

src/main/java/rx/internal/operators/EmptyObservableHolder.java

@@ -0,0 +1,45 @@
+/**
+ * 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.*;
+import rx.Observable.OnSubscribe;
+
+/**
+ * Holds a singleton instance of a never Observable which is stateless doesn't
+ * call any of the Subscriber's methods.
+ */
+public enum NeverObservableHolder implements OnSubscribe<Object> {
+    INSTANCE
+    ;
+    
+    /**
+     * Returns a type-corrected singleton instance of the never Observable.
+     * @return a type-corrected singleton instance of the never Observable.
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> Observable<T> instance() {
+        return (Observable<T>)NEVER;
+    }
+    
+    /** The singleton instance. */
+    static final Observable<Object> NEVER = Observable.create(INSTANCE);
+    
+    @Override
+    public void call(Subscriber<? super Object> child) {
+    }
+}

src/main/java/rx/internal/operators/NeverObservableHolder.java

@@ -0,0 +1,65 @@
+/**
+ * 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.*;
+
+/**
+ * Transforms the downstream Subscriber into a Subscriber via an operator
+ * callback and calls the parent OnSubscribe.call() method with it.
+ * @param <T> the source value type
+ * @param <R> the result value type
+ */
+public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {
+    
+    static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
+
+    final OnSubscribe<T> parent;
+
+    final Operator<? extends R, ? super T> operator;
+
+    public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {
+        this.parent = parent;
+        this.operator = operator;
+    }
+
+    @Override
+    public void call(Subscriber<? super R> o) {
+        try {
+            Subscriber<? super T> st = hook.onLift(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);
+        }
+    }
+}

src/main/java/rx/internal/operators/OnSubscribeLift.java

@@ -0,0 +1,46 @@
+/**
+ * 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.*;
+import rx.Observable.OnSubscribe;
+
+/**
+ * An Observable that invokes {@link Observer#onError onError} when the {@link Observer} subscribes to it.
+ * 
+ * @param <T>
+ *            the type of item (ostensibly) emitted by the Observable
+ */
+public final class OnSubscribeThrow<T> implements OnSubscribe<T> {
+
+    private final Throwable exception;
+
+    public OnSubscribeThrow(Throwable exception) {
+        this.exception = exception;
+    }
+
+    /**
+     * Accepts an {@link Observer} and calls its {@link Observer#onError onError} method.
+     * 
+     * @param observer
+     *            an {@link Observer} of this Observable
+     */
+    @Override
+    public void call(Subscriber<? super T> observer) {
+        observer.onError(exception);
+    }
+}

src/main/java/rx/internal/operators/OnSubscribeThrow.java

@@ -0,0 +1,48 @@
+/**
+ * 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.util;
+
+import rx.*;
+import rx.functions.*;
+
+/**
+ * An Observer that forwards the onXXX method calls to a notification callback
+ * by transforming each signal type into Notifications.
+ * @param <T> the value type
+ */
+public final class ActionNotificationObserver<T> implements Observer<T> {
+    
+    final Action1<Notification<? super T>> onNotification;
+
+    public ActionNotificationObserver(Action1<Notification<? super T>> onNotification) {
+        this.onNotification = onNotification;
+    }
+
+    @Override
+    public void onNext(T t) {
+        onNotification.call(Notification.createOnNext(t));
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        onNotification.call(Notification.createOnError(e));
+    }
+
+    @Override
+    public void onCompleted() {
+        onNotification.call(Notification.createOnCompleted());
+    }
+}

src/main/java/rx/internal/util/ActionNotificationObserver.java

@@ -0,0 +1,51 @@
+/**
+ * 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.util;
+
+import rx.Subscriber;
+import rx.functions.*;
+
+/**
+ * A Subscriber that forwards the onXXX method calls to callbacks.
+ * @param <T> the value type
+ */
+public final class ActionSubscriber<T> extends Subscriber<T> {
+
+    final Action1<? super T> onNext;
+    final Action1<Throwable> onError;
+    final Action0 onCompleted;
+
+    public ActionSubscriber(Action1<? super T> onNext, Action1<Throwable> onError, Action0 onCompleted) {
+        this.onNext = onNext;
+        this.onError = onError;
+        this.onCompleted = onCompleted;
+    }
+
+    @Override
+    public void onNext(T t) {
+        onNext.call(t);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        onError.call(e);
+    }
+
+    @Override
+    public void onCompleted() {
+        onCompleted.call();
+    }
+}