Add Observable filter operations

Author: LEMUBIT

src/main/java/FilteringObservables/First.java

@@ -0,0 +1,34 @@
+/*emit only the first item (or the first item that meets some condition) emitted by an Observable*/
+/*If you are only interested in the first item emitted by an Observable, or the first item that meets
+some criteria, you can filter the Observable with the First operator.*/
+/*n some implementations there is also a Single operator. It behaves similarly to First except that it
+waits until the source Observable terminates in order to guarantee that it only emits a single item
+(otherwise, rather than emitting that item, it terminates with an error). You can use this to not only
+take the first item from the source Observable but to also guarantee that there was only one item.*/
+
+package FilteringObservables;
+
+import rx.Observable;
+
+public class First {
+    public static void main(String[] args) {
+        Observable.just(8,4,5,6).first().subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error:"+throwable),
+                () -> System.out.println("Completed First()")
+        );
+
+        Observable.just(8, 2, 3, 4, 1, 15).first(integer -> integer>10).subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error:"+throwable),
+                () -> System.out.println("Completed First(func)")
+        );
+
+        Observable.just(8).single().subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error:"+throwable),
+                () -> System.out.println("Completed Single")
+        );
+
+    }
+}

src/main/java/FilteringObservables/IgnoreElements.java

@@ -0,0 +1,20 @@
+/*do not emit any items from an Observable but mirror its termination notification*/
+/*The IgnoreElements operator suppresses all of the items emitted by the source Observable,
+but allows its termination notification (either onError or onCompleted) to pass through unchanged.
+
+If you do not care about the items being emitted by an Observable, but you do want to be notified
+when it completes or when it terminates with an error, you can apply the ignoreElements operator to
+the Observable, which will ensure that it will never call its observers’ onNext handlers*/
+package FilteringObservables;
+
+import rx.Observable;
+
+public class IgnoreElements {
+    public static void main(String[] args) {
+        Observable.just(8,4,5,6).ignoreElements().subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("It had an Error"),
+                () -> System.out.println("It's Completed")
+        );
+    }
+}

src/main/java/FilteringObservables/Last.java

@@ -0,0 +1,30 @@
+/*emit only the last item (or the last item that meets some condition) emitted by an Observable*/
+/*If you are only interested in the last item emitted by an Observable, or the last item that meets some criteria,
+you can filter the Observable with the Last operator.
+
+In some implementations, Last is not implemented as a filtering operator that returns an Observable,
+but as a blocking function that returns a particular item when the source Observable terminates.
+In those implementations, if you instead want a filtering operator, you may have better luck with TakeLast(1).*/
+
+/*The BlockingObservable methods do not transform an Observable into another, filtered Observable,
+but rather they break out of the Observable cascade, blocking until the Observable emits the desired
+item, and then return that item itself.*/
+package FilteringObservables;
+
+import rx.Observable;
+
+public class Last {
+    public static void main(String[] args) {
+        Observable.just(8,4,5,6,3).last(integer -> integer%2==0).subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error"),
+                () -> System.out.println("Completed divisible by 2")
+        );
+
+        Observable.just(8,4,5,6,3).last().subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error"),
+                () -> System.out.println("Completed last")
+        );
+    }
+}

src/main/java/FilteringObservables/Sample.java

@@ -0,0 +1,32 @@
+/*emit the most recent items emitted by an Observable within periodic time intervals*/
+/*The Sample operator periodically looks at an Observable and emits whichever item it has most recently emitted since the previous sampling.
+
+In some implementations, there is also a ThrottleFirst operator that is similar,
+but emits not the most-recently emitted item in the sample period, but the first item that was emitted during that period.*/
+package FilteringObservables;
+
+import rx.Observable;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class Sample {
+    public static void main(String[] args) {
+        CountDownLatch latch = new CountDownLatch(1);
+        Observable.interval(1, TimeUnit.MILLISECONDS).sample(2,TimeUnit.SECONDS).subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error!"),
+                () -> {
+                    latch.countDown();
+                    System.out.println("Completed");
+                }
+        );
+
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

src/main/java/FilteringObservables/Skip.java

@@ -0,0 +1,39 @@
+/*You can ignore the first n items emitted by an Observable and attend only to those items that come after,
+by modifying the Observable with the Skip operator.*/
+package FilteringObservables;
+
+import rx.Observable;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class Skip {
+    public static void main(String[] args) throws InterruptedException {
+        CountDownLatch latch = new CountDownLatch(1);
+        Observable.interval(1, TimeUnit.SECONDS).take(5).skip(2).subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error!"),
+                () -> {
+                    latch.countDown();
+                    System.out.println("Completed");
+                }
+        );
+
+            latch.await();
+
+            //todo check
+        CountDownLatch latch2 = new CountDownLatch(1);
+        Observable.interval(1, TimeUnit.SECONDS).take(2).skip(2,TimeUnit.SECONDS).subscribe(
+                integer -> System.out.println(integer),
+                throwable -> System.out.println("Error!"),
+                () -> {
+                    latch2.countDown();
+                    System.out.println("Completed skip seconds");
+                }
+        );
+
+        latch2.await();
+
+    }
+
+}