Skip to content

Commit

Permalink
2.x: Clean up null usages by using ObjectHelper.requireNonNull (#4699)
Browse files Browse the repository at this point in the history
  • Loading branch information
@vanniktech @akarnokd
vanniktech authored and akarnokd committed Oct 12, 2016
1 parent e71d371 commit 4a614dc
Show file tree
Hide file tree
Showing 51 changed files with 146 additions and 611 deletions.
7 changes: 1 addition & 6 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableConcatIterable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -109,18 +109,13 @@ void next() {
CompletableSource c;

try {
c = a.next();
c = ObjectHelper.requireNonNull(a.next(), "The CompletableSource returned is null");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
actual.onError(ex);
return;
}

if (c == null) {
actual.onError(new NullPointerException("The completable returned is null"));
return;
}

c.subscribe(this);
} while (decrementAndGet() != 0);
}
Expand Down
6 changes: 2 additions & 4 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableErrorSupplier.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.completable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.concurrent.Callable;

import io.reactivex.*;
Expand All @@ -32,15 +33,12 @@ protected void subscribeActual(CompletableObserver s) {
Throwable error;

try {
error = errorSupplier.call();
error = ObjectHelper.requireNonNull(errorSupplier.call(), "The error returned is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
error = e;
}

if (error == null) {
error = new NullPointerException("The error supplied is null");
}
EmptyDisposable.error(error, s);
}

Expand Down
7 changes: 1 addition & 6 deletions .../java/io/reactivex/internal/operators/completable/CompletableMergeDelayErrorIterable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,18 +40,13 @@ public void subscribeActual(final CompletableObserver s) {
Iterator<? extends CompletableSource> iterator;

try {
iterator = sources.iterator();
iterator = ObjectHelper.requireNonNull(sources.iterator(), "The source iterator returned is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
s.onError(e);
return;
}

if (iterator == null) {
s.onError(new NullPointerException("The source iterator returned is null"));
return;
}

final AtomicInteger wip = new AtomicInteger(1);

final AtomicThrowable error = new AtomicThrowable();
Expand Down
8 changes: 2 additions & 6 deletions src/main/java/io/reactivex/internal/operators/completable/CompletableMergeIterable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.completable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.Iterator;
import java.util.concurrent.atomic.*;

Expand All @@ -38,18 +39,13 @@ public void subscribeActual(final CompletableObserver s) {
Iterator<? extends CompletableSource> iterator;

try {
iterator = sources.iterator();
iterator = ObjectHelper.requireNonNull(sources.iterator(), "The source iterator returned is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
s.onError(e);
return;
}

if (iterator == null) {
s.onError(new NullPointerException("The source iterator returned is null"));
return;
}

final AtomicInteger wip = new AtomicInteger(1);

MergeCompletableObserver shared = new MergeCompletableObserver(s, set, wip);
Expand Down
27 changes: 4 additions & 23 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBuffer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.flowable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.*;
Expand Down Expand Up @@ -104,20 +105,14 @@ public void onNext(T t) {
if (b == null) {

try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The bufferSupplier returned a null buffer");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
onError(e);
return;
}

if (b == null) {
cancel();

onError(new NullPointerException("The bufferSupplier returned a null buffer"));
return;
}
buffer = b;
}

Expand Down Expand Up @@ -231,7 +226,7 @@ public void onNext(T t) {

if (i % skip == 0L) { // FIXME no need for modulo
try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The bufferSupplier returned a null buffer");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
Expand All @@ -240,13 +235,6 @@ public void onNext(T t) {
return;
}

if (b == null) {
cancel();

onError(new NullPointerException("The bufferSupplier returned a null buffer"));
return;
}

buffer = b;
}

Expand Down Expand Up @@ -390,21 +378,14 @@ public void onNext(T t) {
C b;

try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The bufferSupplier returned a null buffer");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
onError(e);
return;
}

if (b == null) {
cancel();

onError(new NullPointerException("The bufferSupplier returned a null buffer"));
return;
}

bs.offer(b);
}

Expand Down
15 changes: 3 additions & 12 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundary.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.flowable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -173,33 +174,23 @@ void open(Open window) {
U b;

try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The buffer supplied is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
onError(e);
return;
}

if (b == null) {
onError(new NullPointerException("The buffer supplied is null"));
return;
}

Publisher<? extends Close> p;

try {
p = bufferClose.apply(window);
p = ObjectHelper.requireNonNull(bufferClose.apply(window), "The buffer closing publisher is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
onError(e);
return;
}

if (p == null) {
onError(new NullPointerException("The buffer closing publisher is null"));
return;
}

if (cancelled) {
return;
}
Expand Down
35 changes: 5 additions & 30 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundarySupplier.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.flowable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicReference;
Expand Down Expand Up @@ -76,7 +77,7 @@ public void onSubscribe(Subscription s) {
U b;

try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The buffer supplied is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancelled = true;
Expand All @@ -85,18 +86,12 @@ public void onSubscribe(Subscription s) {
return;
}

if (b == null) {
cancelled = true;
s.cancel();
EmptySubscription.error(new NullPointerException("The buffer supplied is null"), actual);
return;
}
buffer = b;

Publisher<B> boundary;

try {
boundary = boundarySupplier.call();
boundary = ObjectHelper.requireNonNull(boundarySupplier.call(), "The boundary publisher supplied is null");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
cancelled = true;
Expand All @@ -105,13 +100,6 @@ public void onSubscribe(Subscription s) {
return;
}

if (boundary == null) {
cancelled = true;
s.cancel();
EmptySubscription.error(new NullPointerException("The boundary publisher supplied is null"), actual);
return;
}

BufferBoundarySubscriber<T, U, B> bs = new BufferBoundarySubscriber<T, U, B>(this);
other.set(bs);

Expand Down Expand Up @@ -185,24 +173,18 @@ void next() {
U next;

try {
next = bufferSupplier.call();
next = ObjectHelper.requireNonNull(bufferSupplier.call(), "The buffer supplied is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
actual.onError(e);
return;
}

if (next == null) {
cancel();
actual.onError(new NullPointerException("The buffer supplied is null"));
return;
}

Publisher<B> boundary;

try {
boundary = boundarySupplier.call();
boundary = ObjectHelper.requireNonNull(boundarySupplier.call(), "The boundary publisher supplied is null");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
cancelled = true;
Expand All @@ -211,13 +193,6 @@ void next() {
return;
}

if (boundary == null) {
cancelled = true;
s.cancel();
actual.onError(new NullPointerException("The boundary publisher supplied is null"));
return;
}

BufferBoundarySubscriber<T, U, B> bs = new BufferBoundarySubscriber<T, U, B>(this);

Disposable o = other.get();
Expand Down
17 changes: 3 additions & 14 deletions src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferExactBoundary.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package io.reactivex.internal.operators.flowable;

import io.reactivex.internal.functions.ObjectHelper;
import java.util.Collection;
import java.util.concurrent.Callable;

Expand Down Expand Up @@ -71,7 +72,7 @@ public void onSubscribe(Subscription s) {
U b;

try {
b = bufferSupplier.call();
b = ObjectHelper.requireNonNull(bufferSupplier.call(), "The buffer supplied is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancelled = true;
Expand All @@ -80,12 +81,6 @@ public void onSubscribe(Subscription s) {
return;
}

if (b == null) {
cancelled = true;
s.cancel();
EmptySubscription.error(new NullPointerException("The buffer supplied is null"), actual);
return;
}
buffer = b;

BufferBoundarySubscriber<T, U, B> bs = new BufferBoundarySubscriber<T, U, B>(this);
Expand Down Expand Up @@ -157,20 +152,14 @@ void next() {
U next;

try {
next = bufferSupplier.call();
next = ObjectHelper.requireNonNull(bufferSupplier.call(), "The buffer supplied is null");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
actual.onError(e);
return;
}

if (next == null) {
cancel();
actual.onError(new NullPointerException("The buffer supplied is null"));
return;
}

U b;
synchronized (this) {
b = buffer;
Expand Down
Loading

0 comments on commit 4a614dc

Please sign in to comment.