Skip to content

Fixes on #284 (groupBy) #289

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

Closed
wants to merge 6 commits into from
Closed

Fixes on #284 (groupBy) #289

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4f68c5a
Add unit test for sub/unsub/sub bug.
Treora Jun 5, 2013
22cd318
Fix sub/unsub/sub bug.
Treora Jun 5, 2013
787ae6a
Add unit test for subscribing twice to GroupBy
Treora Jun 5, 2013
fc89a79
Fix bug to allow subscribing several times
Treora Jun 5, 2013
d097b55
Merge branch 'issue282-groupBy-fixes'
Treora Jun 5, 2013
42af69d
Add a bit of documentation
Treora Jun 30, 2013
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
288 changes: 224 additions & 64 deletions rxjava-core/src/main/java/rx/operators/OperationGroupBy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,12 @@
package rx.operators;

import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
Expand All @@ -29,6 +32,7 @@
import java.util.concurrent.atomic.AtomicReference;

import org.junit.Test;
import org.mockito.InOrder;

import rx.Observable;
import rx.Observer;
Expand Down Expand Up @@ -64,109 +68,129 @@ public static <K, T> Func1<Observer<GroupedObservable<K, T>>, Subscription> grou
private static class GroupBy<K, V> implements Func1<Observer<GroupedObservable<K, V>>, Subscription> {

private final Observable<KeyValue<K, V>> source;
private final ConcurrentHashMap<K, GroupedSubject<K, V>> groupedObservables = new ConcurrentHashMap<K, GroupedSubject<K, V>>();
private final AtomicObservableSubscription actualParentSubscription = new AtomicObservableSubscription();
private final AtomicInteger numGroupSubscriptions = new AtomicInteger();
private final AtomicBoolean unsubscribeRequested = new AtomicBoolean(false);

private GroupBy(Observable<KeyValue<K, V>> source) {
this.source = source;
}

@Override
public Subscription call(final Observer<GroupedObservable<K, V>> observer) {
final GroupBy<K, V> _this = this;
actualParentSubscription.wrap(source.subscribe(new Observer<KeyValue<K, V>>() {

@Override
public void onCompleted() {
// we need to propagate to all children I imagine ... we can't just leave all of those Observable/Observers hanging
for (GroupedSubject<K, V> o : groupedObservables.values()) {
o.onCompleted();
}
// now the parent
observer.onCompleted();
}

@Override
public void onError(Exception e) {
// we need to propagate to all children I imagine ... we can't just leave all of those Observable/Observers hanging
for (GroupedSubject<K, V> o : groupedObservables.values()) {
o.onError(e);
}
// now the parent
observer.onError(e);
}

@Override
public void onNext(KeyValue<K, V> value) {
GroupedSubject<K, V> gs = groupedObservables.get(value.key);
if (gs == null) {
if (unsubscribeRequested.get()) {
// unsubscribe has been requested so don't create new groups
// only send data to groups already created
return;
}
/*
* Technically the source should be single-threaded so we shouldn't need to do this but I am
* programming defensively as most operators are so this can work with a concurrent sequence
* if it ends up receiving one.
*/
GroupedSubject<K, V> newGs = GroupedSubject.<K, V> create(value.key, _this);
GroupedSubject<K, V> existing = groupedObservables.putIfAbsent(value.key, newGs);
if (existing == null) {
// we won so use the one we created
gs = newGs;
// since we won the creation we emit this new GroupedObservable
observer.onNext(gs);
} else {
// another thread beat us so use the existing one
gs = existing;
}
}
gs.onNext(value.value);
}
}));
final AtomicObservableSubscription actualParentSubscription = new AtomicObservableSubscription();
final GroupByObserver<K,V> internalObserver = new GroupByObserver<K, V>(observer, actualParentSubscription);
actualParentSubscription.wrap(source.subscribe(internalObserver));

return new Subscription() {

@Override
public void unsubscribe() {
if (numGroupSubscriptions.get() == 0) {
if (internalObserver.numGroupSubscriptions.get() == 0) {
// if we have no group subscriptions we will unsubscribe
actualParentSubscription.unsubscribe();
} else {
// otherwise we mark to not send any more groups (waiting on existing groups to finish)
unsubscribeRequested.set(true);
internalObserver.unsubscribeRequested.set(true);
}
}
};
}
}

Copy link

Choose a reason for hiding this comment

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

even one-line javadoc would be nice

/**
* The internal observer that is spawned on a subscription to GroupBy.
* Whenever it receives a new KeyValue from the source, it passes the value to the correct GroupedObservable(Subject),
* which is determined by the key. If no GroupedObservable existed for this key, a new one is created first and passed
* to the primary observer.
* When an emitted GroupedObservable is subscribed to, it notifies this internal observer, which keeps count of the
* number of these subscriptions. When both the primary observer has unsubscribed (thus no groups are emitted anymore),
* and no groups have any observers left, the whole thing is considered done and the upstream subscription is terminated.
* Note that to function reliably, this implementation therefore requires that emitted GroupedObservables are subscribed
* to immediately, and not stored for later use (after the GroupBy has been unsubscribed from).
*/
private static class GroupByObserver<K, V> implements Observer<KeyValue<K, V>> {
private final ConcurrentHashMap<K, GroupedSubject<K, V>> groupedObservables = new ConcurrentHashMap<K, GroupedSubject<K, V>>();
private final AtomicInteger numGroupSubscriptions = new AtomicInteger();
private final AtomicBoolean unsubscribeRequested = new AtomicBoolean(false);
private final Observer<GroupedObservable<K, V>> observer;
private final Subscription actualParentSubscription;

private GroupByObserver(Observer<GroupedObservable<K, V>> observer, Subscription actualParentSubscription) {
this.observer = observer;
this.actualParentSubscription = actualParentSubscription;
}

@Override
public void onCompleted() {
// we need to propagate to all children I imagine ... we can't just leave all of those Observable/Observers hanging
for (GroupedSubject<K, V> o : groupedObservables.values()) {
o.onCompleted();
}
// now the parent
observer.onCompleted();
}

@Override
public void onError(Exception e) {
// we need to propagate to all children I imagine ... we can't just leave all of those Observable/Observers hanging
for (GroupedSubject<K, V> o : groupedObservables.values()) {
o.onError(e);
}
// now the parent
observer.onError(e);
}

@Override
public void onNext(KeyValue<K, V> value) {
GroupedSubject<K, V> gs = groupedObservables.get(value.key);
if (gs == null) {
if (unsubscribeRequested.get()) {
// unsubscribe has been requested so don't create new groups
// only send data to groups already created
return;
}
/*
* Technically the source should be single-threaded so we shouldn't need to do this but I am
* programming defensively as most operators are so this can work with a concurrent sequence
* if it ends up receiving one.
*/
GroupedSubject<K, V> newGs = GroupedSubject.<K, V> create(value.key, this);
GroupedSubject<K, V> existing = groupedObservables.putIfAbsent(value.key, newGs);
if (existing == null) {
// we won so use the one we created
gs = newGs;
// since we won the creation we emit this new GroupedObservable
observer.onNext(gs);
} else {
// another thread beat us so use the existing one
gs = existing;
}
}
gs.onNext(value.value);
}

/**
* Children notify of being subscribed to.
*
* @param key
*/
private void subscribeKey(K key) {
numGroupSubscriptions.incrementAndGet();
}

/**
* Children notify of being unsubscribed from.
*
* @param key
*/
private void unsubscribeKey(K key) {
int c = numGroupSubscriptions.decrementAndGet();
if (c == 0) {
if (c == 0 && unsubscribeRequested.get()) {
actualParentSubscription.unsubscribe();
}
}
}

/**
* The GroupedObservables that are passed to the observer.
*/
private static class GroupedSubject<K, T> extends GroupedObservable<K, T> implements Observer<T> {

static <K, T> GroupedSubject<K, T> create(final K key, final GroupBy<K, T> parent) {
static <K, T> GroupedSubject<K, T> create(final K key, final GroupByObserver<K, T> parent) {
@SuppressWarnings("unchecked")
final AtomicReference<Observer<T>> subscribedObserver = new AtomicReference<Observer<T>>(EMPTY_OBSERVER);

Expand Down Expand Up @@ -556,6 +580,142 @@ public String toString() {
}
}

/*
* Test subscribing to a group, unsubscribing from it again, and subscribing to a next group
*/
@Test
public void testSubscribeAndImmediatelyUnsubscribeFirstGroup() {
CounterSource source = new CounterSource();
@SuppressWarnings("unchecked")
final Observer<Integer> observer = mock(Observer.class);

Func1<Integer, Integer> modulo2 = new Func1<Integer, Integer>() {
@Override
public Integer call(Integer x) {
return x%2;
}
};

Subscription outerSubscription = source.groupBy(modulo2).subscribe(new Action1<GroupedObservable<Integer, Integer>>() {
@Override
public void call(GroupedObservable<Integer, Integer> group) {
Subscription innerSubscription = group.subscribe(observer);
if (group.getKey() == 0) {
// We immediately unsubscribe again from the even numbers
innerSubscription.unsubscribe();
// We should still get the group of odd numbers
}
}
});
try {
for (Thread t : source.threads) {
t.join();
}
} catch (InterruptedException ex) {
}

InOrder o = inOrder(observer);
// With a different implementation that subscribes to the group concurrently, we might actually receive 0.
o.verify(observer, never()).onNext(0);
o.verify(observer).onNext(1);
o.verify(observer, never()).onNext(2);
o.verify(observer).onNext(3);
o.verify(observer, never()).onNext(4);
o.verify(observer).onNext(5);
o.verify(observer, never()).onNext(6);
o.verify(observer).onNext(7);
o.verify(observer, never()).onNext(8);
o.verify(observer).onNext(9);
}

@Test
public void testSubscribeTwoTimesToGroupBy() {
CounterSource source = new CounterSource();
@SuppressWarnings("unchecked")
final Observer<Integer> observer1 = mock(Observer.class);
@SuppressWarnings("unchecked")
final Observer<Integer> observer2 = mock(Observer.class);

Func1<Integer, Integer> modulo2 = new Func1<Integer, Integer>() {
@Override
public Integer call(Integer x) {
return x % 2;
}
};

Observable<GroupedObservable<Integer, Integer>> groups = source.groupBy(modulo2);

Subscription outerSubscription1 = groups.subscribe(new Action1<GroupedObservable<Integer, Integer>>() {
@Override
public void call(GroupedObservable<Integer, Integer> group) {
group.subscribe(observer1);
}
});
Subscription outerSubscription2 = groups.subscribe(new Action1<GroupedObservable<Integer, Integer>>() {
@Override
public void call(GroupedObservable<Integer, Integer> group) {
group.subscribe(observer2);
}
});
try {
for (Thread t : source.threads) {
t.join();
}
} catch (InterruptedException ex) {
}

// Receival of values by observer1 and observer2 can be interleaved, so use inOrder on them separately
InOrder o1 = inOrder(observer1);
o1.verify(observer1).onNext(0);
o1.verify(observer1).onNext(1);
o1.verify(observer1).onNext(2);
o1.verify(observer1).onNext(3);
o1.verify(observer1).onNext(4);
o1.verify(observer1).onNext(5);
o1.verify(observer1).onNext(6);
o1.verify(observer1).onNext(7);
o1.verify(observer1).onNext(8);
o1.verify(observer1).onNext(9);
InOrder o2 = inOrder(observer2);
o2.verify(observer2).onNext(0);
o2.verify(observer2).onNext(1);
o2.verify(observer2).onNext(2);
o2.verify(observer2).onNext(3);
o2.verify(observer2).onNext(4);
o2.verify(observer2).onNext(5);
o2.verify(observer2).onNext(6);
o2.verify(observer2).onNext(7);
o2.verify(observer2).onNext(8);
o2.verify(observer2).onNext(9);
}

private class CounterSource extends Observable<Integer> {
public List<Thread> threads = new ArrayList<Thread>();
@Override
public Subscription subscribe(final Observer<Integer> observer) {
final Thread thread = new Thread(new Runnable() {
@Override
public void run() {
int i = 0;
while (i < 10) {
observer.onNext(i++);
if (Thread.interrupted()) {
return;
}
}
}
});
thread.start();
threads.add(thread);
return new Subscription() {
@Override
public void unsubscribe() {
thread.interrupt();
}
};
}
}

}

}