Nono null checks

Author: akarnokd

src/main/java/hu/akarnokd/rxjava2/basetypes/Nono.java

@@ -253,6 +253,9 @@ public static <R> Nono using(Callable<R> resourceSupplier, Function<? super R, ?
 
     public static <R> Nono using(Callable<R> resourceSupplier, Function<? super R, ? extends Nono> sourceSupplier,
             Consumer<? super R> disposer, boolean eager) {
+        ObjectHelper.requireNonNull(resourceSupplier, "resourceSupplier is null");
+        ObjectHelper.requireNonNull(sourceSupplier, "sourceSupplier is null");
+        ObjectHelper.requireNonNull(disposer, "disposer is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
@@ -290,11 +293,14 @@ public static Nono fromObservable(ObservableSource<?> source) {
     // -----------------------------------------------------------
 
     public final <T> Flowable<T> andThen(Publisher<? extends T> other) {
+        ObjectHelper.requireNonNull(other, "other is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono andThen(Nono other) {
+        ObjectHelper.requireNonNull(other, "other is null");
+        // TODO implement
         throw new UnsupportedOperationException();
     }
 
@@ -305,10 +311,13 @@ public final Nono delay(long delay, TimeUnit unit) {
 
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public final Nono delay(long delay, TimeUnit unit, Scheduler scheduler) {
+        ObjectHelper.requireNonNull(unit, "unit is null");
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         return onAssembly(new NonoDelay(this, delay, unit, scheduler));
     }
 
     public final Nono delaySubscription(Publisher<?> other) {
+        ObjectHelper.requireNonNull(other, "other is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
@@ -330,11 +339,16 @@ public final Nono timeout(long delay, TimeUnit unit, Nono fallback) {
     }
 
     public final Nono timeout(long delay, TimeUnit unit, Scheduler scheduler) {
+        ObjectHelper.requireNonNull(unit, "unit is null");
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono timeout(long delay, TimeUnit unit, Scheduler scheduler, Nono fallback) {
+        ObjectHelper.requireNonNull(unit, "unit is null");
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
+        ObjectHelper.requireNonNull(fallback, "fallback is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
@@ -344,17 +358,20 @@ public final Nono onErrorComplete() {
     }
 
     public final Nono onErrorResumeNext(Function<? super Throwable, ? extends Nono> errorHandler) {
+        ObjectHelper.requireNonNull(errorHandler, "errorHandler is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono mapError(Function<? super Throwable, ? extends Throwable> mapper) {
-        // TODO implement
-        throw new UnsupportedOperationException();
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        return onAssembly(new NonoMapError(this, mapper));
     }
 
     public final <T> Flowable<T> flatMap(Function<? super Throwable, ? extends Publisher<? extends T>> onErrorMapper,
             Callable<? extends Publisher<? extends T>> onCompleteMapper) {
+        ObjectHelper.requireNonNull(onErrorMapper, "onErrorMapper is null");
+        ObjectHelper.requireNonNull(onCompleteMapper, "onCompleteMapper is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
@@ -401,47 +418,101 @@ public final <T> Maybe<T> toMaybe() {
     }
 
     public final Nono subscribeOn(Scheduler scheduler) {
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         return onAssembly(new NonoSubscribeOn(this, scheduler));
     }
 
     public final Nono observeOn(Scheduler scheduler) {
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         return onAssembly(new NonoObserveOn(this, scheduler));
     }
 
     public final Nono unsubscribeOn(Scheduler scheduler) {
+        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doOnComplete(Action action) {
+        ObjectHelper.requireNonNull(action, "action is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doOnError(Consumer<? super Throwable> error) {
+        ObjectHelper.requireNonNull(error, "error is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doAfterComplete(Action action) {
+        ObjectHelper.requireNonNull(action, "action is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doAfterTerminate(Action action) {
+        ObjectHelper.requireNonNull(action, "action is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doFinally(Action action) {
+        ObjectHelper.requireNonNull(action, "action is null");
         // TODO implement
         throw new UnsupportedOperationException();
     }
 
     public final Nono doOnCancel(Action action) {
+        ObjectHelper.requireNonNull(action, "action is null");
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono repeat() {
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono repeat(long times) {
         // TODO implement
         throw new UnsupportedOperationException();
     }
+
+    public final Nono repeat(BooleanSupplier stop) {
+        ObjectHelper.requireNonNull(stop, "stop is null");
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono repeatWhen(Function<Flowable<Object>, Publisher<?>> handler) {
+        ObjectHelper.requireNonNull(handler, "handler is null");
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono retry() {
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono retry(long times) {
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono retry(Predicate<? super Throwable> predicate) {
+        ObjectHelper.requireNonNull(predicate, "predicate is null");
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
+    public final Nono retryWhen(Function<Flowable<Throwable>, Publisher<?>> handler) {
+        ObjectHelper.requireNonNull(handler, "handler is null");
+        // TODO implement
+        throw new UnsupportedOperationException();
+    }
+
     // -----------------------------------------------------------
     // Consumers and subscribers (leave)
     // -----------------------------------------------------------

src/main/java/hu/akarnokd/rxjava2/basetypes/NonoMapError.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2016 David Karnok
+ *
+ * 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 hu.akarnokd.rxjava2.basetypes;
+
+import org.reactivestreams.Subscriber;
+
+import io.reactivex.exceptions.*;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.functions.ObjectHelper;
+
+/**
+ * Map an upstream error into another Throwable.
+ */
+final class NonoMapError extends Nono {
+
+    final Nono source;
+
+    final Function<? super Throwable, ? extends Throwable> mapper;
+
+    NonoMapError(Nono source, Function<? super Throwable, ? extends Throwable> mapper) {
+        this.source = source;
+        this.mapper = mapper;
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super Void> s) {
+        source.subscribe(new MapErrorSubscriber(s, mapper));
+    }
+
+    static final class MapErrorSubscriber extends BasicNonoSubscriber {
+
+        final Function<? super Throwable, ? extends Throwable> mapper;
+
+        public MapErrorSubscriber(Subscriber<? super Void> actual, Function<? super Throwable, ? extends Throwable> mapper) {
+            super(actual);
+            this.mapper = mapper;
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            Throwable ex;
+            try {
+                ex = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null Throwable");
+            } catch (Throwable exc) {
+                Exceptions.throwIfFatal(exc);
+                ex = new CompositeException(t, exc);
+            }
+            actual.onError(ex);
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+        }
+
+    }
+}