Skip to content

Commit 006b399

Browse files
akarnokdakarnokd
committed
Simplest tests of groupBy, merge and Notification
1 parent b5c7577 commit 006b399

4 files changed

Lines changed: 207 additions & 1 deletion

File tree

‎src/main/java/io/reactivex/Notification.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public static <T> Try<Optional<T>> error(Throwable e) {
3636
}
3737

3838
public static <T> Try<Optional<T>> next(T value) {
39-
Objects.requireNonNull(value);
39+
Objects.requireNonNull(value); // TODO this coud instead return an error of NPE
4040
return Try.ofValue(Optional.of(value));
4141
}
4242
}

‎src/test/java/io/reactivex/GroupByTests.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/**
2+
* Copyright 2015 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex;
15+
16+
import org.junit.Test;
17+
18+
public class GroupByTests {
19+
20+
@Test
21+
public void testTakeUnsubscribesOnGroupBy() {
22+
Observable.merge(
23+
EventStream.getEventStream("HTTP-ClusterA", 50),
24+
EventStream.getEventStream("HTTP-ClusterB", 20)
25+
)
26+
// group by type (2 clusters)
27+
.groupBy(event -> event.type)
28+
.take(1)
29+
.toBlocking()
30+
.forEach(System.out::println);
31+
32+
System.out.println("**** finished");
33+
}
34+
35+
@Test
36+
public void testTakeUnsubscribesOnFlatMapOfGroupBy() {
37+
Observable.merge(
38+
EventStream.getEventStream("HTTP-ClusterA", 50),
39+
EventStream.getEventStream("HTTP-ClusterB", 20)
40+
)
41+
// group by type (2 clusters)
42+
.groupBy(event -> event.type)
43+
.flatMap(g -> g.map(event -> event.instanceId + " - " + event.values.get("count200")))
44+
.take(20)
45+
.toBlocking()
46+
.forEach(System.out::println);
47+
48+
System.out.println("**** finished");
49+
}
50+
}

‎src/test/java/io/reactivex/MergeTests.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
package io.reactivex;
2+
3+
import static org.junit.Assert.*;
4+
5+
import java.util.List;
6+
7+
import org.junit.Test;
8+
9+
import io.reactivex.CovarianceTest.*;
10+
11+
public class MergeTests {
12+
13+
/**
14+
* This won't compile if super/extends isn't done correctly on generics
15+
*/
16+
@Test
17+
public void testCovarianceOfMerge() {
18+
Observable<HorrorMovie> horrors = Observable.just(new HorrorMovie());
19+
Observable<Observable<HorrorMovie>> metaHorrors = Observable.just(horrors);
20+
Observable.<Media> merge(metaHorrors);
21+
}
22+
23+
@Test
24+
public void testMergeCovariance() {
25+
Observable<Media> o1 = Observable.<Media> just(new HorrorMovie(), new Movie());
26+
Observable<Media> o2 = Observable.just(new Media(), new HorrorMovie());
27+
28+
Observable<Observable<Media>> os = Observable.just(o1, o2);
29+
30+
List<Media> values = Observable.merge(os).toList().toBlocking().single();
31+
32+
assertEquals(4, values.size());
33+
}
34+
35+
@Test
36+
public void testMergeCovariance2() {
37+
Observable<Media> o1 = Observable.just(new HorrorMovie(), new Movie(), new Media());
38+
Observable<Media> o2 = Observable.just(new Media(), new HorrorMovie());
39+
40+
Observable<Observable<Media>> os = Observable.just(o1, o2);
41+
42+
List<Media> values = Observable.merge(os).toList().toBlocking().single();
43+
44+
assertEquals(5, values.size());
45+
}
46+
47+
@Test
48+
public void testMergeCovariance3() {
49+
Observable<Movie> o1 = Observable.just(new HorrorMovie(), new Movie());
50+
Observable<Media> o2 = Observable.just(new Media(), new HorrorMovie());
51+
52+
List<Media> values = Observable.merge(o1, o2).toList().toBlocking().single();
53+
54+
assertTrue(values.get(0) instanceof HorrorMovie);
55+
assertTrue(values.get(1) instanceof Movie);
56+
assertTrue(values.get(2) != null);
57+
assertTrue(values.get(3) instanceof HorrorMovie);
58+
}
59+
60+
@Test
61+
public void testMergeCovariance4() {
62+
63+
Observable<Movie> o1 = Observable.defer(() -> Observable.just(
64+
new HorrorMovie(),
65+
new Movie()
66+
));
67+
68+
Observable<Media> o2 = Observable.just(new Media(), new HorrorMovie());
69+
70+
List<Media> values = Observable.merge(o1, o2).toList().toBlocking().single();
71+
72+
assertTrue(values.get(0) instanceof HorrorMovie);
73+
assertTrue(values.get(1) instanceof Movie);
74+
assertTrue(values.get(2) != null);
75+
assertTrue(values.get(3) instanceof HorrorMovie);
76+
}
77+
78+
}

‎src/test/java/io/reactivex/NotificationTest.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
/**
2+
* Copyright 2015 Netflix, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex;
15+
16+
import java.util.Optional;
17+
18+
import org.junit.*;
19+
20+
public class NotificationTest {
21+
22+
@Test(expected = NullPointerException.class)
23+
public void testOnNextIntegerNotificationDoesNotEqualNullNotification(){
24+
final Try<Optional<Integer>> integerNotification = Notification.next(1);
25+
final Try<Optional<Integer>> nullNotification = Notification.next(null);
26+
Assert.assertFalse(integerNotification.equals(nullNotification));
27+
}
28+
29+
@Test(expected = NullPointerException.class)
30+
public void testOnNextNullNotificationDoesNotEqualIntegerNotification(){
31+
final Try<Optional<Integer>> integerNotification = Notification.next(1);
32+
final Try<Optional<Integer>> nullNotification = Notification.next(null);
33+
Assert.assertFalse(nullNotification.equals(integerNotification));
34+
}
35+
36+
@Test
37+
public void testOnNextIntegerNotificationsWhenEqual(){
38+
final Try<Optional<Integer>> integerNotification = Notification.next(1);
39+
final Try<Optional<Integer>> integerNotification2 = Notification.next(1);
40+
Assert.assertTrue(integerNotification.equals(integerNotification2));
41+
}
42+
43+
@Test
44+
public void testOnNextIntegerNotificationsWhenNotEqual(){
45+
final Try<Optional<Integer>> integerNotification = Notification.next(1);
46+
final Try<Optional<Integer>> integerNotification2 = Notification.next(2);
47+
Assert.assertFalse(integerNotification.equals(integerNotification2));
48+
}
49+
50+
@Test
51+
public void testOnErrorIntegerNotificationDoesNotEqualNullNotification(){
52+
final Try<Optional<Integer>> integerNotification = Notification.error(new Exception());
53+
final Try<Optional<Integer>> nullNotification = Notification.error(null);
54+
Assert.assertFalse(integerNotification.equals(nullNotification));
55+
}
56+
57+
@Test
58+
public void testOnErrorNullNotificationDoesNotEqualIntegerNotification(){
59+
final Try<Optional<Integer>> integerNotification = Notification.error(new Exception());
60+
final Try<Optional<Integer>> nullNotification = Notification.error(null);
61+
Assert.assertFalse(nullNotification.equals(integerNotification));
62+
}
63+
64+
@Test
65+
public void testOnErrorIntegerNotificationsWhenEqual(){
66+
final Exception exception = new Exception();
67+
final Try<Optional<Integer>> onErrorNotification = Notification.error(exception);
68+
final Try<Optional<Integer>> onErrorNotification2 = Notification.error(exception);
69+
Assert.assertTrue(onErrorNotification.equals(onErrorNotification2));
70+
}
71+
72+
@Test
73+
public void testOnErrorIntegerNotificationWhenNotEqual(){
74+
final Try<Optional<Integer>> onErrorNotification = Notification.error(new Exception());
75+
final Try<Optional<Integer>> onErrorNotification2 = Notification.error(new Exception());
76+
Assert.assertFalse(onErrorNotification.equals(onErrorNotification2));
77+
}
78+
}

0 commit comments

Comments
 (0)