Skip to content

Commit 8b0f3b0

Browse files
zsxwingzsxwing
committed
Implemented the rest 'merge' overloads
1 parent 5a10454 commit 8b0f3b0

File tree

3 files changed

+238
-163
lines changed

3 files changed

+238
-163
lines changed

‎rxjava-core/src/main/java/rx/Observable.java

Lines changed: 147 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1257,10 +1257,8 @@ public static <T> Observable<T> merge(Observable<? extends Observable<? extends
12571257
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
12581258
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
12591259
*/
1260-
@SuppressWarnings("unchecked")
1261-
// suppress because the types are checked by the method signature before using a vararg
12621260
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2) {
1263-
return create(OperationMerge.merge(t1, t2));
1261+
return merge(from(t1, t2));
12641262
}
12651263

12661264
/**
@@ -1280,10 +1278,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
12801278
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
12811279
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
12821280
*/
1283-
@SuppressWarnings("unchecked")
1284-
// suppress because the types are checked by the method signature before using a vararg
12851281
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3) {
1286-
return create(OperationMerge.merge(t1, t2, t3));
1282+
return merge(from(t1, t2, t3));
12871283
}
12881284

12891285
/**
@@ -1304,10 +1300,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
13041300
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
13051301
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
13061302
*/
1307-
@SuppressWarnings("unchecked")
1308-
// suppress because the types are checked by the method signature before using a vararg
13091303
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4) {
1310-
return create(OperationMerge.merge(t1, t2, t3, t4));
1304+
return merge(from(t1, t2, t3, t4));
13111305
}
13121306

13131307
/**
@@ -1329,10 +1323,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
13291323
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
13301324
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
13311325
*/
1332-
@SuppressWarnings("unchecked")
1333-
// suppress because the types are checked by the method signature before using a vararg
13341326
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5) {
1335-
return create(OperationMerge.merge(t1, t2, t3, t4, t5));
1327+
return merge(from(t1, t2, t3, t4, t5));
13361328
}
13371329

13381330
/**
@@ -1355,10 +1347,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
13551347
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
13561348
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
13571349
*/
1358-
@SuppressWarnings("unchecked")
1359-
// suppress because the types are checked by the method signature before using a vararg
13601350
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6) {
1361-
return create(OperationMerge.merge(t1, t2, t3, t4, t5, t6));
1351+
return merge(from(t1, t2, t3, t4, t5, t6));
13621352
}
13631353

13641354
/**
@@ -1382,10 +1372,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
13821372
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
13831373
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
13841374
*/
1385-
@SuppressWarnings("unchecked")
1386-
// suppress because the types are checked by the method signature before using a vararg
13871375
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7) {
1388-
return create(OperationMerge.merge(t1, t2, t3, t4, t5, t6, t7));
1376+
return merge(from(t1, t2, t3, t4, t5, t6, t7));
13891377
}
13901378

13911379
/**
@@ -1410,10 +1398,8 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
14101398
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
14111399
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
14121400
*/
1413-
@SuppressWarnings("unchecked")
1414-
// suppress because the types are checked by the method signature before using a vararg
14151401
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8) {
1416-
return create(OperationMerge.merge(t1, t2, t3, t4, t5, t6, t7, t8));
1402+
return merge(from(t1, t2, t3, t4, t5, t6, t7, t8));
14171403
}
14181404

14191405
/**
@@ -1439,10 +1425,148 @@ public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? e
14391425
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
14401426
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099.aspx">MSDN: Observable.Merge</a>
14411427
*/
1442-
@SuppressWarnings("unchecked")
14431428
// suppress because the types are checked by the method signature before using a vararg
14441429
public static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2, Observable<? extends T> t3, Observable<? extends T> t4, Observable<? extends T> t5, Observable<? extends T> t6, Observable<? extends T> t7, Observable<? extends T> t8, Observable<? extends T> t9) {
1445-
return create(OperationMerge.merge(t1, t2, t3, t4, t5, t6, t7, t8, t9));
1430+
return merge(from(t1, t2, t3, t4, t5, t6, t7, t8, t9));
1431+
}
1432+
1433+
/**
1434+
* Flattens a sequence of Observables emitted by an Observable into one Observable, without any transformation.
1435+
* The number of concurrent subscriptions to the Observables is limited by maxConcurrent.
1436+
* <p>
1437+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1438+
* <p>
1439+
* You can combine the items emitted by multiple Observables so that they
1440+
* act like a single Observable, by using the {@code merge} method.
1441+
*
1442+
* @param source an Observable that emits Observables
1443+
* @param maxConcurrent the maximum number of Observables being subscribed to concurrently
1444+
* @return an Observable that emits items that are the result of flattening
1445+
* the items emitted by the Observables emitted by the
1446+
* {@code source} Observable
1447+
* @throw IllegalArgumentException if maxConcurrent <= 0
1448+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1449+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh211914(v=vs.103).aspx">MSDN: Observable.Merge</a>
1450+
*/
1451+
public static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source, int maxConcurrent) {
1452+
return create(OperationMerge.merge(source, maxConcurrent));
1453+
}
1454+
1455+
/**
1456+
* Flattens an Observable Iterable into one Observable, without any transformation.
1457+
* <p>
1458+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1459+
* <p>
1460+
* You can combine the items emitted by multiple Observables so that they
1461+
* act like a single Observable, by using the {@code merge} method.
1462+
*
1463+
* @param sequences the Observable Iterable
1464+
* @return an Observable that emits items that are the result of flattening
1465+
* the items emitted by the Observables in the Iterable
1466+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1467+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229590(v=vs.103).aspx">MSDN: Observable.Merge</a>
1468+
*/
1469+
public static <T> Observable<T> merge(Iterable<? extends Observable<? extends T>> sequences) {
1470+
return merge(from(sequences));
1471+
}
1472+
1473+
/**
1474+
* Flattens an Observable Iterable into one Observable, without any transformation.
1475+
* The number of concurrent subscriptions to the Observables is limited by maxConcurrent.
1476+
* <p>
1477+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1478+
* <p>
1479+
* You can combine the items emitted by multiple Observables so that they
1480+
* act like a single Observable, by using the {@code merge} method.
1481+
*
1482+
* @param sequences the Observable Iterable
1483+
* @param maxConcurrent the maximum number of Observables being subscribed to concurrently
1484+
* @return an Observable that emits items that are the result of flattening
1485+
* the items emitted by the Observables in the Iterable
1486+
* @throw IllegalArgumentException if maxConcurrent <= 0
1487+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1488+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229923(v=vs.103).aspx">MSDN: Observable.Merge</a>
1489+
*/
1490+
public static <T> Observable<T> merge(Iterable<? extends Observable<? extends T>> sequences, int maxConcurrent) {
1491+
return merge(from(sequences), maxConcurrent);
1492+
}
1493+
1494+
/**
1495+
* Flattens an Observable Iterable into one Observable, without any transformation.
1496+
* The number of concurrent subscriptions to the Observables is limited by maxConcurrent.
1497+
* <p>
1498+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1499+
* <p>
1500+
* You can combine the items emitted by multiple Observables so that they
1501+
* act like a single Observable, by using the {@code merge} method.
1502+
*
1503+
* @param sequences the Observable Iterable
1504+
* @param maxConcurrent the maximum number of Observables being subscribed to concurrently
1505+
* @param scheduler the scheduler to traversal the Observable array on
1506+
* @return an Observable that emits items that are the result of flattening
1507+
* the items emitted by the Observables in the Iterable
1508+
* @throw IllegalArgumentException if maxConcurrent <= 0
1509+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1510+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh244329(v=vs.103).aspx">MSDN: Observable.Merge</a>
1511+
*/
1512+
public static <T> Observable<T> merge(Iterable<? extends Observable<? extends T>> sequences, int maxConcurrent, Scheduler scheduler) {
1513+
return merge(from(sequences, scheduler), maxConcurrent);
1514+
}
1515+
1516+
/**
1517+
* Flattens an Observable Iterable into one Observable, without any transformation.
1518+
* <p>
1519+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1520+
* <p>
1521+
* You can combine the items emitted by multiple Observables so that they
1522+
* act like a single Observable, by using the {@code merge} method.
1523+
*
1524+
* @param sequences the Observable Iterable
1525+
* @param scheduler the scheduler to traversal the Observable array on
1526+
* @return an Observable that emits items that are the result of flattening
1527+
* the items emitted by the Observables in the Iterable
1528+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1529+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh244336(v=vs.103).aspx">MSDN: Observable.Merge</a>
1530+
*/
1531+
public static <T> Observable<T> merge(Iterable<? extends Observable<? extends T>> sequences, Scheduler scheduler) {
1532+
return merge(from(sequences, scheduler));
1533+
}
1534+
1535+
/**
1536+
* Flattens an Observable array into one Observable, without any transformation.
1537+
* <p>
1538+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1539+
* <p>
1540+
* You can combine the items emitted by multiple Observables so that they
1541+
* act like a single Observable, by using the {@code merge} method.
1542+
*
1543+
* @param sequences the Observable array
1544+
* @return an Observable that emits items that are the result of flattening
1545+
* the items emitted by the Observables in the array
1546+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1547+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229099(v=vs.103).aspx">MSDN: Observable.Merge</a>
1548+
*/
1549+
public static <T> Observable<T> merge(Observable<? extends T>[] sequences) {
1550+
return merge(from(sequences));
1551+
}
1552+
1553+
/**
1554+
* Flattens an Observable array into one Observable, without any transformation.
1555+
* <p>
1556+
* <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/merge.png">
1557+
* <p>
1558+
* You can combine the items emitted by multiple Observables so that they
1559+
* act like a single Observable, by using the {@code merge} method.
1560+
*
1561+
* @param sequences the Observable array
1562+
* @param scheduler the scheduler to traversal the Observable array on
1563+
* @return an Observable that emits items that are the result of flattening
1564+
* the items emitted by the Observables in the array
1565+
* @see <a href="https://github.com/Netflix/RxJava/wiki/Combining-Observables#merge">RxJava Wiki: merge()</a>
1566+
* @see <a href="http://msdn.microsoft.com/en-us/library/hh229061(v=vs.103).aspx">MSDN: Observable.Merge</a>
1567+
*/
1568+
public static <T> Observable<T> merge(Observable<? extends T>[] sequences, Scheduler scheduler) {
1569+
return merge(from(sequences, scheduler));
14461570
}
14471571

14481572
/**

0 commit comments

Comments
 (0)