Skip to content

2.x: coverage, fixes, cleanup 8/27-1 #4431

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 26, 2016
Merged

2.x: coverage, fixes, cleanup 8/27-1 #4431

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
187 changes: 128 additions & 59 deletions src/main/java/io/reactivex/internal/operators/single/SingleCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,97 +13,166 @@

package io.reactivex.internal.operators.single;

import java.util.*;
import java.util.concurrent.atomic.*;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.internal.disposables.EmptyDisposable;
import io.reactivex.internal.util.NotificationLite;

public final class SingleCache<T> extends Single<T> {
public final class SingleCache<T> extends Single<T> implements SingleObserver<T> {

@SuppressWarnings("rawtypes")
static final CacheDisposable[] EMPTY = new CacheDisposable[0];
@SuppressWarnings("rawtypes")
static final CacheDisposable[] TERMINATED = new CacheDisposable[0];

final SingleSource<? extends T> source;

final AtomicInteger wip;
final AtomicReference<Object> notification;
final List<SingleObserver<? super T>> subscribers;

final AtomicReference<CacheDisposable<T>[]> observers;

T value;

Throwable error;

@SuppressWarnings("unchecked")
public SingleCache(SingleSource<? extends T> source) {
this.source = source;
this.wip = new AtomicInteger();
this.notification = new AtomicReference<Object>();
this.subscribers = new ArrayList<SingleObserver<? super T>>();
this.observers = new AtomicReference<CacheDisposable<T>[]>(EMPTY);
}

@Override
protected void subscribeActual(SingleObserver<? super T> s) {

Object o = notification.get();
if (o != null) {
s.onSubscribe(EmptyDisposable.INSTANCE);
if (NotificationLite.isError(o)) {
s.onError(NotificationLite.getError(o));
protected void subscribeActual(final SingleObserver<? super T> s) {
CacheDisposable<T> d = new CacheDisposable<T>(s, this);
s.onSubscribe(d);

if (add(d)) {
if (d.isDisposed()) {
remove(d);
}
} else {
Throwable ex = error;
if (ex != null) {
s.onError(ex);
} else {
s.onSuccess(NotificationLite.<T>getValue(o));
s.onSuccess(value);
}
return;
}

synchronized (subscribers) {
o = notification.get();
if (o == null) {
subscribers.add(s);
if (wip.getAndIncrement() == 0) {
source.subscribe(this);
}
}

boolean add(CacheDisposable<T> observer) {
for (;;) {
CacheDisposable<T>[] a = observers.get();
if (a == TERMINATED) {
return false;
}
int n = a.length;
@SuppressWarnings("unchecked")
CacheDisposable<T>[] b = new CacheDisposable[n + 1];
System.arraycopy(a, 0, b, 0, n);
b[n] = observer;
if (observers.compareAndSet(a, b)) {
return true;
}
}
if (o != null) {
s.onSubscribe(EmptyDisposable.INSTANCE);
if (NotificationLite.isError(o)) {
s.onError(NotificationLite.getError(o));
}

@SuppressWarnings("unchecked")
void remove(CacheDisposable<T> observer) {
for (;;) {
CacheDisposable<T>[] a = observers.get();
int n = a.length;
if (n == 0) {
return;
}

int j = -1;
for (int i = 0; i < n; i++) {
if (a[i] == observer) {
j = i;
break;
}
}

if (j < 0) {
return;
}

CacheDisposable<T>[] b;

if (n == 1) {
b = EMPTY;
} else {
s.onSuccess(NotificationLite.<T>getValue(o));
b = new CacheDisposable[n - 1];
System.arraycopy(a, 0, b, 0, j);
System.arraycopy(a, j + 1, b, j, n - j - 1);
}
if (observers.compareAndSet(a, b)) {
return;
}
return;
}
}

@Override
public void onSubscribe(Disposable d) {
// not supported by this operator
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

throw assertion error to ensure it never happens?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don't support dispose() once the cache has subscribed but the upstream should still run - this method is a no-op. Throwing here is not allowed anyway.

However, I'm willing to extend the SingleCache (and the others) by implementing Disposable and people can then cast and cancel a cache() too.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah. I don't think that's worth it. I just misunderstood the comment and thought "not allowed" instead of "not supported".

}

@SuppressWarnings("unchecked")
@Override
public void onSuccess(T value) {
this.value = value;

if (wip.getAndIncrement() != 0) {
return;
for (CacheDisposable<T> d : observers.getAndSet(TERMINATED)) {
if (!d.isDisposed()) {
d.actual.onSuccess(value);
}
}
}

@SuppressWarnings("unchecked")
@Override
public void onError(Throwable e) {
this.error = e;

source.subscribe(new SingleObserver<T>() {

@Override
public void onSubscribe(Disposable d) {

for (CacheDisposable<T> d : observers.getAndSet(TERMINATED)) {
if (!d.isDisposed()) {
d.actual.onError(e);
}
}
}

static final class CacheDisposable<T>
extends AtomicBoolean
implements Disposable {
/** */
private static final long serialVersionUID = 7514387411091976596L;

@Override
public void onSuccess(T value) {
notification.set(NotificationLite.next(value));
List<SingleObserver<? super T>> list;
synchronized (subscribers) {
list = new ArrayList<SingleObserver<? super T>>(subscribers);
subscribers.clear();
}
for (SingleObserver<? super T> s1 : list) {
s1.onSuccess(value);
}
}
final SingleObserver<? super T> actual;

final SingleCache<T> parent;

@Override
public void onError(Throwable e) {
notification.set(NotificationLite.error(e));
List<SingleObserver<? super T>> list;
synchronized (subscribers) {
list = new ArrayList<SingleObserver<? super T>>(subscribers);
subscribers.clear();
}
for (SingleObserver<? super T> s1 : list) {
s1.onError(e);
}
public CacheDisposable(SingleObserver<? super T> actual, SingleCache<T> parent) {
this.actual = actual;
this.parent = parent;
}

@Override
public boolean isDisposed() {
return get();
}

@Override
public void dispose() {
if (compareAndSet(false, true)) {
parent.remove(this);
}

});
}
}

}
56 changes: 0 additions & 56 deletions src/main/java/io/reactivex/internal/subscribers/observable/EmptyObserver.java
View file
Open in desktop

This file was deleted.

Loading