From 466db837ba431e4f5a82bb036b62fae25faf696a Mon Sep 17 00:00:00 2001
From: longbkiter07 <long.bkiter07@gmail.com>
Date: Mon, 5 Dec 2016 12:08:59 +0700
Subject: [PATCH] * Fixed behavior buffer completing events

---
 config.gradle                                 |  2 +-
 .../ObservableAdapterManager.java             | 36 +++++++++----------
 2 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/config.gradle b/config.gradle
index 90837b3..6e69875 100644
--- a/config.gradle
+++ b/config.gradle
@@ -22,7 +22,7 @@ ext {
       minSdkVersion    : 15,
       targetSdkVersion : 23,
       versionCode      : 25,
-      versionName      : "0.3.2"
+      versionName      : "0.3.3"
   ]
 
   bintray = [
diff --git a/observablerm/src/main/java/me/silong/observablerm/ObservableAdapterManager.java b/observablerm/src/main/java/me/silong/observablerm/ObservableAdapterManager.java
index 40a7de7..ee6c8ac 100644
--- a/observablerm/src/main/java/me/silong/observablerm/ObservableAdapterManager.java
+++ b/observablerm/src/main/java/me/silong/observablerm/ObservableAdapterManager.java
@@ -13,6 +13,7 @@
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
 import rx.schedulers.Schedulers;
+import rx.subjects.BehaviorSubject;
 import rx.subjects.PublishSubject;
 
 /**
@@ -28,7 +29,7 @@ public class ObservableAdapterManager<D> {
 
   private final PublishSubject<Behavior<D>> mProcessingSubject;
 
-  private final PublishSubject<Long> mFinishedSubject;
+  private final BehaviorSubject<Behavior<D>> mFinishedSubject;
 
   @Nullable private RecyclerView.Adapter mAdapter;
 
@@ -38,7 +39,7 @@ public ObservableAdapterManager(@Nullable RecyclerView.Adapter adapter, List<D>
     mItems = items;
     mDataComparable = dataComparable;
     mProcessingSubject = PublishSubject.create();
-    mFinishedSubject = PublishSubject.create();
+    mFinishedSubject = BehaviorSubject.create();
     mAdapter = adapter;
     init();
   }
@@ -63,7 +64,7 @@ private void init() {
         .onBackpressureBuffer()
         .observeOn(Schedulers.computation())
         .concatMap(dBehavior -> processBehaviors(dBehavior).doOnNext(aVoid1 -> {
-          mFinishedSubject.onNext(dBehavior.createdAt);
+          mFinishedSubject.onNext(dBehavior);
         }))
         .subscribe();
   }
@@ -175,23 +176,25 @@ private boolean shouldCallUpdate(D oldItem, D newItem) {
   }
 
   private Observable<Void> processBehaviors(Behavior<D> behavior) {
-    if (behavior.mAction == Action.SET) {
-      if (mDataComparable != null && mItems.size() <= MAX_SIZE_TO_CALL_DIFF
-          && behavior.mItems.size() <= MAX_SIZE_TO_CALL_DIFF) {
-        return processSetWithDiffCallback(behavior);
+    return Observable.defer(() -> {
+      if (behavior.mAction == Action.SET) {
+        if (mDataComparable != null && mItems.size() <= MAX_SIZE_TO_CALL_DIFF
+            && behavior.mItems.size() <= MAX_SIZE_TO_CALL_DIFF) {
+          return processSetWithDiffCallback(behavior);
+        } else {
+          return processSetWithNotify(behavior);
+        }
       } else {
-        return processSetWithNotify(behavior);
+        return processSingleOperator(behavior);
       }
-    } else {
-      return processSingleOperator(behavior);
-    }
+    });
   }
 
   private Observable<Void> submitBehavior(Behavior<D> behavior) {
-    return Observable.defer(() -> mFinishedSubject
-        .doOnSubscribe(() -> mProcessingSubject.onNext(behavior))
-        .filter(aLong -> aLong >= behavior.createdAt).take(1)
-        .<Void>map(dBehavior -> null));
+    return mFinishedSubject.filter(dBehavior -> dBehavior == behavior)
+        .take(1)
+        .<Void>map(aLong -> null)
+        .doOnSubscribe(() -> mProcessingSubject.onNext(behavior));
   }
 
   public Observable<Void> add(D item) {
@@ -262,8 +265,6 @@ private static class Behavior<D> {
 
     final int mDestPos;
 
-    final long createdAt;
-
     public Behavior(D item, int pos, Action action) {
       this(Arrays.asList(item), pos, action, pos);
     }
@@ -285,7 +286,6 @@ public Behavior(List<? extends D> items, int pos, Action action, int destPos) {
       mPos = pos;
       mAction = action;
       mDestPos = destPos;
-      createdAt = System.currentTimeMillis();
     }
 
     public Behavior(D item, Action action) {