From 51d8c0cdb7caccf19aa743fcab8282c43c77a25b Mon Sep 17 00:00:00 2001
From: Tomasz Lelek <tomasz.lelek@allegrogroup.com>
Date: Wed, 1 Feb 2017 16:43:15 +0100
Subject: [PATCH] BEAL-572 onBackpressure operations

---
 .../com/baelding/rxjava/ComputeFunction.java  |  8 +++++
 .../rxjava/HotObservableOnBackPressure.java   | 31 +++++++++++++++++++
 2 files changed, 39 insertions(+)
 create mode 100644 rxjava/src/main/java/com/baelding/rxjava/HotObservableOnBackPressure.java

diff --git a/rxjava/src/main/java/com/baelding/rxjava/ComputeFunction.java b/rxjava/src/main/java/com/baelding/rxjava/ComputeFunction.java
index f83f34c3ee..557eb04612 100644
--- a/rxjava/src/main/java/com/baelding/rxjava/ComputeFunction.java
+++ b/rxjava/src/main/java/com/baelding/rxjava/ComputeFunction.java
@@ -34,4 +34,12 @@ public class ComputeFunction {
     }
 
 
+    public static void compute(Long v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
 }
diff --git a/rxjava/src/main/java/com/baelding/rxjava/HotObservableOnBackPressure.java b/rxjava/src/main/java/com/baelding/rxjava/HotObservableOnBackPressure.java
new file mode 100644
index 0000000000..f61536e827
--- /dev/null
+++ b/rxjava/src/main/java/com/baelding/rxjava/HotObservableOnBackPressure.java
@@ -0,0 +1,31 @@
+package com.baelding.rxjava;
+
+
+import rx.BackpressureOverflow;
+import rx.Observable;
+import rx.schedulers.Schedulers;
+
+import java.util.concurrent.TimeUnit;
+
+public class HotObservableOnBackPressure {
+    public static void main(String[] args) throws InterruptedException {
+        Observable.range(1, 1_000_000)
+                .onBackpressureBuffer(16, () -> {
+                        },
+                        BackpressureOverflow.ON_OVERFLOW_DROP_OLDEST)
+                .observeOn(Schedulers.computation())
+                .subscribe(e -> {
+                }, Throwable::printStackTrace);
+
+
+        Observable.interval(1, TimeUnit.MINUTES)
+                .onBackpressureDrop()
+//                .onBackpressureLatest()
+                .observeOn(Schedulers.io())
+                .doOnNext(ComputeFunction::compute)
+                .subscribe(v -> {
+                }, Throwable::printStackTrace);
+        Thread.sleep(10_000);
+
+    }
+}