- Bump gradle wrapper version.

- Bump android build tools.
- Bump android compile sdk version. 
- Disable frodo for now.

Author: android10

app/build.gradle

@@ -1,10 +1,10 @@
 apply plugin: 'com.android.application'
 apply plugin: 'me.tatarka.retrolambda'
-apply plugin: 'com.fernandocejas.frodo'
+//apply plugin: 'com.fernandocejas.frodo'
 
 android {
-  compileSdkVersion 23
-  buildToolsVersion "21.1.2"
+  compileSdkVersion 25
+  buildToolsVersion "23.0.2"
 
   defaultConfig {
     applicationId "com.fernandocejas.android10.rx.sample"
@@ -45,7 +45,7 @@ android {
 }
 
 dependencies {
-  compile 'com.android.support:recyclerview-v7:23.1.1'
+  compile 'com.android.support:recyclerview-v7:25.0.1'
   compile 'com.jakewharton:butterknife:7.0.1'
   compile 'com.squareup.okhttp:okhttp:2.5.0'
   compile 'io.reactivex:rxjava:1.0.14'

app/src/main/java/com/fernandocejas/android10/rx/sample/activity/backpressure/ActivityBackpressureSamples.java

@@ -93,26 +93,30 @@ private void toggleProgressBar(boolean visible) {
   }
 
   @OnClick(R.id.btn_backpressureException) void onBackpressureExceptionClick() {
-    dataManager.milliseconds(1000)
+    //dataManager.milliseconds(1000)
+    //    .observeOn(AndroidSchedulers.mainThread())
+    //    .subscribeOn(Schedulers.computation())
+    //    .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_exception)));
+    dataManager.milliseconds(100)
         .observeOn(AndroidSchedulers.mainThread())
         .subscribeOn(Schedulers.computation())
-        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_exception)));
+        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_exception), 15));
   }
 
   @OnClick(R.id.btn_backpressureDrop) void onBackpressureDropClick() {
     dataManager.milliseconds(1000)
         .onBackpressureDrop()
         .observeOn(AndroidSchedulers.mainThread())
         .subscribeOn(Schedulers.computation())
-        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_drop), 1000));
+        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_drop)));
   }
 
   @OnClick(R.id.btn_backpressureBuffer) void onBackpressureBuffer() {
     dataManager.milliseconds(600)
         .onBackpressureBuffer(200)
         .observeOn(AndroidSchedulers.mainThread())
         .subscribeOn(Schedulers.computation())
-        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_buffer), 600));
+        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_buffer)));
   }
 
   @OnClick(R.id.btn_backpressureWindow) void onBackpressureWindow() {
@@ -128,15 +132,15 @@ private void toggleProgressBar(boolean visible) {
         .toList()
         .observeOn(AndroidSchedulers.mainThread())
         .subscribeOn(Schedulers.computation())
-        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_toList), 1000));
+        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_toList)));
   }
 
   @OnClick(R.id.btn_backpressureThrottleLast) void onBackpressureThrottleLast() {
     dataManager.milliseconds(10000)
         .throttleLast(10, TimeUnit.MILLISECONDS)
         .observeOn(AndroidSchedulers.mainThread())
         .subscribeOn(Schedulers.computation())
-        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_throttleLast), 10000));
+        .subscribe(new BackpressureSubscriber<>(this, getString(R.string.btn_text_backpressure_throttleLast)));
   }
 
   @Override
@@ -145,7 +149,14 @@ public void onOperationStart(String name, long itemsRequested) {
     toggleProgressBar(true);
     sv_container.fullScroll(ScrollView.FOCUS_DOWN);
     tv_sampleDescription.setText(name);
-    tv_itemsRequested.setText(String.valueOf(itemsRequested));
+    //This check is because you can pass a magic number to request, request(Long.MAX_VALUE),
+    //to disable reactive pull backpressure and to ask the Observable to emit items at its own pace.
+    //https://github.com/ReactiveX/RxJava/wiki/Backpressure
+    if (itemsRequested != Long.MAX_VALUE) {
+      tv_itemsRequested.setText(String.valueOf(itemsRequested));
+    } else {
+      tv_itemsRequested.setText("Observable emitting at its own pace");
+    }
   }
 
   @Override public void onOperationProgress(long itemsProcessedSoFar) {

app/src/main/java/com/fernandocejas/android10/rx/sample/activity/backpressure/BackpressureSubscriber.java

@@ -1,12 +1,12 @@
 /**
  * Copyright (C) 2016 android10.org Open Source Project
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -15,80 +15,109 @@
  */
 package com.fernandocejas.android10.rx.sample.activity.backpressure;
 
-import com.fernandocejas.frodo.annotation.RxLogSubscriber;
-import java.lang.ref.WeakReference;
+import rx.Producer;
 import rx.Subscriber;
 import rx.exceptions.MissingBackpressureException;
 
-@RxLogSubscriber
+import java.lang.ref.WeakReference;
+
+//@RxLogSubscriber
 class BackpressureSubscriber<T> extends Subscriber<T> {
 
-  interface BackPressureResultListener {
-    void onOperationStart(String name, long itemsRequested);
-    void onOperationProgress(long itemsProcessedSoFar);
-    void onOperationResult(long itemsEmitted, OperationResult operation);
-  }
-
-  private final WeakReference<BackPressureResultListener> backPressureListener;
-  private final String name;
-  private final long itemsRequested;
-  private boolean shouldRequestItem;
-
-  private long itemsEmitted = 0;
-
-  BackpressureSubscriber(BackPressureResultListener listener, String name) {
-    this(listener, name, Long.MAX_VALUE);
-  }
-
-  BackpressureSubscriber(BackPressureResultListener listener, String name, long requestedItems) {
-    this.backPressureListener = new WeakReference<>(listener);
-    this.name = name;
-    this.itemsRequested = requestedItems;
-    this.shouldRequestItem = requestedItems != Long.MAX_VALUE;
-  }
-
-  @Override public void onStart() {
-    requestItems();
-    sendStartData();
-  }
-
-  @Override public void onNext(T item) {
-    itemsEmitted++;
-    sendProgressData(itemsEmitted);
-    requestItems();
-  }
-
-  @Override public void onCompleted() {
-    sendResultData(OperationResult.success());
-  }
-
-  @Override public void onError(Throwable throwable) {
-    if (throwable instanceof MissingBackpressureException) {
-      sendResultData(OperationResult.failure(throwable));
+    interface BackPressureResultListener {
+        void onOperationStart(String name, long itemsRequested);
+
+        void onOperationProgress(long itemsProcessedSoFar);
+
+        void onOperationResult(long itemsEmitted, OperationResult operation);
+    }
+
+    private final WeakReference<BackPressureResultListener> backPressureListener;
+    private final String name;
+    private final long itemsRequested;
+    private boolean shouldRequestItem;
+
+    private long itemsEmitted = 0;
+    private long itemsRequestedProcessed = 0;
+
+    BackpressureSubscriber(BackPressureResultListener listener, String name) {
+        this(listener, name, Long.MAX_VALUE);
     }
-  }
 
-  private void requestItems() {
-    if (shouldRequestItem) {
-      request(itemsRequested);
+    BackpressureSubscriber(BackPressureResultListener listener, String name, long requestedItems) {
+        this.backPressureListener = new WeakReference<>(listener);
+        this.name = name;
+        this.itemsRequested = requestedItems;
+        this.shouldRequestItem = requestedItems != Long.MAX_VALUE;
     }
-  }
 
-  private void sendStartData() {
-    if (backPressureListener.get() != null) {
-      backPressureListener.get().onOperationStart(name, itemsRequested);
+    @Override
+    public void onStart() {
+        if (shouldRequestItem) {
+            //setProducer(new QueuedValueProducer<>(this));
+//            setProducer(new BackPressureProducer());
+            request(itemsRequested);
+        }
+        sendStartData();
     }
-  }
 
-  private void sendProgressData(long itemsProcessedSoFar) {
-    if (backPressureListener.get() != null) {
-      backPressureListener.get().onOperationProgress(itemsProcessedSoFar);
+    @Override
+    public void onNext(T item) {
+        itemsEmitted++;
+        sendProgressData(itemsEmitted);
+        requestMoreItems();
     }
-  }
 
-  private void sendResultData(OperationResult operation) {
-    if (backPressureListener.get() != null) {
-      backPressureListener.get().onOperationResult(itemsEmitted, operation);
+    @Override
+    public void onCompleted() {
+        sendResultData(OperationResult.success());
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        if (throwable instanceof MissingBackpressureException) {
+            sendResultData(OperationResult.failure(throwable));
+        }
+    }
+
+    private void requestMoreItems() {
+        itemsRequestedProcessed++;
+        if (shouldRequestItem && itemsRequestedProcessed >= itemsRequested) {
+            request(itemsRequested);
+            itemsRequestedProcessed = 0;
+        }
+    }
+
+    private void sendStartData() {
+        if (backPressureListener.get() != null) {
+            backPressureListener.get().onOperationStart(name, itemsRequested);
+        }
+    }
+
+    private void sendProgressData(long itemsProcessedSoFar) {
+        if (backPressureListener.get() != null) {
+            backPressureListener.get().onOperationProgress(itemsProcessedSoFar);
+        }
+    }
+
+    private void sendResultData(OperationResult operation) {
+        if (backPressureListener.get() != null) {
+            backPressureListener.get().onOperationResult(itemsEmitted, operation);
+        }
+    }
+
+    private static class BackPressureProducer<T> implements Producer {
+        private final Subscriber<T> subscriber;
+
+        BackPressureProducer(Subscriber subscriber) {
+            this.subscriber = subscriber;
+        }
+
+        @Override
+        public void request(long requested) {
+            for (int i = 0; i < requested && !subscriber.isUnsubscribed(); i++) {
+//                subscriber.onNext(T);
+            }
+        }
     }
-  }
 }

app/src/main/java/com/fernandocejas/android10/rx/sample/data/DataManager.java

@@ -38,6 +38,10 @@ public Observable<Integer> numbers() {
     return Observable.from(numberGenerator.numbers());
   }
 
+  public Observable<Long> numbers(int upUntil) {
+    return Observable.from(numberGenerator.numbers(upUntil));
+  }
+
   public Observable<Long> milliseconds(int upUntil) {
     return Observable.interval(0, 1, TimeUnit.MILLISECONDS).take(upUntil);
   }

app/src/main/java/com/fernandocejas/android10/rx/sample/data/NumberGenerator.java

@@ -26,4 +26,12 @@ public NumberGenerator() {}
   List<Integer> numbers() {
     return new ArrayList<>(Arrays.asList(2, 3, 4, 5, 6, 7, 8, 9, 10));
   }
+
+  List<Long> numbers(int upUntil) {
+    final ArrayList<Long> numberList = new ArrayList<>(upUntil);
+    for (long i = 0; i < upUntil; i++) {
+      numberList.add(i);
+    }
+    return numberList;
+  }
 }

build.gradle

@@ -5,9 +5,9 @@ buildscript {
     mavenLocal()
   }
   dependencies {
-    classpath 'com.android.tools.build:gradle:2.1.0'
+    classpath 'com.android.tools.build:gradle:2.2.2'
     classpath 'me.tatarka:gradle-retrolambda:3.2.2'
-    classpath "com.fernandocejas.frodo:frodo-plugin:0.8.1"
+//    classpath "com.fernandocejas.frodo:frodo-plugin:0.8.1"
   }
 }
 
@@ -18,3 +18,17 @@ allprojects {
     mavenLocal()
   }
 }
+
+task deployDebug(type: Exec, dependsOn: 'app:installDebug') {
+  def rootDir = project.rootDir
+  def localProperties = new File(rootDir, "local.properties")
+  if (localProperties.exists()) {
+    Properties properties = new Properties()
+    localProperties.withInputStream {
+      inputStream -> properties.load(inputStream)
+    }
+    def sdkDir = properties.getProperty('sdk.dir')
+    def adb = "$sdkDir/platform-tools/adb"
+    commandLine "$adb", 'shell', 'am', 'start', '-n', 'com.fernandocejas.android10.rx.sample/com.fernandocejas.android10.rx.sample.activity.MainActivity'
+  }
+}

gradle.properties

@@ -10,9 +10,9 @@
 # Specifies the JVM arguments used for the daemon process.
 # The setting is particularly useful for tweaking memory settings.
 # Default value: -Xmx10248m -XX:MaxPermSize=256m
-# org.gradle.jvmargs=-Xmx2048m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError -Dfile.encoding=UTF-8
+org.gradle.jvmargs=-Xmx2048m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError -Dfile.encoding=UTF-8
 
 # When configured, Gradle will run in incubating parallel mode.
 # This option should only be used with decoupled projects. More details, visit
 # http://www.gradle.org/docs/current/userguide/multi_project_builds.html#sec:decoupled_projects
-# org.gradle.parallel=true
+org.gradle.parallel=true

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,6 @@
-#Fri May 06 11:27:41 CEST 2016
+#Sat Dec 10 22:13:26 CET 2016
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-2.12-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-3.2.1-all.zip