Gradle update (#52)

Versions in constants

build.gradle

@@ -1,9 +1,7 @@
 apply plugin: 'com.android.library'
-apply plugin: 'me.tatarka.retrolambda'
 
 android {
-    compileSdkVersion 25
-    buildToolsVersion '25.0.3'
+    compileSdkVersion compileSdk
 
     compileOptions {
         sourceCompatibility JavaVersion.VERSION_1_8
@@ -16,6 +14,7 @@ android {
 }
 
 dependencies {
-    provided 'com.android.support:support-annotations:25.3.1'
-    provided 'io.reactivex:rxandroid:1.2.1'
+    compileOnly "com.android.support:support-annotations:$supportLibraryVersion"
+    compileOnly "io.reactivex:rxandroid:$rxAndroidVersion"
+    compileOnly "io.reactivex:rxjava:$rxJavaVersion"
 }

src/main/java/ru/touchin/roboswag/core/android/support/v7/util/BatchingListUpdateCallback.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright (C) 2016 The Android Open Source Project
+ *
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ru.touchin.roboswag.core.android.support.v7.util;
+
+/**
+ * Wraps a {@link ListUpdateCallback} callback and batches operations that can be merged.
+ *
+ * <p>For instance, when 2 add operations comes that adds 2 consecutive elements,
+ * BatchingListUpdateCallback merges them and calls the wrapped callback only once.
+ *
+ * <p>This is a general purpose class and is also used by
+ * {@link DiffUtil.DiffResult DiffResult}
+ *
+ * <p>If you use this class to batch updates, you must call {@link #dispatchLastEvent()} when the
+ * stream of update events drain.
+ */
+@SuppressWarnings({"PMD", "checkstyle:all"})
+public class BatchingListUpdateCallback implements ListUpdateCallback {
+
+    private static final int TYPE_NONE = 0;
+    private static final int TYPE_ADD = 1;
+    private static final int TYPE_REMOVE = 2;
+    private static final int TYPE_CHANGE = 3;
+
+    private final ListUpdateCallback mWrapped;
+
+    private int mLastEventType = TYPE_NONE;
+    private int mLastEventPosition = -1;
+    private int mLastEventCount = -1;
+    private Object mLastEventPayload = null;
+
+    public BatchingListUpdateCallback(ListUpdateCallback callback) {
+        mWrapped = callback;
+    }
+
+    /**
+     * BatchingListUpdateCallback holds onto the last event to see if it can be merged with the
+     * next one. When stream of events finish, you should call this method to dispatch the last
+     * event.
+     */
+    public void dispatchLastEvent() {
+        if (mLastEventType == TYPE_NONE) {
+            return;
+        }
+        switch (mLastEventType) {
+            case TYPE_ADD:
+                mWrapped.onInserted(mLastEventPosition, mLastEventCount);
+                break;
+            case TYPE_REMOVE:
+                mWrapped.onRemoved(mLastEventPosition, mLastEventCount);
+                break;
+            case TYPE_CHANGE:
+                mWrapped.onChanged(mLastEventPosition, mLastEventCount, mLastEventPayload);
+                break;
+        }
+        mLastEventPayload = null;
+        mLastEventType = TYPE_NONE;
+    }
+
+    @Override
+    public void onInserted(int position, int count) {
+        if (mLastEventType == TYPE_ADD && position >= mLastEventPosition
+                && position <= mLastEventPosition + mLastEventCount) {
+            mLastEventCount += count;
+            mLastEventPosition = Math.min(position, mLastEventPosition);
+            return;
+        }
+        dispatchLastEvent();
+        mLastEventPosition = position;
+        mLastEventCount = count;
+        mLastEventType = TYPE_ADD;
+    }
+
+    @Override
+    public void onRemoved(int position, int count) {
+        if (mLastEventType == TYPE_REMOVE && mLastEventPosition >= position &&
+                mLastEventPosition <= position + count) {
+            mLastEventCount += count;
+            mLastEventPosition = position;
+            return;
+        }
+        dispatchLastEvent();
+        mLastEventPosition = position;
+        mLastEventCount = count;
+        mLastEventType = TYPE_REMOVE;
+    }
+
+    @Override
+    public void onMoved(int fromPosition, int toPosition) {
+        dispatchLastEvent(); // moves are not merged
+        mWrapped.onMoved(fromPosition, toPosition);
+    }
+
+    @Override
+    public void onChanged(int position, int count, Object payload) {
+        if (mLastEventType == TYPE_CHANGE &&
+                !(position > mLastEventPosition + mLastEventCount
+                        || position + count < mLastEventPosition || mLastEventPayload != payload)) {
+            // take potential overlap into account
+            int previousEnd = mLastEventPosition + mLastEventCount;
+            mLastEventPosition = Math.min(position, mLastEventPosition);
+            mLastEventCount = Math.max(previousEnd, position + count) - mLastEventPosition;
+            return;
+        }
+        dispatchLastEvent();
+        mLastEventPosition = position;
+        mLastEventCount = count;
+        mLastEventPayload = payload;
+        mLastEventType = TYPE_CHANGE;
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/android/support/v7/util/DiffUtil.java

@@ -0,0 +1,783 @@
+/*
+ * Copyright (C) 2016 The Android Open Source Project
+ *
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ru.touchin.roboswag.core.android.support.v7.util;
+
+import android.support.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * DiffUtil is a utility class that can calculate the difference between two lists and output a
+ * list of update operations that converts the first list into the second one.
+ *
+ * <p>It can be used to calculate updates for a RecyclerView Adapter.
+ *
+ * <p>DiffUtil uses Eugene W. Myers's difference algorithm to calculate the minimal number of updates
+ * to convert one list into another. Myers's algorithm does not handle items that are moved so
+ * DiffUtil runs a second pass on the result to detect items that were moved.
+ *
+ * <p>If the lists are large, this operation may take significant time so you are advised to run this
+ * on a background thread, get the {@link DiffResult} then apply it on the RecyclerView on the main
+ * thread.
+ *
+ * <p>This algorithm is optimized for space and uses O(N) space to find the minimal
+ * number of addition and removal operations between the two lists. It has O(N + D^2) expected time
+ * performance where D is the length of the edit script.
+ *
+ * <p>If move detection is enabled, it takes an additional O(N^2) time where N is the total number of
+ * added and removed items. If your lists are already sorted by the same constraint (e.g. a created
+ * timestamp for a list of posts), you can disable move detection to improve performance.
+ *
+ * <p>The actual runtime of the algorithm significantly depends on the number of changes in the list
+ * and the cost of your comparison methods. Below are some average run times for reference:
+ * (The areSame list is composed of random UUID Strings and the tests are run on Nexus 5X with M)
+ * <ul>
+ * <li>100 items and 10 modifications: avg: 0.39 ms, median: 0.35 ms
+ * <li>100 items and 100 modifications: 3.82 ms, median: 3.75 ms
+ * <li>100 items and 100 modifications without moves: 2.09 ms, median: 2.06 ms
+ * <li>1000 items and 50 modifications: avg: 4.67 ms, median: 4.59 ms
+ * <li>1000 items and 50 modifications without moves: avg: 3.59 ms, median: 3.50 ms
+ * <li>1000 items and 200 modifications: 27.07 ms, median: 26.92 ms
+ * <li>1000 items and 200 modifications without moves: 13.54 ms, median: 13.36 ms
+ * </ul>
+ *
+ * <p>Due to implementation constraints, the max size of the list can be 2^26.
+ */
+@SuppressWarnings({"PMD", "checkstyle:all"})
+public class DiffUtil {
+
+    private DiffUtil() {
+        // utility class, no instance.
+    }
+
+    private static final Comparator<Snake> SNAKE_COMPARATOR = new Comparator<Snake>() {
+        @Override
+        public int compare(Snake o1, Snake o2) {
+            int cmpX = o1.x - o2.x;
+            return cmpX == 0 ? o1.y - o2.y : cmpX;
+        }
+    };
+
+    // Myers' algorithm uses two lists as axis labels. In DiffUtil's implementation, `x` axis is
+    // used for old list and `y` axis is used for new list.
+
+    /**
+     * Calculates the list of update operations that can covert one list into the other one.
+     *
+     * @param cb The callback that acts as a gateway to the backing list data
+     * @return A DiffResult that contains the information about the edit sequence to convert the
+     * old list into the new list.
+     */
+    public static DiffResult calculateDiff(Callback cb) {
+        return calculateDiff(cb, true);
+    }
+
+    /**
+     * Calculates the list of update operations that can covert one list into the other one.
+     * <p>
+     * If your old and new lists are sorted by the same constraint and items never move (swap
+     * positions), you can disable move detection which takes <code>O(N^2)</code> time where
+     * N is the number of added, moved, removed items.
+     *
+     * @param cb          The callback that acts as a gateway to the backing list data
+     * @param detectMoves True if DiffUtil should try to detect moved items, false otherwise.
+     * @return A DiffResult that contains the information about the edit sequence to convert the
+     * old list into the new list.
+     */
+    public static DiffResult calculateDiff(Callback cb, boolean detectMoves) {
+        final int oldSize = cb.getOldListSize();
+        final int newSize = cb.getNewListSize();
+
+        final List<Snake> snakes = new ArrayList<>();
+
+        // instead of a recursive implementation, we keep our own stack to avoid potential stack
+        // overflow exceptions
+        final List<Range> stack = new ArrayList<>();
+
+        stack.add(new Range(0, oldSize, 0, newSize));
+
+        final int max = oldSize + newSize + Math.abs(oldSize - newSize);
+        // allocate forward and backward k-lines. K lines are diagonal lines in the matrix. (see the
+        // paper for details)
+        // These arrays lines keep the max reachable position for each k-line.
+        final int[] forward = new int[max * 2];
+        final int[] backward = new int[max * 2];
+
+        // We pool the ranges to avoid allocations for each recursive call.
+        final List<Range> rangePool = new ArrayList<>();
+        while (!stack.isEmpty()) {
+            final Range range = stack.remove(stack.size() - 1);
+            final Snake snake = diffPartial(cb, range.oldListStart, range.oldListEnd,
+                    range.newListStart, range.newListEnd, forward, backward, max);
+            if (snake != null) {
+                if (snake.size > 0) {
+                    snakes.add(snake);
+                }
+                // offset the snake to convert its coordinates from the Range's area to global
+                snake.x += range.oldListStart;
+                snake.y += range.newListStart;
+
+                // add new ranges for left and right
+                final Range left = rangePool.isEmpty() ? new Range() : rangePool.remove(
+                        rangePool.size() - 1);
+                left.oldListStart = range.oldListStart;
+                left.newListStart = range.newListStart;
+                if (snake.reverse) {
+                    left.oldListEnd = snake.x;
+                    left.newListEnd = snake.y;
+                } else {
+                    if (snake.removal) {
+                        left.oldListEnd = snake.x - 1;
+                        left.newListEnd = snake.y;
+                    } else {
+                        left.oldListEnd = snake.x;
+                        left.newListEnd = snake.y - 1;
+                    }
+                }
+                stack.add(left);
+
+                // re-use range for right
+                //noinspection UnnecessaryLocalVariable
+                final Range right = range;
+                if (snake.reverse) {
+                    if (snake.removal) {
+                        right.oldListStart = snake.x + snake.size + 1;
+                        right.newListStart = snake.y + snake.size;
+                    } else {
+                        right.oldListStart = snake.x + snake.size;
+                        right.newListStart = snake.y + snake.size + 1;
+                    }
+                } else {
+                    right.oldListStart = snake.x + snake.size;
+                    right.newListStart = snake.y + snake.size;
+                }
+                stack.add(right);
+            } else {
+                rangePool.add(range);
+            }
+
+        }
+        // sort snakes
+        Collections.sort(snakes, SNAKE_COMPARATOR);
+
+        return new DiffResult(cb, snakes, forward, backward, detectMoves);
+
+    }
+
+    private static Snake diffPartial(Callback cb, int startOld, int endOld,
+                                     int startNew, int endNew, int[] forward, int[] backward, int kOffset) {
+        final int oldSize = endOld - startOld;
+        final int newSize = endNew - startNew;
+
+        if (endOld - startOld < 1 || endNew - startNew < 1) {
+            return null;
+        }
+
+        final int delta = oldSize - newSize;
+        final int dLimit = (oldSize + newSize + 1) / 2;
+        Arrays.fill(forward, kOffset - dLimit - 1, kOffset + dLimit + 1, 0);
+        Arrays.fill(backward, kOffset - dLimit - 1 + delta, kOffset + dLimit + 1 + delta, oldSize);
+        final boolean checkInFwd = delta % 2 != 0;
+        for (int d = 0; d <= dLimit; d++) {
+            for (int k = -d; k <= d; k += 2) {
+                // find forward path
+                // we can reach k from k - 1 or k + 1. Check which one is further in the graph
+                int x;
+                final boolean removal;
+                if (k == -d || k != d && forward[kOffset + k - 1] < forward[kOffset + k + 1]) {
+                    x = forward[kOffset + k + 1];
+                    removal = false;
+                } else {
+                    x = forward[kOffset + k - 1] + 1;
+                    removal = true;
+                }
+                // set y based on x
+                int y = x - k;
+                // move diagonal as long as items match
+                while (x < oldSize && y < newSize
+                        && cb.areItemsTheSame(startOld + x, startNew + y)) {
+                    x++;
+                    y++;
+                }
+                forward[kOffset + k] = x;
+                if (checkInFwd && k >= delta - d + 1 && k <= delta + d - 1) {
+                    if (forward[kOffset + k] >= backward[kOffset + k]) {
+                        Snake outSnake = new Snake();
+                        outSnake.x = backward[kOffset + k];
+                        outSnake.y = outSnake.x - k;
+                        outSnake.size = forward[kOffset + k] - backward[kOffset + k];
+                        outSnake.removal = removal;
+                        outSnake.reverse = false;
+                        return outSnake;
+                    }
+                }
+            }
+            for (int k = -d; k <= d; k += 2) {
+                // find reverse path at k + delta, in reverse
+                final int backwardK = k + delta;
+                int x;
+                final boolean removal;
+                if (backwardK == d + delta || backwardK != -d + delta
+                        && backward[kOffset + backwardK - 1] < backward[kOffset + backwardK + 1]) {
+                    x = backward[kOffset + backwardK - 1];
+                    removal = false;
+                } else {
+                    x = backward[kOffset + backwardK + 1] - 1;
+                    removal = true;
+                }
+
+                // set y based on x
+                int y = x - backwardK;
+                // move diagonal as long as items match
+                while (x > 0 && y > 0
+                        && cb.areItemsTheSame(startOld + x - 1, startNew + y - 1)) {
+                    x--;
+                    y--;
+                }
+                backward[kOffset + backwardK] = x;
+                if (!checkInFwd && k + delta >= -d && k + delta <= d) {
+                    if (forward[kOffset + backwardK] >= backward[kOffset + backwardK]) {
+                        Snake outSnake = new Snake();
+                        outSnake.x = backward[kOffset + backwardK];
+                        outSnake.y = outSnake.x - backwardK;
+                        outSnake.size =
+                                forward[kOffset + backwardK] - backward[kOffset + backwardK];
+                        outSnake.removal = removal;
+                        outSnake.reverse = true;
+                        return outSnake;
+                    }
+                }
+            }
+        }
+        throw new IllegalStateException("DiffUtil hit an unexpected case while trying to calculate"
+                + " the optimal path. Please make sure your data is not changing during the"
+                + " diff calculation.");
+    }
+
+    /**
+     * A Callback class used by DiffUtil while calculating the diff between two lists.
+     */
+    public abstract static class Callback {
+        /**
+         * Returns the size of the old list.
+         *
+         * @return The size of the old list.
+         */
+        public abstract int getOldListSize();
+
+        /**
+         * Returns the size of the new list.
+         *
+         * @return The size of the new list.
+         */
+        public abstract int getNewListSize();
+
+        /**
+         * Called by the DiffUtil to decide whether two object represent the same Item.
+         * <p>
+         * For example, if your items have unique ids, this method should check their id equality.
+         *
+         * @param oldItemPosition The position of the item in the old list
+         * @param newItemPosition The position of the item in the new list
+         * @return True if the two items represent the same object or false if they are different.
+         */
+        public abstract boolean areItemsTheSame(int oldItemPosition, int newItemPosition);
+
+        /**
+         * Called by the DiffUtil when it wants to check whether two items have the same data.
+         * DiffUtil uses this information to detect if the contents of an item has changed.
+         * <p>
+         * DiffUtil uses this method to check equality instead of {@link Object#equals(Object)}
+         * so that you can change its behavior depending on your UI.
+         * <p>
+         * This method is called only if {@link #areItemsTheSame(int, int)} returns
+         * {@code true} for these items.
+         *
+         * @param oldItemPosition The position of the item in the old list
+         * @param newItemPosition The position of the item in the new list which replaces the
+         *                        oldItem
+         * @return True if the contents of the items are the same or false if they are different.
+         */
+        public abstract boolean areContentsTheSame(int oldItemPosition, int newItemPosition);
+
+        /**
+         * When {@link #areItemsTheSame(int, int)} returns {@code true} for two items and
+         * {@link #areContentsTheSame(int, int)} returns false for them, DiffUtil
+         * calls this method to get a payload about the change.
+         * <p>
+         * Default implementation returns {@code null}.
+         *
+         * @param oldItemPosition The position of the item in the old list
+         * @param newItemPosition The position of the item in the new list
+         * @return A payload object that represents the change between the two items.
+         */
+        @Nullable
+        public Object getChangePayload(int oldItemPosition, int newItemPosition) {
+            return null;
+        }
+    }
+
+    /**
+     * Snakes represent a match between two lists. It is optionally prefixed or postfixed with an
+     * add or remove operation. See the Myers' paper for details.
+     */
+    static class Snake {
+        /**
+         * Position in the old list
+         */
+        int x;
+
+        /**
+         * Position in the new list
+         */
+        int y;
+
+        /**
+         * Number of matches. Might be 0.
+         */
+        int size;
+
+        /**
+         * If true, this is a removal from the original list followed by {@code size} matches.
+         * If false, this is an addition from the new list followed by {@code size} matches.
+         */
+        boolean removal;
+
+        /**
+         * If true, the addition or removal is at the end of the snake.
+         * If false, the addition or removal is at the beginning of the snake.
+         */
+        boolean reverse;
+    }
+
+    /**
+     * Represents a range in two lists that needs to be solved.
+     * <p>
+     * This internal class is used when running Myers' algorithm without recursion.
+     */
+    static class Range {
+
+        int oldListStart, oldListEnd;
+
+        int newListStart, newListEnd;
+
+        public Range() {
+        }
+
+        public Range(int oldListStart, int oldListEnd, int newListStart, int newListEnd) {
+            this.oldListStart = oldListStart;
+            this.oldListEnd = oldListEnd;
+            this.newListStart = newListStart;
+            this.newListEnd = newListEnd;
+        }
+    }
+
+    /**
+     * This class holds the information about the result of a
+     * {@link DiffUtil#calculateDiff(Callback, boolean)} call.
+     * <p>
+     * You can consume the updates in a DiffResult via
+     * {@link #dispatchUpdatesTo(ListUpdateCallback)}
+     */
+    public static class DiffResult {
+        /**
+         * While reading the flags below, keep in mind that when multiple items move in a list,
+         * Myers's may pick any of them as the anchor item and consider that one NOT_CHANGED while
+         * picking others as additions and removals. This is completely fine as we later detect
+         * all moves.
+         * <p>
+         * Below, when an item is mentioned to stay in the same "location", it means we won't
+         * dispatch a move/add/remove for it, it DOES NOT mean the item is still in the same
+         * position.
+         */
+        // item stayed the same.
+        private static final int FLAG_NOT_CHANGED = 1;
+        // item stayed in the same location but changed.
+        private static final int FLAG_CHANGED = FLAG_NOT_CHANGED << 1;
+        // Item has moved and also changed.
+        private static final int FLAG_MOVED_CHANGED = FLAG_CHANGED << 1;
+        // Item has moved but did not change.
+        private static final int FLAG_MOVED_NOT_CHANGED = FLAG_MOVED_CHANGED << 1;
+        // Ignore this update.
+        // If this is an addition from the new list, it means the item is actually removed from an
+        // earlier position and its move will be dispatched when we process the matching removal
+        // from the old list.
+        // If this is a removal from the old list, it means the item is actually added back to an
+        // earlier index in the new list and we'll dispatch its move when we are processing that
+        // addition.
+        private static final int FLAG_IGNORE = FLAG_MOVED_NOT_CHANGED << 1;
+
+        // since we are re-using the int arrays that were created in the Myers' step, we mask
+        // change flags
+        private static final int FLAG_OFFSET = 5;
+
+        private static final int FLAG_MASK = (1 << FLAG_OFFSET) - 1;
+
+        // The Myers' snakes. At this point, we only care about their diagonal sections.
+        private final List<Snake> mSnakes;
+
+        // The list to keep oldItemStatuses. As we traverse old items, we assign flags to them
+        // which also includes whether they were a real removal or a move (and its new index).
+        private final int[] mOldItemStatuses;
+        // The list to keep newItemStatuses. As we traverse new items, we assign flags to them
+        // which also includes whether they were a real addition or a move(and its old index).
+        private final int[] mNewItemStatuses;
+        // The callback that was given to calcualte diff method.
+        private final Callback mCallback;
+
+        private final int mOldListSize;
+
+        private final int mNewListSize;
+
+        private final boolean mDetectMoves;
+
+        /**
+         * @param callback        The callback that was used to calculate the diff
+         * @param snakes          The list of Myers' snakes
+         * @param oldItemStatuses An int[] that can be re-purposed to keep metadata
+         * @param newItemStatuses An int[] that can be re-purposed to keep metadata
+         * @param detectMoves     True if this DiffResult will try to detect moved items
+         */
+        DiffResult(Callback callback, List<Snake> snakes, int[] oldItemStatuses,
+                   int[] newItemStatuses, boolean detectMoves) {
+            mSnakes = snakes;
+            mOldItemStatuses = oldItemStatuses;
+            mNewItemStatuses = newItemStatuses;
+            Arrays.fill(mOldItemStatuses, 0);
+            Arrays.fill(mNewItemStatuses, 0);
+            mCallback = callback;
+            mOldListSize = callback.getOldListSize();
+            mNewListSize = callback.getNewListSize();
+            mDetectMoves = detectMoves;
+            addRootSnake();
+            findMatchingItems();
+        }
+
+        /**
+         * We always add a Snake to 0/0 so that we can run loops from end to beginning and be done
+         * when we run out of snakes.
+         */
+        private void addRootSnake() {
+            Snake firstSnake = mSnakes.isEmpty() ? null : mSnakes.get(0);
+            if (firstSnake == null || firstSnake.x != 0 || firstSnake.y != 0) {
+                Snake root = new Snake();
+                root.x = 0;
+                root.y = 0;
+                root.removal = false;
+                root.size = 0;
+                root.reverse = false;
+                mSnakes.add(0, root);
+            }
+        }
+
+        /**
+         * This method traverses each addition / removal and tries to match it to a previous
+         * removal / addition. This is how we detect move operations.
+         * <p>
+         * This class also flags whether an item has been changed or not.
+         * <p>
+         * DiffUtil does this pre-processing so that if it is running on a big list, it can be moved
+         * to background thread where most of the expensive stuff will be calculated and kept in
+         * the statuses maps. DiffResult uses this pre-calculated information while dispatching
+         * the updates (which is probably being called on the main thread).
+         */
+        private void findMatchingItems() {
+            int posOld = mOldListSize;
+            int posNew = mNewListSize;
+            // traverse the matrix from right bottom to 0,0.
+            for (int i = mSnakes.size() - 1; i >= 0; i--) {
+                final Snake snake = mSnakes.get(i);
+                final int endX = snake.x + snake.size;
+                final int endY = snake.y + snake.size;
+                if (mDetectMoves) {
+                    while (posOld > endX) {
+                        // this is a removal. Check remaining snakes to see if this was added before
+                        findAddition(posOld, posNew, i);
+                        posOld--;
+                    }
+                    while (posNew > endY) {
+                        // this is an addition. Check remaining snakes to see if this was removed
+                        // before
+                        findRemoval(posOld, posNew, i);
+                        posNew--;
+                    }
+                }
+                for (int j = 0; j < snake.size; j++) {
+                    // matching items. Check if it is changed or not
+                    final int oldItemPos = snake.x + j;
+                    final int newItemPos = snake.y + j;
+                    final boolean theSame = mCallback
+                            .areContentsTheSame(oldItemPos, newItemPos);
+                    final int changeFlag = theSame ? FLAG_NOT_CHANGED : FLAG_CHANGED;
+                    mOldItemStatuses[oldItemPos] = (newItemPos << FLAG_OFFSET) | changeFlag;
+                    mNewItemStatuses[newItemPos] = (oldItemPos << FLAG_OFFSET) | changeFlag;
+                }
+                posOld = snake.x;
+                posNew = snake.y;
+            }
+        }
+
+        private void findAddition(int x, int y, int snakeIndex) {
+            if (mOldItemStatuses[x - 1] != 0) {
+                return; // already set by a latter item
+            }
+            findMatchingItem(x, y, snakeIndex, false);
+        }
+
+        private void findRemoval(int x, int y, int snakeIndex) {
+            if (mNewItemStatuses[y - 1] != 0) {
+                return; // already set by a latter item
+            }
+            findMatchingItem(x, y, snakeIndex, true);
+        }
+
+        /**
+         * Finds a matching item that is before the given coordinates in the matrix
+         * (before : left and above).
+         *
+         * @param x          The x position in the matrix (position in the old list)
+         * @param y          The y position in the matrix (position in the new list)
+         * @param snakeIndex The current snake index
+         * @param removal    True if we are looking for a removal, false otherwise
+         * @return True if such item is found.
+         */
+        private boolean findMatchingItem(final int x, final int y, final int snakeIndex,
+                                         final boolean removal) {
+            final int myItemPos;
+            int curX;
+            int curY;
+            if (removal) {
+                myItemPos = y - 1;
+                curX = x;
+                curY = y - 1;
+            } else {
+                myItemPos = x - 1;
+                curX = x - 1;
+                curY = y;
+            }
+            for (int i = snakeIndex; i >= 0; i--) {
+                final Snake snake = mSnakes.get(i);
+                final int endX = snake.x + snake.size;
+                final int endY = snake.y + snake.size;
+                if (removal) {
+                    // check removals for a match
+                    for (int pos = curX - 1; pos >= endX; pos--) {
+                        if (mCallback.areItemsTheSame(pos, myItemPos)) {
+                            // found!
+                            final boolean theSame = mCallback.areContentsTheSame(pos, myItemPos);
+                            final int changeFlag = theSame ? FLAG_MOVED_NOT_CHANGED
+                                    : FLAG_MOVED_CHANGED;
+                            mNewItemStatuses[myItemPos] = (pos << FLAG_OFFSET) | FLAG_IGNORE;
+                            mOldItemStatuses[pos] = (myItemPos << FLAG_OFFSET) | changeFlag;
+                            return true;
+                        }
+                    }
+                } else {
+                    // check for additions for a match
+                    for (int pos = curY - 1; pos >= endY; pos--) {
+                        if (mCallback.areItemsTheSame(myItemPos, pos)) {
+                            // found
+                            final boolean theSame = mCallback.areContentsTheSame(myItemPos, pos);
+                            final int changeFlag = theSame ? FLAG_MOVED_NOT_CHANGED
+                                    : FLAG_MOVED_CHANGED;
+                            mOldItemStatuses[x - 1] = (pos << FLAG_OFFSET) | FLAG_IGNORE;
+                            mNewItemStatuses[pos] = ((x - 1) << FLAG_OFFSET) | changeFlag;
+                            return true;
+                        }
+                    }
+                }
+                curX = snake.x;
+                curY = snake.y;
+            }
+            return false;
+        }
+
+        /**
+         * Dispatches update operations to the given Callback.
+         * <p>
+         * These updates are atomic such that the first update call effects every update call that
+         * comes after it (the same as RecyclerView).
+         *
+         * @param updateCallback The callback to receive the update operations.
+         */
+        public void dispatchUpdatesTo(ListUpdateCallback updateCallback) {
+            final BatchingListUpdateCallback batchingCallback;
+            if (updateCallback instanceof BatchingListUpdateCallback) {
+                batchingCallback = (BatchingListUpdateCallback) updateCallback;
+            } else {
+                batchingCallback = new BatchingListUpdateCallback(updateCallback);
+                // replace updateCallback with a batching callback and override references to
+                // updateCallback so that we don't call it directly by mistake
+                //noinspection UnusedAssignment
+                updateCallback = batchingCallback;
+            }
+            // These are add/remove ops that are converted to moves. We track their positions until
+            // their respective update operations are processed.
+            final List<PostponedUpdate> postponedUpdates = new ArrayList<>();
+            int posOld = mOldListSize;
+            int posNew = mNewListSize;
+            for (int snakeIndex = mSnakes.size() - 1; snakeIndex >= 0; snakeIndex--) {
+                final Snake snake = mSnakes.get(snakeIndex);
+                final int snakeSize = snake.size;
+                final int endX = snake.x + snakeSize;
+                final int endY = snake.y + snakeSize;
+                if (endX < posOld) {
+                    dispatchRemovals(postponedUpdates, batchingCallback, endX, posOld - endX, endX);
+                }
+
+                if (endY < posNew) {
+                    dispatchAdditions(postponedUpdates, batchingCallback, endX, posNew - endY,
+                            endY);
+                }
+                for (int i = snakeSize - 1; i >= 0; i--) {
+                    if ((mOldItemStatuses[snake.x + i] & FLAG_MASK) == FLAG_CHANGED) {
+                        batchingCallback.onChanged(snake.x + i, 1,
+                                mCallback.getChangePayload(snake.x + i, snake.y + i));
+                    }
+                }
+                posOld = snake.x;
+                posNew = snake.y;
+            }
+            batchingCallback.dispatchLastEvent();
+        }
+
+        private static PostponedUpdate removePostponedUpdate(List<PostponedUpdate> updates,
+                                                             int pos, boolean removal) {
+            for (int i = updates.size() - 1; i >= 0; i--) {
+                final PostponedUpdate update = updates.get(i);
+                if (update.posInOwnerList == pos && update.removal == removal) {
+                    updates.remove(i);
+                    for (int j = i; j < updates.size(); j++) {
+                        // offset other ops since they swapped positions
+                        updates.get(j).currentPos += removal ? 1 : -1;
+                    }
+                    return update;
+                }
+            }
+            return null;
+        }
+
+        private void dispatchAdditions(List<PostponedUpdate> postponedUpdates,
+                                       ListUpdateCallback updateCallback, int start, int count, int globalIndex) {
+            if (!mDetectMoves) {
+                updateCallback.onInserted(start, count);
+                return;
+            }
+            for (int i = count - 1; i >= 0; i--) {
+                int status = mNewItemStatuses[globalIndex + i] & FLAG_MASK;
+                switch (status) {
+                    case 0: // real addition
+                        updateCallback.onInserted(start, 1);
+                        for (PostponedUpdate update : postponedUpdates) {
+                            update.currentPos += 1;
+                        }
+                        break;
+                    case FLAG_MOVED_CHANGED:
+                    case FLAG_MOVED_NOT_CHANGED:
+                        final int pos = mNewItemStatuses[globalIndex + i] >> FLAG_OFFSET;
+                        final PostponedUpdate update = removePostponedUpdate(postponedUpdates, pos,
+                                true);
+                        // the item was moved from that position
+                        //noinspection ConstantConditions
+                        updateCallback.onMoved(update.currentPos, start);
+                        if (status == FLAG_MOVED_CHANGED) {
+                            // also dispatch a change
+                            updateCallback.onChanged(start, 1,
+                                    mCallback.getChangePayload(pos, globalIndex + i));
+                        }
+                        break;
+                    case FLAG_IGNORE: // ignoring this
+                        postponedUpdates.add(new PostponedUpdate(globalIndex + i, start, false));
+                        break;
+                    default:
+                        throw new IllegalStateException(
+                                "unknown flag for pos " + (globalIndex + i) + " " + Long
+                                        .toBinaryString(status));
+                }
+            }
+        }
+
+        private void dispatchRemovals(List<PostponedUpdate> postponedUpdates,
+                                      ListUpdateCallback updateCallback, int start, int count, int globalIndex) {
+            if (!mDetectMoves) {
+                updateCallback.onRemoved(start, count);
+                return;
+            }
+            for (int i = count - 1; i >= 0; i--) {
+                final int status = mOldItemStatuses[globalIndex + i] & FLAG_MASK;
+                switch (status) {
+                    case 0: // real removal
+                        updateCallback.onRemoved(start + i, 1);
+                        for (PostponedUpdate update : postponedUpdates) {
+                            update.currentPos -= 1;
+                        }
+                        break;
+                    case FLAG_MOVED_CHANGED:
+                    case FLAG_MOVED_NOT_CHANGED:
+                        final int pos = mOldItemStatuses[globalIndex + i] >> FLAG_OFFSET;
+                        final PostponedUpdate update = removePostponedUpdate(postponedUpdates, pos,
+                                false);
+                        // the item was moved to that position. we do -1 because this is a move not
+                        // add and removing current item offsets the target move by 1
+                        //noinspection ConstantConditions
+                        updateCallback.onMoved(start + i, update.currentPos - 1);
+                        if (status == FLAG_MOVED_CHANGED) {
+                            // also dispatch a change
+                            updateCallback.onChanged(update.currentPos - 1, 1,
+                                    mCallback.getChangePayload(globalIndex + i, pos));
+                        }
+                        break;
+                    case FLAG_IGNORE: // ignoring this
+                        postponedUpdates.add(new PostponedUpdate(globalIndex + i, start + i, true));
+                        break;
+                    default:
+                        throw new IllegalStateException(
+                                "unknown flag for pos " + (globalIndex + i) + " " + Long
+                                        .toBinaryString(status));
+                }
+            }
+        }
+    }
+
+    /**
+     * Represents an update that we skipped because it was a move.
+     * <p>
+     * When an update is skipped, it is tracked as other updates are dispatched until the matching
+     * add/remove operation is found at which point the tracked position is used to dispatch the
+     * update.
+     */
+    private static class PostponedUpdate {
+
+        int posInOwnerList;
+
+        int currentPos;
+
+        boolean removal;
+
+        public PostponedUpdate(int posInOwnerList, int currentPos, boolean removal) {
+            this.posInOwnerList = posInOwnerList;
+            this.currentPos = currentPos;
+            this.removal = removal;
+        }
+
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/android/support/v7/util/ListUpdateCallback.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2016 The Android Open Source Project
+ *
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ru.touchin.roboswag.core.android.support.v7.util;
+
+/**
+ * An interface that can receive Update operations that are applied to a list.
+ *
+ * <p>This class can be used together with DiffUtil to detect changes between two lists.
+ */
+@SuppressWarnings({"PMD", "checkstyle:all"})
+public interface ListUpdateCallback {
+
+    /**
+     * Called when {@code count} number of items are inserted at the given position.
+     *
+     * @param position The position of the new item.
+     * @param count    The number of items that have been added.
+     */
+    void onInserted(int position, int count);
+
+    /**
+     * Called when {@code count} number of items are removed from the given position.
+     *
+     * @param position The position of the item which has been removed.
+     * @param count    The number of items which have been removed.
+     */
+    void onRemoved(int position, int count);
+
+    /**
+     * Called when an item changes its position in the list.
+     *
+     * @param fromPosition The previous position of the item before the move.
+     * @param toPosition   The new position of the item.
+     */
+    void onMoved(int fromPosition, int toPosition);
+
+    /**
+     * Called when {@code count} number of items are updated at the given position.
+     *
+     * @param position The position of the item which has been updated.
+     * @param count    The number of items which has changed.
+     */
+    void onChanged(int position, int count, Object payload);
+
+}

src/main/java/ru/touchin/roboswag/core/observables/BaseChangeable.java

@@ -0,0 +1,112 @@
+/*
+ *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+import ru.touchin.roboswag.core.utils.ObjectUtils;
+import ru.touchin.roboswag.core.utils.Optional;
+import rx.Observable;
+import rx.subjects.BehaviorSubject;
+
+/**
+ * Created by Gavriil Sitnikov on 24/03/2016.
+ * Wrapper over {@link BehaviorSubject} which could be serialized.
+ * Such object is useful as view model and also as value in Android that could be passed into {@link android.os.Bundle}.
+ *
+ * @param <TValue>       Type of Changeable value;
+ * @param <TReturnValue> Type of actual value operating by Changeable. Could be same as {@link TValue}.
+ */
+public abstract class BaseChangeable<TValue, TReturnValue> implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private transient BehaviorSubject<Optional<TValue>> valueSubject;
+
+    public BaseChangeable(@Nullable final TValue defaultValue) {
+        valueSubject = BehaviorSubject.create(new Optional<>(defaultValue));
+    }
+
+    @NonNull
+    protected Observable<Optional<TValue>> observeOptionalValue() {
+        return valueSubject.distinctUntilChanged();
+    }
+
+    /**
+     * Sets current value.
+     *
+     * @param value Value to set.
+     */
+    public void set(@Nullable final TValue value) {
+        valueSubject.onNext(new Optional<>(value));
+    }
+
+    /**
+     * Returns current value.
+     *
+     * @return Current value.
+     */
+    @Nullable
+    public TValue get() {
+        return valueSubject.getValue().get();
+    }
+
+    /**
+     * Returns {@link Observable} which is emits current value and then emitting changes of current value.
+     *
+     * @return Current value {@link Observable}.
+     */
+    @NonNull
+    public abstract Observable<TReturnValue> observe();
+
+    private void writeObject(@NonNull final ObjectOutputStream outputStream) throws IOException {
+        outputStream.writeObject(valueSubject.getValue());
+    }
+
+    @SuppressWarnings("unchecked")
+    private void readObject(@NonNull final ObjectInputStream inputStream) throws IOException, ClassNotFoundException {
+        valueSubject = BehaviorSubject.create((Optional<TValue>) inputStream.readObject());
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+
+        final BaseChangeable<?, ?> that = (BaseChangeable<?, ?>) object;
+        return ObjectUtils.equals(valueSubject.getValue(), that.valueSubject.getValue());
+    }
+
+    @Override
+    public int hashCode() {
+        return valueSubject.getValue() != null ? valueSubject.getValue().hashCode() : 0;
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/Changeable.java

@@ -22,47 +22,19 @@ package ru.touchin.roboswag.core.observables;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
-
-import ru.touchin.roboswag.core.utils.ObjectUtils;
+import ru.touchin.roboswag.core.utils.Optional;
 import rx.Observable;
-import rx.subjects.BehaviorSubject;
 
 /**
  * Created by Gavriil Sitnikov on 24/03/2016.
- * Wrapper over {@link BehaviorSubject} which could be serialized.
- * Such object is useful as view model and also as value in Android that could be passed into {@link android.os.Bundle}.
+ * Variant of {@link BaseChangeable} which is allows to set nullable values.
+ * Needed to separate non-null Changeable from nullable Changeable.
  */
-public class Changeable<T> implements Serializable {
-
-    private static final long serialVersionUID = 1L;
-
-    private transient BehaviorSubject<T> subject;
+//COMPATIBILITY NOTE: in RxJava2 it should extends BaseChangeable<T, Optional<T>>
+public class Changeable<T> extends BaseChangeable<T, T> {
 
     public Changeable(@Nullable final T defaultValue) {
-        subject = BehaviorSubject.create(defaultValue);
-    }
-
-    /**
-     * Sets current value.
-     *
-     * @param value Value to set.
-     */
-    public void set(@Nullable final T value) {
-        subject.onNext(value);
-    }
-
-    /**
-     * Returns current value.
-     *
-     * @return Current value.
-     */
-    @Nullable
-    public T get() {
-        return subject.getValue();
+        super(defaultValue);
     }
 
     /**
@@ -71,35 +43,10 @@ public class Changeable<T> implements Serializable {
      * @return Current value {@link Observable}.
      */
     @NonNull
+    @Override
+    //COMPATIBILITY NOTE: in RxJava2 it should be Observable<Optional<T>>
     public Observable<T> observe() {
-        return subject.distinctUntilChanged();
-    }
-
-    private void writeObject(@NonNull final ObjectOutputStream outputStream) throws IOException {
-        outputStream.writeObject(subject.getValue());
-    }
-
-    @SuppressWarnings("unchecked")
-    private void readObject(@NonNull final ObjectInputStream inputStream) throws IOException, ClassNotFoundException {
-        subject = BehaviorSubject.create((T) inputStream.readObject());
-    }
-
-    @Override
-    public boolean equals(@Nullable final Object object) {
-        if (this == object) {
-            return true;
-        }
-        if (object == null || getClass() != object.getClass()) {
-            return false;
-        }
-
-        final Changeable<?> that = (Changeable<?>) object;
-        return ObjectUtils.equals(subject.getValue(), that.subject.getValue());
-    }
-
-    @Override
-    public int hashCode() {
-        return subject.getValue() != null ? subject.getValue().hashCode() : 0;
+        return observeOptionalValue().map(Optional::get);
     }
 
 }

src/main/java/ru/touchin/roboswag/core/observables/NonNullChangeable.java

@@ -21,18 +21,22 @@ package ru.touchin.roboswag.core.observables;
 
 import android.support.annotation.NonNull;
 
+import ru.touchin.roboswag.core.log.Lc;
 import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
+import rx.Observable;
 
 /**
  * Created by Gavriil Sitnikov on 24/03/2016.
- * Variant of {@link Changeable} which is allows to set only non-null values.
+ * Variant of {@link BaseChangeable} which is allows to set only non-null values.
+ * Needed to separate non-null Changeable from nullable Changeable.
  */
-public class NonNullChangeable<T> extends Changeable<T> {
-
-    private static final long serialVersionUID = 1L;
+public class NonNullChangeable<T> extends BaseChangeable<T, T> {
 
     public NonNullChangeable(@NonNull final T defaultValue) {
         super(defaultValue);
+        if (defaultValue == null) {
+            throw new ShouldNotHappenException();
+        }
     }
 
     @NonNull
@@ -45,11 +49,30 @@ public class NonNullChangeable<T> extends Changeable<T> {
         return value;
     }
 
-    @SuppressWarnings("PMD.UselessOverridingMethod")
-    // UselessOverridingMethod: we need only annotation change
     @Override
     public void set(@NonNull final T value) {
+        if (value == null) {
+            Lc.assertion("value is null");
+            return;
+        }
         super.set(value);
     }
 
+    /**
+     * Returns {@link Observable} which is emits current value and then emitting changes of current value.
+     *
+     * @return Current value {@link Observable}.
+     */
+    @NonNull
+    @Override
+    public Observable<T> observe() {
+        return observeOptionalValue()
+                .map(optional -> {
+                    if (optional.get() == null) {
+                        throw new ShouldNotHappenException();
+                    }
+                    return optional.get();
+                });
+    }
+
 }

src/main/java/ru/touchin/roboswag/core/observables/ObservableResult.java

@@ -1,59 +0,0 @@
-package ru.touchin.roboswag.core.observables;
-
-import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Created by Gavriil Sitnikov on 21/05/2016.
- * Object represents Observable's execution result. Contains all items and errors emitted by Observable during subscription.
- */
-public class ObservableResult<T> {
-
-    @NonNull
-    private final List<T> items = new LinkedList<>();
-    @Nullable
-    private Throwable error;
-
-    /**
-     * Passes item to collect.
-     *
-     * @param item Emitted item.
-     */
-    public void onNext(@Nullable final T item) {
-        items.add(item);
-    }
-
-    /**
-     * Passes error to collect.
-     *
-     * @param error Emitted error.
-     */
-    public void onError(@NonNull final Throwable error) {
-        this.error = error;
-    }
-
-    /**
-     * Returns list of collected items.
-     *
-     * @return Items.
-     */
-    @NonNull
-    public List<T> getItems() {
-        return new ArrayList<>(items);
-    }
-
-    /**
-     * Returns collected error.
-     *
-     * @return Error.
-     */
-    @Nullable
-    public Throwable getError() {
-        return error;
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/OnSubscribeRefCountWithCacheTime.java

@@ -28,7 +28,6 @@ import rx.Observable.OnSubscribe;
 import rx.Scheduler;
 import rx.Subscriber;
 import rx.Subscription;
-import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.observables.ConnectableObservable;
 import rx.schedulers.Schedulers;
@@ -114,19 +113,15 @@ public final class OnSubscribeRefCountWithCacheTime<T> implements OnSubscribe<T>
     @NonNull
     private Action1<Subscription> onSubscribe(@NonNull final Subscriber<? super T> subscriber,
                                               @NonNull final AtomicBoolean writeLocked) {
-        return new Action1<Subscription>() {
-            @Override
-            public void call(@NonNull final Subscription subscription) {
-
-                try {
-                    baseSubscription.add(subscription);
-                    // ready to subscribe to source so do it
-                    doSubscribe(subscriber, baseSubscription);
-                } finally {
-                    // release the write lock
-                    lock.unlock();
-                    writeLocked.set(false);
-                }
+        return subscription -> {
+            try {
+                baseSubscription.add(subscription);
+                // ready to subscribe to source so do it
+                doSubscribe(subscriber, baseSubscription);
+            } finally {
+                // release the write lock
+                lock.unlock();
+                writeLocked.set(false);
             }
         };
     }
@@ -152,14 +147,14 @@ public final class OnSubscribeRefCountWithCacheTime<T> implements OnSubscribe<T>
             }
 
             private void cleanup() {
-                // on error or completion we need to unsubscribe the base subscription
-                // and set the subscriptionCount to 0 
+                // on error or completion we need to unsubscribe the base subscription and set the subscriptionCount to 0
                 lock.lock();
                 try {
                     if (baseSubscription == currentBase) {
-                        if (worker != null) {
-                            worker.unsubscribe();
-                            worker = null;
+                        cleanupWorker();
+                        // backdoor into the ConnectableObservable to cleanup and reset its state
+                        if (source instanceof Subscription) {
+                            ((Subscription) source).unsubscribe();
                         }
                         baseSubscription.unsubscribe();
                         baseSubscription = new CompositeSubscription();
@@ -174,41 +169,45 @@ public final class OnSubscribeRefCountWithCacheTime<T> implements OnSubscribe<T>
 
     @NonNull
     private Subscription disconnect(@NonNull final CompositeSubscription current) {
-        return Subscriptions.create(new Action0() {
-
-            @Override
-            public void call() {
-                lock.lock();
-                try {
-                    if (baseSubscription == current && subscriptionCount.decrementAndGet() == 0) {
-                        if (worker != null) {
-                            worker.unsubscribe();
-                        } else {
-                            worker = scheduler.createWorker();
-                        }
-                        worker.schedule(new Action0() {
-                            @Override
-                            public void call() {
-                                lock.lock();
-                                try {
-                                    if (subscriptionCount.get() == 0) {
-                                        baseSubscription.unsubscribe();
-                                        // need a new baseSubscription because once
-                                        // unsubscribed stays that way
-                                        worker.unsubscribe();
-                                        worker = null;
-                                        baseSubscription = new CompositeSubscription();
-                                    }
-                                } finally {
-                                    lock.unlock();
-                                }
-                            }
-                        }, cacheTime, cacheTimeUnit);
+        return Subscriptions.create(() -> {
+            lock.lock();
+            try {
+                if (baseSubscription == current && subscriptionCount.decrementAndGet() == 0) {
+                    if (worker != null) {
+                        worker.unsubscribe();
+                    } else {
+                        worker = scheduler.createWorker();
                     }
-                } finally {
-                    lock.unlock();
+                    worker.schedule(() -> {
+                        lock.lock();
+                        try {
+                            if (subscriptionCount.get() == 0) {
+                                cleanupWorker();
+                                // backdoor into the ConnectableObservable to cleanup and reset its state
+                                if (source instanceof Subscription) {
+                                    ((Subscription) source).unsubscribe();
+                                }
+                                baseSubscription.unsubscribe();
+                                // need a new baseSubscription because once
+                                // unsubscribed stays that way
+                                baseSubscription = new CompositeSubscription();
+                            }
+                        } finally {
+                            lock.unlock();
+                        }
+                    }, cacheTime, cacheTimeUnit);
                 }
+            } finally {
+                lock.unlock();
             }
         });
     }
+
+    private void cleanupWorker() {
+        if (worker != null) {
+            worker.unsubscribe();
+            worker = null;
+        }
+    }
+
 }

src/main/java/ru/touchin/roboswag/core/observables/RxAndroidUtils.java

@@ -20,9 +20,11 @@
 package ru.touchin.roboswag.core.observables;
 
 import android.app.Service;
+import android.content.BroadcastReceiver;
 import android.content.ComponentName;
 import android.content.Context;
 import android.content.Intent;
+import android.content.IntentFilter;
 import android.content.ServiceConnection;
 import android.os.IBinder;
 import android.os.Looper;
@@ -33,9 +35,9 @@ import java.util.concurrent.CountDownLatch;
 
 import ru.touchin.roboswag.core.log.Lc;
 import ru.touchin.roboswag.core.utils.ServiceBinder;
+import rx.Emitter;
 import rx.Observable;
 import rx.Scheduler;
-import rx.Subscriber;
 import rx.android.schedulers.AndroidSchedulers;
 
 /**
@@ -55,23 +57,50 @@ public final class RxAndroidUtils {
     @NonNull
     public static <T extends Service> Observable<T> observeService(@NonNull final Context context, @NonNull final Class<T> serviceClass) {
         return Observable
-                .just(new SubscribeServiceConnection<T>())
-                .switchMap(serviceConnection -> Observable
-                        .<T>create(subscriber -> {
-                            serviceConnection.subscriber = subscriber;
-                            context.bindService(new Intent(context, serviceClass), serviceConnection, Context.BIND_AUTO_CREATE);
-                        })
-                        .doOnUnsubscribe(() -> context.unbindService(serviceConnection)))
+                .just(new OnSubscribeServiceConnection<T>())
+                .switchMap(onSubscribeServiceConnection -> Observable
+                        .<T>create(emitter -> {
+                            onSubscribeServiceConnection.emitter = emitter;
+                            context.bindService(new Intent(context, serviceClass), onSubscribeServiceConnection, Context.BIND_AUTO_CREATE);
+                        }, Emitter.BackpressureMode.LATEST)
+                        .doOnUnsubscribe(() -> {
+                            context.unbindService(onSubscribeServiceConnection);
+                            onSubscribeServiceConnection.emitter = null;
+                        }))
                 .distinctUntilChanged()
                 .replay(1)
                 .refCount();
     }
 
+    /**
+     * Observes classic Android broadcast with {@link BroadcastReceiver} as source of Observable items and Intent as items.
+     *
+     * @param context      Context to register {@link BroadcastReceiver};
+     * @param intentFilter {@link IntentFilter} to register {@link BroadcastReceiver};
+     * @return Observable that observes Android broadcasts.
+     */
+    @NonNull
+    public static Observable<Intent> observeBroadcastEvent(@NonNull final Context context, @NonNull final IntentFilter intentFilter) {
+        return Observable
+                .just(new OnSubscribeBroadcastReceiver())
+                .switchMap(onOnSubscribeBroadcastReceiver -> Observable
+                        .<Intent>create(emitter -> {
+                            onOnSubscribeBroadcastReceiver.emitter = emitter;
+                            context.registerReceiver(onOnSubscribeBroadcastReceiver, intentFilter);
+                        }, Emitter.BackpressureMode.LATEST)
+                        .doOnUnsubscribe(() -> {
+                            context.unregisterReceiver(onOnSubscribeBroadcastReceiver);
+                            onOnSubscribeBroadcastReceiver.emitter = null;
+                        }))
+                .share();
+    }
+
     /**
      * Creating {@link Scheduler} that is scheduling work on specific thread with {@link Looper}.
      * Do not use it much times - it is creating endless thread every call.
      * It's good to use it only like a constant like:
      * private static final Scheduler SCHEDULER = RxAndroidUtils.createLooperScheduler();
+     * IMPORTANT NOTE: looper thread will live forever! Do not create a lot of such Schedulers.
      *
      * @return Looper thread based {@link Scheduler}.
      */
@@ -91,20 +120,19 @@ public final class RxAndroidUtils {
     private RxAndroidUtils() {
     }
 
-    private static class SubscribeServiceConnection<T> implements ServiceConnection {
-
+    private static class OnSubscribeServiceConnection<TService extends Service> implements ServiceConnection {
         @Nullable
-        private Subscriber<? super T> subscriber;
+        private Emitter<? super TService> emitter;
 
         @SuppressWarnings("unchecked")
         @Override
         public void onServiceConnected(@NonNull final ComponentName name, @Nullable final IBinder service) {
-            if (subscriber == null) {
+            if (emitter == null) {
                 return;
             }
 
             if (service instanceof ServiceBinder) {
-                subscriber.onNext((T) ((ServiceBinder) service).getService());
+                emitter.onNext((TService) ((ServiceBinder) service).getService());
             } else {
                 Lc.assertion("IBinder should be instance of ServiceBinder.");
             }
@@ -112,8 +140,21 @@ public final class RxAndroidUtils {
 
         @Override
         public void onServiceDisconnected(@NonNull final ComponentName name) {
-            if (subscriber != null) {
-                subscriber.onNext(null);
+            // service have been killed/crashed and destroyed. instead of emit null just wait service reconnection.
+            // even if someone keeps reference to dead service it is problem of service object to work correctly after destroy.
+        }
+
+    }
+
+    private static class OnSubscribeBroadcastReceiver extends BroadcastReceiver {
+
+        @Nullable
+        private Emitter<? super Intent> emitter;
+
+        @Override
+        public void onReceive(@NonNull final Context context, @NonNull final Intent intent) {
+            if (emitter != null) {
+                emitter.onNext(intent);
             }
         }
 

src/main/java/ru/touchin/roboswag/core/observables/SequenceObservableExecutor.java

@@ -1,78 +0,0 @@
-package ru.touchin.roboswag.core.observables;
-
-
-import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
-
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-
-import ru.touchin.roboswag.core.utils.ProcessPriorityThreadFactory;
-import rx.Observable;
-import rx.Scheduler;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.functions.Actions;
-import rx.schedulers.Schedulers;
-
-//TODO: errors/next/completion/single/observable/block next after first fail
-public class SequenceObservableExecutor {
-
-    private static void safeUnsubscribe(@Nullable final Subscription subscription) {
-        if (subscription != null && !subscription.isUnsubscribed()) {
-            subscription.unsubscribe();
-        }
-    }
-
-    @NonNull
-    private final Scheduler sendingScheduler = Schedulers.from(Executors.newSingleThreadExecutor());
-    @NonNull
-    private final Scheduler executeScheduler = Schedulers.from(Executors.newSingleThreadExecutor(
-            new ProcessPriorityThreadFactory(Thread.MIN_PRIORITY)));
-
-    @NonNull
-    public Observable<?> execute(@NonNull final Observable<?> completable) {
-        final Task task = new Task(completable);
-        return Observable
-                .create(task)
-                .doOnUnsubscribe(task::cancel);
-    }
-
-    private class Task implements Observable.OnSubscribe<Subscriber> {
-
-        @NonNull
-        private final Observable<?> completable;
-        @Nullable
-        private Subscription scheduleSubscription;
-        @Nullable
-        private Subscription executeSubscription;
-
-        public Task(@NonNull final Observable<?> completable) {
-            this.completable = completable;
-        }
-
-        @Override
-        public void call(@NonNull final Subscriber subscriber) {
-            scheduleSubscription = sendingScheduler.createWorker().schedule(() -> {
-                final CountDownLatch blocker = new CountDownLatch(1);
-                executeSubscription = completable
-                        .subscribeOn(executeScheduler)
-                        .doOnUnsubscribe(blocker::countDown)
-                        .subscribe(Actions.empty(), subscriber::onError, subscriber::onCompleted);
-                try {
-                    blocker.await();
-                } catch (final InterruptedException exception) {
-                    safeUnsubscribe(executeSubscription);
-                    subscriber.onError(exception);
-                }
-            });
-        }
-
-        public void cancel() {
-            safeUnsubscribe(scheduleSubscription);
-            safeUnsubscribe(executeSubscription);
-        }
-
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/collections/Change.java

@@ -1,225 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.collections;
-
-import android.support.annotation.NonNull;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-
-/**
- * Created by Gavriil Sitnikov on 23/05/16.
- * Class representing simple change of collection like insertion, remove or replacing/changing items.
- *
- * @param <TItem> Type of changing collection's items.
- */
-public class Change<TItem> {
-
-    /**
-     * Method to calculate changes between two collections.
-     *
-     * @param initialCollection           Initial collection;
-     * @param modifiedCollection          Changed collection;
-     * @param shrinkChangesToModifiedSize Flag to make position of changed items be less then modified collection size.
-     *                                    It is needed sometimes to not get exceptions like {@link ArrayIndexOutOfBoundsException}.
-     * @param <TItem>                     Type of collections items.
-     * @return Changes between collections.
-     */
-    @NonNull
-    public static <TItem> Collection<Change<TItem>> calculateCollectionChanges(@NonNull final Collection<TItem> initialCollection,
-                                                                               @NonNull final Collection<TItem> modifiedCollection,
-                                                                               final boolean shrinkChangesToModifiedSize) {
-        return new CollectionsChangesCalculator<>(initialCollection, modifiedCollection, shrinkChangesToModifiedSize).calculateChanges();
-    }
-
-    @NonNull
-    private final Type type;
-    @NonNull
-    private final Collection<TItem> changedItems;
-    private final int start;
-    private final int count;
-
-    public Change(@NonNull final Type type, @NonNull final Collection<TItem> changedItems, final int start) {
-        this.type = type;
-        this.changedItems = Collections.unmodifiableCollection(new ArrayList<>(changedItems));
-        this.start = start;
-        this.count = changedItems.size();
-    }
-
-    /**
-     * Returns type of change.
-     *
-     * @return Type of change.
-     */
-    @NonNull
-    public Type getType() {
-        return type;
-    }
-
-    /**
-     * Returns collection of items which this change applied to.
-     *
-     * @return Changed items.
-     */
-    @NonNull
-    public Collection<TItem> getChangedItems() {
-        return changedItems;
-    }
-
-    /**
-     * Returns first index of changed item.
-     *
-     * @return Start of change.
-     */
-    public int getStart() {
-        return start;
-    }
-
-    /**
-     * Returns count of changed items.
-     *
-     * @return Count of changed items.
-     */
-    public int getCount() {
-        return count;
-    }
-
-    @NonNull
-    @Override
-    public String toString() {
-        return type + " change of " + start + ":" + count;
-    }
-
-    /**
-     * Type of change.
-     */
-    public enum Type {
-        INSERTED,
-        CHANGED,
-        REMOVED
-    }
-
-    private static class CollectionsChangesCalculator<TItem> {
-
-        @NonNull
-        private final Collection<TItem> initialCollection;
-        @NonNull
-        private final Collection<TItem> modifiedCollection;
-        private final boolean shrinkChangesToModifiedSize;
-        private int initialOffset;
-        @NonNull
-        private final Collection<TItem> itemsToAdd = new ArrayList<>();
-        private int currentSize;
-        private int oldSize;
-        private int newSize;
-        private int couldBeAdded;
-
-        public CollectionsChangesCalculator(@NonNull final Collection<TItem> initialCollection,
-                                            @NonNull final Collection<TItem> modifiedCollection,
-                                            final boolean shrinkChangesToModifiedSize) {
-            this.initialCollection = initialCollection;
-            this.modifiedCollection = modifiedCollection;
-            this.shrinkChangesToModifiedSize = shrinkChangesToModifiedSize;
-        }
-
-        @NonNull
-        public Collection<Change<TItem>> calculateChanges() {
-            initialOffset = 0;
-            itemsToAdd.clear();
-            currentSize = 0;
-            oldSize = initialCollection.size();
-            newSize = modifiedCollection.size();
-            couldBeAdded = modifiedCollection.size() - initialCollection.size();
-            final List<Change<TItem>> result = new ArrayList<>();
-            for (final TItem modifiedItem : modifiedCollection) {
-                int foundPosition = 0;
-                for (final Object initialObject : initialCollection) {
-                    if (foundPosition >= initialOffset && modifiedItem.equals(initialObject)) {
-                        if (tryAddSkipped(result) == MethodAction.RETURN
-                                || tryRemoveRest(result, initialOffset, foundPosition - initialOffset) == MethodAction.RETURN) {
-                            return result;
-                        }
-                        initialOffset = foundPosition + 1;
-                        currentSize++;
-                        break;
-                    }
-                    foundPosition++;
-                }
-                // if not found
-                if (foundPosition >= initialCollection.size()) {
-                    itemsToAdd.add(modifiedItem);
-                }
-            }
-
-            if (tryAddSkipped(result) == MethodAction.RETURN) {
-                return result;
-            }
-            tryRemoveRest(result, initialOffset, initialCollection.size() - initialOffset);
-            return result;
-        }
-
-        @NonNull
-        private MethodAction tryAddSkipped(@NonNull final Collection<Change<TItem>> changes) {
-            if (!itemsToAdd.isEmpty()) {
-                if (shrinkChangesToModifiedSize && couldBeAdded < itemsToAdd.size()) {
-                    addSimpleDifferenceChanges(changes);
-                    return MethodAction.RETURN;
-                }
-                changes.add(new Change<>(Type.INSERTED, itemsToAdd, currentSize));
-                currentSize += itemsToAdd.size();
-                couldBeAdded -= itemsToAdd.size();
-                itemsToAdd.clear();
-            }
-            return MethodAction.CONTINUE;
-        }
-
-        @NonNull
-        private MethodAction tryRemoveRest(@NonNull final Collection<Change<TItem>> changes,
-                                           final int initialOffset,
-                                           final int itemsToRemove) {
-            if (itemsToRemove > 0) {
-                if (shrinkChangesToModifiedSize && couldBeAdded < -itemsToRemove) {
-                    addSimpleDifferenceChanges(changes);
-                    return MethodAction.RETURN;
-                }
-                changes.add(new Change<>(Change.Type.REMOVED,
-                        new ArrayList<>(initialCollection).subList(initialOffset, initialOffset + itemsToRemove),
-                        currentSize));
-            }
-            return MethodAction.CONTINUE;
-        }
-
-        private void addSimpleDifferenceChanges(@NonNull final Collection<Change<TItem>> changes) {
-            changes.add(new Change<>(Type.CHANGED, new ArrayList<>(modifiedCollection).subList(currentSize, newSize), currentSize));
-            if (oldSize - newSize > 0) {
-                changes.add(new Change<>(Type.REMOVED, new ArrayList<>(initialCollection).subList(newSize, oldSize), newSize));
-            }
-        }
-
-        private enum MethodAction {
-            RETURN,
-            CONTINUE
-        }
-
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/collections/ObservableCollection.java

@@ -25,33 +25,32 @@ import android.support.annotation.Nullable;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.io.Serializable;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import ru.touchin.roboswag.core.observables.collections.changes.Change;
+import ru.touchin.roboswag.core.observables.collections.changes.CollectionChanges;
+import rx.Emitter;
 import rx.Observable;
-import rx.Subscriber;
 
 /**
  * Created by Gavriil Sitnikov on 23/05/16.
  * Class to represent collection which is providing it's inner changes in Rx observable way.
  * Use {@link #observeChanges()} and {@link #observeItems()} to observe collection changes.
- * Use {@link #loadItem(int)} to load item asynchronously.
  * Methods {@link #size()} and {@link #get(int)} will return only already loaded items info.
  *
  * @param <TItem> Type of collection's items.
  */
-public abstract class ObservableCollection<TItem> implements Serializable {
+public abstract class ObservableCollection<TItem> {
 
     private int changesCount;
     @NonNull
-    private transient Observable<CollectionChange<TItem>> changesObservable;
+    private transient Observable<CollectionChanges<TItem>> changesObservable;
     @NonNull
     private transient Observable<Collection<TItem>> itemsObservable;
     @Nullable
-    private transient Subscriber<? super CollectionChange<TItem>> changesSubscriber;
+    private transient Emitter<? super CollectionChanges<TItem>> changesEmitter;
 
     public ObservableCollection() {
         this.changesObservable = createChangesObservable();
@@ -59,21 +58,18 @@ public abstract class ObservableCollection<TItem> implements Serializable {
     }
 
     @NonNull
-    private Observable<CollectionChange<TItem>> createChangesObservable() {
+    private Observable<CollectionChanges<TItem>> createChangesObservable() {
         return Observable
-                .<CollectionChange<TItem>>create(subscriber -> this.changesSubscriber = subscriber)
-                .doOnUnsubscribe(() -> this.changesSubscriber = null)
-                .replay(0)
-                .refCount();
+                .<CollectionChanges<TItem>>create(emitter -> this.changesEmitter = emitter, Emitter.BackpressureMode.BUFFER)
+                .doOnUnsubscribe(() -> this.changesEmitter = null)
+                .share();
     }
 
     @NonNull
     private Observable<Collection<TItem>> createItemsObservable() {
         return Observable
-                .<Collection<TItem>>switchOnNext(Observable.create(subscriber -> {
-                    subscriber.onNext(observeChanges().map(changes -> getItems()).startWith(getItems()));
-                    subscriber.onCompleted();
-                }))
+                //switchOnNext to calculate getItems() on subscription but not on that method calling moment
+                .switchOnNext(Observable.fromCallable(() -> observeChanges().map(changes -> getItems()).startWith(getItems())))
                 .replay(1)
                 .refCount();
     }
@@ -92,19 +88,28 @@ public abstract class ObservableCollection<TItem> implements Serializable {
      *
      * @param change Change of collection.
      */
-    protected void notifyAboutChange(@NonNull final Change<TItem> change) {
-        notifyAboutChanges(Collections.singleton(change));
+    protected void notifyAboutChange(@NonNull final List<TItem> insertedItems,
+                                     @NonNull final List<TItem> removedItems,
+                                     @NonNull final Change change) {
+        notifyAboutChanges(insertedItems, removedItems, Collections.singleton(change));
     }
 
     /**
      * Method to notify that collection have changed.
      *
-     * @param changes Changes of collection.
+     * @param insertedItems Collection of inserted items;
+     * @param removedItems  Collection of removed items;
+     * @param changes       Changes of collection.
      */
-    protected void notifyAboutChanges(@NonNull final Collection<Change<TItem>> changes) {
+    protected void notifyAboutChanges(@NonNull final List<TItem> insertedItems,
+                                      @NonNull final List<TItem> removedItems,
+                                      @NonNull final Collection<Change> changes) {
+        if (changes.isEmpty()) {
+            return;
+        }
         changesCount++;
-        if (changesSubscriber != null) {
-            changesSubscriber.onNext(new CollectionChange<>(changesCount, Collections.unmodifiableCollection(changes)));
+        if (changesEmitter != null) {
+            changesEmitter.onNext(new CollectionChanges<>(changesCount, insertedItems, removedItems, changes));
         }
     }
 
@@ -114,7 +119,7 @@ public abstract class ObservableCollection<TItem> implements Serializable {
      * @return List of changes applied to collection.
      */
     @NonNull
-    public Observable<CollectionChange<TItem>> observeChanges() {
+    public Observable<CollectionChanges<TItem>> observeChanges() {
         return changesObservable;
     }
 
@@ -164,36 +169,6 @@ public abstract class ObservableCollection<TItem> implements Serializable {
         return size() == 0;
     }
 
-    /**
-     * Returns {@link Observable} which is loading item by position.
-     * It could return null in onNext callback if there is no item to load for such position.
-     *
-     * @param position Position to load item;
-     * @return {@link Observable} to load item.
-     */
-    @NonNull
-    public abstract Observable<TItem> loadItem(int position);
-
-    /**
-     * Returns {@link Observable} which is loading item by range.
-     * It will return collection of loaded items in onNext callback.
-     *
-     * @param first First position of item to load;
-     * @param last  Last position of item to load;
-     * @return {@link Observable} to load items.
-     */
-    @NonNull
-    public Observable<Collection<TItem>> loadRange(final int first, final int last) {
-        final List<Observable<TItem>> itemsRequests = new ArrayList<>();
-        for (int i = first; i <= last; i++) {
-            itemsRequests.add(loadItem(i));
-        }
-        return Observable.concatEager(itemsRequests)
-                .filter(loadedItem -> loadedItem != null)
-                .toList()
-                .map(Collections::unmodifiableCollection);
-    }
-
     private void writeObject(@NonNull final ObjectOutputStream outputStream) throws IOException {
         outputStream.writeInt(changesCount);
     }
@@ -204,41 +179,4 @@ public abstract class ObservableCollection<TItem> implements Serializable {
         this.itemsObservable = createItemsObservable();
     }
 
-    /**
-     * Class which is representing change of collection. There could be multiple changes applied to collection.
-     *
-     * @param <TItem> Type of collection's items.
-     */
-    public static class CollectionChange<TItem> {
-
-        private final int number;
-        @NonNull
-        private final Collection<Change<TItem>> changes;
-
-        protected CollectionChange(final int number, @NonNull final Collection<Change<TItem>> changes) {
-            this.number = number;
-            this.changes = changes;
-        }
-
-        /**
-         * Returns number of change.
-         *
-         * @return Number of change.
-         */
-        public int getNumber() {
-            return number;
-        }
-
-        /**
-         * Returns collection of changes.
-         *
-         * @return Collection of changes.
-         */
-        @NonNull
-        public Collection<Change<TItem>> getChanges() {
-            return changes;
-        }
-
-    }
-
 }

src/main/java/ru/touchin/roboswag/core/observables/collections/ObservableFilteredList.java

@@ -7,10 +7,13 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Executors;
 
-import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
-import rx.Observable;
+import ru.touchin.roboswag.core.observables.collections.changes.DefaultCollectionsChangesCalculator;
+import rx.Scheduler;
+import rx.Subscription;
 import rx.functions.Func1;
+import rx.schedulers.Schedulers;
 
 /**
  * Created by Gavriil Sitnikov on 02/06/2016.
@@ -21,9 +24,15 @@ import rx.functions.Func1;
  */
 public class ObservableFilteredList<TItem> extends ObservableCollection<TItem> {
 
+    // we need to filter on 1 thread to prevent parallel filtering
+    private static final Scheduler FILTER_SCHEDULER = Schedulers.from(Executors.newSingleThreadExecutor());
+
     @NonNull
     private static <TItem> List<TItem> filterCollection(@NonNull final Collection<TItem> sourceCollection,
-                                                        @NonNull final Func1<TItem, Boolean> filter) {
+                                                        @Nullable final Func1<TItem, Boolean> filter) {
+        if (filter == null) {
+            return new ArrayList<>(sourceCollection);
+        }
         final List<TItem> result = new ArrayList<>(sourceCollection.size());
         for (final TItem item : sourceCollection) {
             if (filter.call(item)) {
@@ -33,35 +42,43 @@ public class ObservableFilteredList<TItem> extends ObservableCollection<TItem> {
         return result;
     }
 
-    @Nullable
+    @NonNull
     private List<TItem> filteredList;
-    @Nullable
-    private Collection<TItem> sourceCollection;
+    @NonNull
+    private ObservableCollection<TItem> sourceCollection;
     @Nullable
     private Func1<TItem, Boolean> filter;
+    @Nullable
+    private Subscription sourceCollectionSubscription;
 
     public ObservableFilteredList() {
-        super();
-        //do nothing
-    }
-
-    public ObservableFilteredList(@NonNull final Collection<TItem> sourceCollection) {
-        super();
-        this.sourceCollection = new ArrayList<>(sourceCollection);
-        this.filteredList = new ArrayList<>(sourceCollection);
+        this(new ArrayList<>(), null);
     }
 
     public ObservableFilteredList(@NonNull final Func1<TItem, Boolean> filter) {
-        super();
-        this.filter = filter;
+        this(new ArrayList<>(), filter);
     }
 
-    public ObservableFilteredList(@NonNull final Collection<TItem> sourceCollection,
-                                  @NonNull final Func1<TItem, Boolean> filter) {
+    public ObservableFilteredList(@NonNull final Collection<TItem> sourceCollection, @Nullable final Func1<TItem, Boolean> filter) {
+        this(new ObservableList<>(sourceCollection), filter);
+    }
+
+    public ObservableFilteredList(@NonNull final ObservableCollection<TItem> sourceCollection, @Nullable final Func1<TItem, Boolean> filter) {
         super();
-        this.sourceCollection = new ArrayList<>(sourceCollection);
         this.filter = filter;
-        filteredList = filterCollection(this.sourceCollection, this.filter);
+        this.sourceCollection = sourceCollection;
+        this.filteredList = filterCollection(this.sourceCollection.getItems(), this.filter);
+        updateInternal();
+    }
+
+    /**
+     * Sets collection of items to filter.
+     *
+     * @param sourceCollection Collection with items.
+     */
+    public void setSourceCollection(@Nullable final ObservableCollection<TItem> sourceCollection) {
+        this.sourceCollection = sourceCollection != null ? sourceCollection : new ObservableList<>();
+        updateInternal();
     }
 
     /**
@@ -70,8 +87,8 @@ public class ObservableFilteredList<TItem> extends ObservableCollection<TItem> {
      * @param sourceCollection Collection with items.
      */
     public void setSourceCollection(@Nullable final Collection<TItem> sourceCollection) {
-        this.sourceCollection = sourceCollection != null ? new ArrayList<>(sourceCollection) : null;
-        updateCollections();
+        this.sourceCollection = sourceCollection != null ? new ObservableList<>(sourceCollection) : new ObservableList<>();
+        updateInternal();
     }
 
     /**
@@ -81,73 +98,57 @@ public class ObservableFilteredList<TItem> extends ObservableCollection<TItem> {
      */
     public void setFilter(@Nullable final Func1<TItem, Boolean> filter) {
         this.filter = filter;
-        updateCollections();
+        updateInternal();
+    }
+
+    private void updateInternal() {
+        if (sourceCollectionSubscription != null) {
+            sourceCollectionSubscription.unsubscribe();
+            sourceCollectionSubscription = null;
+        }
+        sourceCollectionSubscription = sourceCollection.observeItems()
+                .observeOn(FILTER_SCHEDULER)
+                .subscribe(items -> {
+                    final List<TItem> oldFilteredList = filteredList;
+                    filteredList = filterCollection(items, filter);
+                    final DefaultCollectionsChangesCalculator<TItem> calculator
+                            = new DefaultCollectionsChangesCalculator<>(oldFilteredList, filteredList, false);
+                    notifyAboutChanges(calculator.calculateInsertedItems(), calculator.calculateRemovedItems(), calculator.calculateChanges());
+                });
     }
 
     /**
      * Updates collection by current filter. Use it if some item's parameter which is important for filtering have changing.
      */
-    public void updateCollections() {
-        if (sourceCollection == null) {
-            if (filteredList != null) {
-                final Change<TItem> change = new Change<>(Change.Type.REMOVED, filteredList, 0);
-                filteredList = null;
-                notifyAboutChange(change);
-            }
-            return;
-        }
-        final List<TItem> oldFilteredList = filteredList;
-        if (filter != null) {
-            filteredList = filterCollection(sourceCollection, filter);
-        } else {
-            filteredList = new ArrayList<>(sourceCollection);
-        }
-        if (oldFilteredList != null) {
-            final Collection<Change<TItem>> changes = Change.calculateCollectionChanges(oldFilteredList, filteredList, false);
-            if (!changes.isEmpty()) {
-                notifyAboutChanges(changes);
-            }
-        } else {
-            notifyAboutChange(new Change<>(Change.Type.INSERTED, filteredList, 0));
-        }
+    public void update() {
+        updateInternal();
     }
 
     @Override
     public int size() {
-        return filteredList != null ? filteredList.size() : 0;
+        return filteredList.size();
     }
 
     @NonNull
     @Override
     public TItem get(final int position) {
-        if (filteredList == null) {
-            throw new ShouldNotHappenException();
-        }
         return filteredList.get(position);
     }
 
     @NonNull
     @Override
     public Collection<TItem> getItems() {
-        return filteredList != null ? Collections.unmodifiableCollection(filteredList) : Collections.emptyList();
+        return Collections.unmodifiableCollection(filteredList);
     }
 
     /**
-     * Returns source non-filtered collection of items.
+     * Returns source non-filtered observable collection of items.
      *
      * @return Non-filtered collection of items.
      */
     @NonNull
-    public Collection<TItem> getSourceItems() {
-        return sourceCollection != null ? Collections.unmodifiableCollection(sourceCollection) : Collections.emptyList();
-    }
-
-    @NonNull
-    @Override
-    public Observable<TItem> loadItem(final int position) {
-        return filteredList != null && filteredList.size() > position
-                ? Observable.just(filteredList.get(position))
-                : Observable.just(null);
+    public ObservableCollection<TItem> getSourceCollection() {
+        return sourceCollection;
     }
 
 }

src/main/java/ru/touchin/roboswag/core/observables/collections/ObservableList.java

@@ -20,6 +20,7 @@
 package ru.touchin.roboswag.core.observables.collections;
 
 import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
 
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -31,7 +32,12 @@ import java.util.Collections;
 import java.util.List;
 
 import ru.touchin.roboswag.core.log.Lc;
-import rx.Observable;
+import ru.touchin.roboswag.core.observables.collections.changes.Change;
+import ru.touchin.roboswag.core.observables.collections.changes.ChangePayloadProducer;
+import ru.touchin.roboswag.core.observables.collections.changes.CollectionsChangesCalculator;
+import ru.touchin.roboswag.core.observables.collections.changes.DefaultCollectionsChangesCalculator;
+import ru.touchin.roboswag.core.observables.collections.changes.DiffCollectionsChangesCalculator;
+import ru.touchin.roboswag.core.observables.collections.changes.SameItemsPredicate;
 
 /**
  * Created by Gavriil Sitnikov on 23/05/16.
@@ -47,6 +53,13 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
 
     @NonNull
     private List<TItem> items;
+    private boolean detectMoves;
+    @Nullable
+    private SameItemsPredicate<TItem> sameItemsPredicate;
+    @Nullable
+    private ChangePayloadProducer<TItem> changePayloadProducer;
+    @Nullable
+    private ObservableList<TItem> diffUtilsSource;
 
     public ObservableList() {
         super();
@@ -76,7 +89,7 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
     public void add(final int position, @NonNull final TItem item) {
         synchronized (this) {
             items.add(position, item);
-            notifyAboutChange(new Change<>(Change.Type.INSERTED, Collections.singleton(item), position));
+            notifyAboutChange(Collections.singletonList(item), Collections.emptyList(), new Change.Inserted(position, 1));
         }
     }
 
@@ -99,7 +112,7 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
         synchronized (this) {
             if (!itemsToAdd.isEmpty()) {
                 items.addAll(position, itemsToAdd);
-                notifyAboutChange(new Change<>(Change.Type.INSERTED, itemsToAdd, position));
+                notifyAboutChange(new ArrayList<>(itemsToAdd), Collections.emptyList(), new Change.Inserted(position, itemsToAdd.size()));
             }
         }
     }
@@ -140,12 +153,11 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
             return;
         }
         synchronized (this) {
-            final List<TItem> changedItems = new ArrayList<>(count);
+            final List<TItem> removedItems = new ArrayList<>(count);
             for (int i = 0; i < count; i++) {
-                changedItems.add(items.get(position));
-                items.remove(position);
+                removedItems.add(items.remove(position));
             }
-            notifyAboutChange(new Change<>(Change.Type.REMOVED, changedItems, position));
+            notifyAboutChange(Collections.emptyList(), removedItems, new Change.Removed(position, count));
         }
     }
 
@@ -154,10 +166,10 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
      */
     public void clear() {
         synchronized (this) {
-            final Change<TItem> change = new Change<>(Change.Type.REMOVED, items, 0);
-            if (!change.getChangedItems().isEmpty()) {
+            if (!items.isEmpty()) {
+                final List<TItem> removedItems = new ArrayList<>(items);
                 items.clear();
-                notifyAboutChange(change);
+                notifyAboutChange(Collections.emptyList(), removedItems, new Change.Removed(0, removedItems.size()));
             }
         }
     }
@@ -204,7 +216,7 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
                 items.set(index, item);
                 index++;
             }
-            notifyAboutChange(new Change<>(Change.Type.CHANGED, updatedItems, position));
+            notifyAboutChange(Collections.emptyList(), Collections.emptyList(), new Change.Changed(position, updatedItems.size(), null));
         }
     }
 
@@ -215,12 +227,24 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
      */
     public void set(@NonNull final Collection<TItem> newItems) {
         synchronized (this) {
-            final Collection<Change<TItem>> changes = Change.calculateCollectionChanges(items, newItems, false);
+            final List<TItem> oldList = new ArrayList<>(items);
+            final List<TItem> newList = new ArrayList<>(newItems);
+            final CollectionsChangesCalculator<TItem> calculator;
+            if (diffUtilsSource != null) {
+                if (diffUtilsSource.sameItemsPredicate != null) {
+                    calculator = new DiffCollectionsChangesCalculator<>(oldList, newList,
+                            diffUtilsSource.detectMoves, diffUtilsSource.sameItemsPredicate, diffUtilsSource.changePayloadProducer);
+                } else {
+                    calculator = new DefaultCollectionsChangesCalculator<>(oldList, newList, false);
+                }
+            } else if (sameItemsPredicate != null) {
+                calculator = new DiffCollectionsChangesCalculator<>(oldList, newList, detectMoves, sameItemsPredicate, changePayloadProducer);
+            } else {
+                calculator = new DefaultCollectionsChangesCalculator<>(oldList, newList, false);
+            }
             items.clear();
             items.addAll(newItems);
-            if (!changes.isEmpty()) {
-                notifyAboutChanges(changes);
-            }
+            notifyAboutChanges(calculator.calculateInsertedItems(), calculator.calculateRemovedItems(), calculator.calculateChanges());
         }
     }
 
@@ -231,6 +255,46 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
         }
     }
 
+    /**
+     * Enable diff utils algorithm in collection changes.
+     *
+     * @param detectMoves           The flag that determines whether the {@link Change.Moved} changes will be generated or not;
+     * @param sameItemsPredicate    Predicate for the determination of the same elements;
+     * @param changePayloadProducer Function that calculate change payload when items the same but contents are different.
+     */
+    public void enableDiffUtils(final boolean detectMoves,
+                                @NonNull final SameItemsPredicate<TItem> sameItemsPredicate,
+                                @Nullable final ChangePayloadProducer<TItem> changePayloadProducer) {
+        this.detectMoves = detectMoves;
+        this.sameItemsPredicate = sameItemsPredicate;
+        this.changePayloadProducer = changePayloadProducer;
+    }
+
+    /**
+     * Disable diff utils algorithm.
+     */
+    public void disableDiffUtils() {
+        this.sameItemsPredicate = null;
+    }
+
+    /**
+     * Returns enabled flag of diff utils.
+     *
+     * @return true if diff utils is enabled.
+     */
+    public boolean diffUtilsIsEnabled() {
+        return diffUtilsSource != null ? diffUtilsSource.diffUtilsIsEnabled() : sameItemsPredicate != null;
+    }
+
+    /**
+     * Sets observableCollection as a source of diff utils parameters;
+     *
+     * @param diffUtilsSource Source of diff utils parameters.
+     */
+    public void setDiffUtilsSource(@Nullable final ObservableList<TItem> diffUtilsSource) {
+        this.diffUtilsSource = diffUtilsSource;
+    }
+
     /**
      * Returns position of item in list.
      *
@@ -243,14 +307,6 @@ public class ObservableList<TItem> extends ObservableCollection<TItem> implement
         }
     }
 
-    @NonNull
-    @Override
-    public Observable<TItem> loadItem(final int position) {
-        synchronized (this) {
-            return position < items.size() ? Observable.just(items.get(position)) : Observable.just(null);
-        }
-    }
-
     private void writeObject(@NonNull final ObjectOutputStream outputStream) throws IOException {
         outputStream.writeObject(items);
     }

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/Change.java

@@ -0,0 +1,134 @@
+/*
+ *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.Nullable;
+
+/**
+ * Created by Gavriil Sitnikov on 23/05/16.
+ * Class representing simple change of collection like insertion, remove or replacing/changing items.
+ */
+public abstract class Change {
+
+    /**
+     * Represents a insert operation in collection.
+     */
+    public static class Inserted extends Change {
+
+        private final int position;
+        private final int count;
+
+        public Inserted(final int position, final int count) {
+            super();
+            this.position = position;
+            this.count = count;
+        }
+
+        public int getPosition() {
+            return position;
+        }
+
+        public int getCount() {
+            return count;
+        }
+
+    }
+
+    /**
+     * Represents a remove operation from collection.
+     */
+    public static class Removed extends Change {
+
+        private final int position;
+        private final int count;
+
+        public Removed(final int position, final int count) {
+            super();
+            this.position = position;
+            this.count = count;
+        }
+
+        public int getPosition() {
+            return position;
+        }
+
+        public int getCount() {
+            return count;
+        }
+
+    }
+
+    /**
+     * Represents a move operation in collection.
+     */
+    public static class Moved extends Change {
+
+        private final int fromPosition;
+        private final int toPosition;
+
+        public Moved(final int fromPosition, final int toPosition) {
+            super();
+            this.fromPosition = fromPosition;
+            this.toPosition = toPosition;
+        }
+
+        public int getFromPosition() {
+            return fromPosition;
+        }
+
+        public int getToPosition() {
+            return toPosition;
+        }
+
+    }
+
+    /**
+     * Represents a modification operation in a collection.
+     */
+    public static class Changed extends Change {
+
+        private final int position;
+        private final int count;
+        @Nullable
+        private final Object payload;
+
+        public Changed(final int position, final int count, @Nullable final Object payload) {
+            super();
+            this.position = position;
+            this.count = count;
+            this.payload = payload;
+        }
+
+        public int getPosition() {
+            return position;
+        }
+
+        public int getCount() {
+            return count;
+        }
+
+        @Nullable
+        public Object getPayload() {
+            return payload;
+        }
+
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/ChangePayloadProducer.java

@@ -0,0 +1,42 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+/**
+ * Functional interface for calculating change payload between two items same type.
+ * Payload calculating when items are same {@link SameItemsPredicate}, but content different.
+ */
+public interface ChangePayloadProducer<TItem> {
+
+
+    /**
+     * Calculate change payload between two items.
+     *
+     * @param item1 First item;
+     * @param item2 Second item;
+     * @return Object that represents minimal changes between two items.
+     */
+    @Nullable
+    Object getChangePayload(@NonNull TItem item1, @NonNull TItem item2);
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/CollectionChanges.java

@@ -0,0 +1,90 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Class which is representing change of collection. There could be multiple changes applied to collection.
+ */
+public class CollectionChanges<TItem> {
+
+    private final int number;
+    @NonNull
+    private final List<TItem> insertedItems;
+    @NonNull
+    private final List<TItem> removedItems;
+    @NonNull
+    private final Collection<Change> changes;
+
+    public CollectionChanges(final int number,
+                             @NonNull final List<TItem> insertedItems,
+                             @NonNull final List<TItem> removedItems,
+                             @NonNull final Collection<Change> changes) {
+        this.number = number;
+        this.insertedItems = Collections.unmodifiableList(insertedItems);
+        this.removedItems = Collections.unmodifiableList(removedItems);
+        this.changes = Collections.unmodifiableCollection(changes);
+    }
+
+    /**
+     * Returns number of change.
+     *
+     * @return Number of change.
+     */
+    public int getNumber() {
+        return number;
+    }
+
+    /**
+     * Returns collection of changes.
+     *
+     * @return Collection of changes.
+     */
+    @NonNull
+    public Collection<Change> getChanges() {
+        return changes;
+    }
+
+    /**
+     * Returns inserted items in change.
+     *
+     * @return Inserted items.
+     */
+    @NonNull
+    public List<TItem> getInsertedItems() {
+        return insertedItems;
+    }
+
+    /**
+     * Returns removed items in change.
+     *
+     * @return Removed items.
+     */
+    @NonNull
+    public List<TItem> getRemovedItems() {
+        return removedItems;
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/CollectionsChangesCalculator.java

@@ -0,0 +1,55 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+
+import java.util.List;
+
+/**
+ * Interface that represent changes calculator between two collections.
+ */
+public interface CollectionsChangesCalculator<TItem> {
+
+    /**
+     * Calculate changes between two collection as collection of objects {@link Change}.
+     *
+     * @return List of changes.
+     */
+    @NonNull
+    List<Change> calculateChanges();
+
+    /**
+     * Calculate changes between two collection as collection of inserted items.
+     *
+     * @return List of inserted item.
+     */
+    @NonNull
+    List<TItem> calculateInsertedItems();
+
+    /**
+     * Calculate changes between two collection as collection of removed items.
+     *
+     * @return List of removed item.
+     */
+    @NonNull
+    List<TItem> calculateRemovedItems();
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/DefaultCollectionsChangesCalculator.java

@@ -0,0 +1,162 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * Default calculator between two collections that use equals function.
+ */
+public class DefaultCollectionsChangesCalculator<TItem> implements CollectionsChangesCalculator<TItem> {
+
+    @NonNull
+    private final Collection<TItem> initialCollection;
+    @NonNull
+    private final Collection<TItem> modifiedCollection;
+    private final boolean shrinkChangesToModifiedSize;
+    @NonNull
+    private final Collection<TItem> itemsToAdd = new ArrayList<>();
+    private int currentSize;
+    private int oldSize;
+    private int newSize;
+    private int couldBeAdded;
+
+    /**
+     * Default calculator of changes between two collections.
+     *
+     * @param initialCollection           Initial collection;
+     * @param modifiedCollection          Changed collection;
+     * @param shrinkChangesToModifiedSize Flag to make position of changed items be less then modified collection size.
+     *                                    It is needed sometimes to not get exceptions like {@link ArrayIndexOutOfBoundsException}.
+     */
+    public DefaultCollectionsChangesCalculator(@NonNull final Collection<TItem> initialCollection,
+                                               @NonNull final Collection<TItem> modifiedCollection,
+                                               final boolean shrinkChangesToModifiedSize) {
+        super();
+        this.initialCollection = initialCollection;
+        this.modifiedCollection = modifiedCollection;
+        this.shrinkChangesToModifiedSize = shrinkChangesToModifiedSize;
+    }
+
+    @NonNull
+    @Override
+    public List<Change> calculateChanges() {
+        int initialOffset = 0;
+        itemsToAdd.clear();
+        currentSize = 0;
+        oldSize = initialCollection.size();
+        newSize = modifiedCollection.size();
+        couldBeAdded = modifiedCollection.size() - initialCollection.size();
+        final List<Change> result = new ArrayList<>();
+        for (final TItem modifiedItem : modifiedCollection) {
+            int foundPosition = 0;
+            for (final Object initialObject : initialCollection) {
+                if (foundPosition >= initialOffset && modifiedItem.equals(initialObject)) {
+                    if (tryAddSkipped(result) == MethodAction.RETURN
+                            || tryRemoveRest(result, foundPosition - initialOffset) == MethodAction.RETURN) {
+                        return result;
+                    }
+                    initialOffset = foundPosition + 1;
+                    currentSize++;
+                    break;
+                }
+                foundPosition++;
+            }
+            // if not found
+            if (foundPosition >= initialCollection.size()) {
+                itemsToAdd.add(modifiedItem);
+            }
+        }
+
+        if (tryAddSkipped(result) == MethodAction.RETURN) {
+            return result;
+        }
+        tryRemoveRest(result, initialCollection.size() - initialOffset);
+        return result;
+    }
+
+    @NonNull
+    @Override
+    public List<TItem> calculateInsertedItems() {
+        final List<TItem> insertedItems = new ArrayList<>();
+        for (final TItem newItem : modifiedCollection) {
+            if (!initialCollection.contains(newItem)) {
+                insertedItems.add(newItem);
+            }
+        }
+        return insertedItems;
+    }
+
+    @NonNull
+    @Override
+    public List<TItem> calculateRemovedItems() {
+        final List<TItem> removedItems = new ArrayList<>();
+        for (final TItem oldItem : initialCollection) {
+            if (!modifiedCollection.contains(oldItem)) {
+                removedItems.add(oldItem);
+            }
+        }
+        return removedItems;
+    }
+
+    @NonNull
+    private MethodAction tryAddSkipped(@NonNull final Collection<Change> changes) {
+        if (!itemsToAdd.isEmpty()) {
+            if (shrinkChangesToModifiedSize && couldBeAdded < itemsToAdd.size()) {
+                addSimpleDifferenceChanges(changes);
+                return MethodAction.RETURN;
+            }
+            changes.add(new Change.Inserted(currentSize, itemsToAdd.size()));
+            currentSize += itemsToAdd.size();
+            couldBeAdded -= itemsToAdd.size();
+            itemsToAdd.clear();
+        }
+        return MethodAction.CONTINUE;
+    }
+
+    @NonNull
+    private MethodAction tryRemoveRest(@NonNull final Collection<Change> changes, final int itemsToRemove) {
+        if (itemsToRemove > 0) {
+            if (shrinkChangesToModifiedSize && couldBeAdded < -itemsToRemove) {
+                addSimpleDifferenceChanges(changes);
+                return MethodAction.RETURN;
+            }
+            changes.add(new Change.Removed(currentSize, itemsToRemove));
+        }
+        return MethodAction.CONTINUE;
+    }
+
+    private void addSimpleDifferenceChanges(@NonNull final Collection<Change> changes) {
+        changes.add(new Change.Changed(currentSize, newSize - currentSize, null));
+        if (oldSize - newSize > 0) {
+            changes.add(new Change.Removed(newSize, oldSize - newSize));
+        }
+    }
+
+    private enum MethodAction {
+        RETURN,
+        CONTINUE
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/DiffCollectionsChangesCalculator.java

@@ -0,0 +1,147 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import ru.touchin.roboswag.core.android.support.v7.util.DiffUtil;
+import ru.touchin.roboswag.core.android.support.v7.util.ListUpdateCallback;
+
+/**
+ * Implementation of {@link CollectionsChangesCalculator} based on DiffUtils from support library.
+ */
+public class DiffCollectionsChangesCalculator<TItem> extends DiffUtil.Callback implements CollectionsChangesCalculator<TItem> {
+
+    @NonNull
+    private final List<TItem> oldList;
+    @NonNull
+    private final List<TItem> newList;
+    private final boolean detectMoves;
+    @NonNull
+    private final SameItemsPredicate<TItem> sameItemsPredicate;
+    @Nullable
+    private final ChangePayloadProducer<TItem> changePayloadProducer;
+
+    public DiffCollectionsChangesCalculator(@NonNull final List<TItem> oldList,
+                                            @NonNull final List<TItem> newList,
+                                            final boolean detectMoves,
+                                            @NonNull final SameItemsPredicate<TItem> sameItemsPredicate,
+                                            @Nullable final ChangePayloadProducer<TItem> changePayloadProducer) {
+        super();
+        this.oldList = oldList;
+        this.newList = newList;
+        this.detectMoves = detectMoves;
+        this.sameItemsPredicate = sameItemsPredicate;
+        this.changePayloadProducer = changePayloadProducer;
+    }
+
+    @NonNull
+    @Override
+    public List<Change> calculateChanges() {
+        final List<Change> changes = new ArrayList<>();
+        DiffUtil.calculateDiff(this, detectMoves).dispatchUpdatesTo(new ListUpdateCallback() {
+            @Override
+            public void onInserted(final int position, final int count) {
+                changes.add(new Change.Inserted(position, count));
+            }
+
+            @Override
+            public void onRemoved(final int position, final int count) {
+                changes.add(new Change.Removed(position, count));
+            }
+
+            @Override
+            public void onMoved(final int fromPosition, final int toPosition) {
+                changes.add(new Change.Moved(fromPosition, toPosition));
+            }
+
+            @Override
+            public void onChanged(final int position, final int count, @Nullable final Object payload) {
+                changes.add(new Change.Changed(position, count, payload));
+            }
+        });
+        return changes;
+    }
+
+    @NonNull
+    @Override
+    public List<TItem> calculateInsertedItems() {
+        final List<TItem> insertedItems = new ArrayList<>();
+        for (final TItem newItem : newList) {
+            if (!containsByPredicate(newItem, oldList)) {
+                insertedItems.add(newItem);
+            }
+        }
+        return insertedItems;
+    }
+
+    @NonNull
+    @Override
+    public List<TItem> calculateRemovedItems() {
+        final List<TItem> removedItems = new ArrayList<>();
+        for (final TItem oldItem : oldList) {
+            if (!containsByPredicate(oldItem, newList)) {
+                removedItems.add(oldItem);
+            }
+        }
+        return removedItems;
+    }
+
+    @Override
+    public int getOldListSize() {
+        return oldList.size();
+    }
+
+    @Override
+    public int getNewListSize() {
+        return newList.size();
+    }
+
+    @Override
+    public boolean areItemsTheSame(final int oldItemPosition, final int newItemPosition) {
+        return sameItemsPredicate.areSame(oldList.get(oldItemPosition), newList.get(newItemPosition));
+    }
+
+    @Override
+    public boolean areContentsTheSame(final int oldItemPosition, final int newItemPosition) {
+        return oldList.get(oldItemPosition).equals(newList.get(newItemPosition));
+    }
+
+    @Nullable
+    @Override
+    public Object getChangePayload(final int oldItemPosition, final int newItemPosition) {
+        return changePayloadProducer != null
+                ? changePayloadProducer.getChangePayload(oldList.get(oldItemPosition), newList.get(newItemPosition)) : null;
+    }
+
+    private boolean containsByPredicate(@NonNull final TItem searchedItem, @NonNull final List<TItem> items) {
+        for (final TItem item : items) {
+            if (sameItemsPredicate.areSame(item, searchedItem)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/changes/SameItemsPredicate.java

@@ -0,0 +1,40 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.collections.changes;
+
+import android.support.annotation.NonNull;
+
+/**
+ * Functional interface for determine same objects. Usually this is just the comparison by id.
+ *
+ * @param <TItem> Type of objects
+ */
+public interface SameItemsPredicate<TItem> {
+
+    /**
+     * Function for determine same objects.
+     *
+     * @param item1 First object;
+     * @param item2 Second object;
+     * @return True if items are same.
+     */
+    boolean areSame(@NonNull TItem item1, @NonNull TItem item2);
+
+}

src/main/java/ru/touchin/roboswag/core/observables/collections/loadable/LoadedRenewableItems.java

@@ -1,49 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.collections.loadable;
-
-import android.support.annotation.Nullable;
-
-/**
- * Created by Gavriil Sitnikov on 23/05/16.
- * Object represents loaded items with reference to load other parts and info of are there more items to load or not.
- *
- * @param <TItem>           Type of items to load;
- * @param <TReference>      Type of reference to load other parts of items;
- * @param <TNewerReference> Type of reference to load newer parts of items.
- */
-public interface LoadedRenewableItems<TItem, TReference, TNewerReference> extends LoadedItems<TItem, TReference> {
-
-    /**
-     * Returns count of new items other than loaded.
-     *
-     * @return Count of new items other than loaded.
-     */
-    int getNewerItemsCount();
-
-    /**
-     * Returns reference to load newer items from this loaded part.
-     *
-     * @return Reference to load newer items.
-     */
-    @Nullable
-    TNewerReference getNewerReference();
-
-}

src/main/java/ru/touchin/roboswag/core/observables/collections/loadable/LoadingMoreList.java

@@ -24,14 +24,16 @@ import android.support.annotation.Nullable;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.concurrent.Executors;
 
 import ru.touchin.roboswag.core.log.Lc;
-import ru.touchin.roboswag.core.observables.collections.Change;
 import ru.touchin.roboswag.core.observables.collections.ObservableCollection;
 import ru.touchin.roboswag.core.observables.collections.ObservableList;
+import ru.touchin.roboswag.core.observables.collections.changes.Change;
+import ru.touchin.roboswag.core.observables.collections.changes.CollectionChanges;
 import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
 import rx.Observable;
 import rx.Scheduler;
@@ -46,6 +48,7 @@ import rx.subjects.BehaviorSubject;
  * {@link ObservableCollection} which is loading items more and more by paging/limit-offset/reference-based mechanisms.
  * To use this collection {@link MoreItemsLoader} should be created.
  * {@link MoreItemsLoader} is an object to load next block of items by info from previous loaded block (last loaded item/reference etc.).
+ * Use {@link #loadItem(int)} and {@link #loadRange(int, int)} to load items asynchronously.
  *
  * @param <TItem>          Type of collection's items;
  * @param <TMoreReference> Type of reference object to help rightly loading next block of items;
@@ -82,10 +85,7 @@ public class LoadingMoreList<TItem, TMoreReference, TLoadedItems extends LoadedI
                            @Nullable final LoadedItems<TItem, TMoreReference> initialItems) {
         super();
         this.loadingMoreObservable = Observable
-                .switchOnNext(Observable.<Observable<TLoadedItems>>create(subscriber -> {
-                    subscriber.onNext(createLoadRequestBasedObservable(this::createActualRequest, moreMoreItemsLoader::load));
-                    subscriber.onCompleted();
-                }))
+                .switchOnNext(Observable.fromCallable(() -> createLoadRequestBasedObservable(this::createActualRequest, moreMoreItemsLoader::load)))
                 .single()
                 .doOnError(throwable -> {
                     if (throwable instanceof IllegalArgumentException || throwable instanceof NoSuchElementException) {
@@ -136,12 +136,14 @@ public class LoadingMoreList<TItem, TMoreReference, TLoadedItems extends LoadedI
 
     @NonNull
     @Override
-    public Observable<CollectionChange<TItem>> observeChanges() {
+    public Observable<CollectionChanges<TItem>> observeChanges() {
         return innerList.observeChanges();
     }
 
     @Override
-    protected void notifyAboutChanges(@NonNull final Collection<Change<TItem>> changes) {
+    protected void notifyAboutChanges(@NonNull final List<TItem> insertedItems,
+                                      @NonNull final List<TItem> removedItems,
+                                      @NonNull final Collection<Change> changes) {
         Lc.assertion("Illegal operation. Modify getInnerList()");
     }
 
@@ -287,25 +289,50 @@ public class LoadingMoreList<TItem, TMoreReference, TLoadedItems extends LoadedI
         return loadingMoreObservable;
     }
 
+    /**
+     * Returns {@link Observable} which is loading item by position.
+     * It could return null in onNext callback if there is no item to load for such position.
+     *
+     * @param position Position to load item;
+     * @return {@link Observable} to load item.
+     */
     @NonNull
-    @Override
     public Observable<TItem> loadItem(final int position) {
         return Observable
                 .switchOnNext(Observable
-                        .<Observable<TItem>>create(subscriber -> {
+                        .fromCallable(() -> {
                             if (position < size()) {
-                                subscriber.onNext(Observable.just(get(position)));
+                                return Observable.just(get(position));
                             } else if (moreItemsCount.getValue() == 0) {
-                                subscriber.onNext(Observable.just((TItem) null));
+                                return Observable.just((TItem) null);
                             } else {
-                                subscriber.onNext(loadingMoreObservable.switchMap(ignored -> Observable.<TItem>error(new NotLoadedYetException())));
+                                return loadingMoreObservable.switchMap(ignored -> Observable.<TItem>error(new NotLoadedYetException()));
                             }
-                            subscriber.onCompleted();
                         })
                         .subscribeOn(loaderScheduler))
                 .retry((number, throwable) -> throwable instanceof NotLoadedYetException);
     }
 
+    /**
+     * Returns {@link Observable} which is loading item by range.
+     * It will return collection of loaded items in onNext callback.
+     *
+     * @param first First position of item to load;
+     * @param last  Last position of item to load;
+     * @return {@link Observable} to load items.
+     */
+    @NonNull
+    public Observable<Collection<TItem>> loadRange(final int first, final int last) {
+        final List<Observable<TItem>> itemsRequests = new ArrayList<>();
+        for (int i = first; i <= last; i++) {
+            itemsRequests.add(loadItem(i));
+        }
+        return Observable.concatEager(itemsRequests)
+                .filter(loadedItem -> loadedItem != null)
+                .toList()
+                .map(Collections::unmodifiableCollection);
+    }
+
     /**
      * Remove all loaded items and resets collection's state.
      */
@@ -335,7 +362,7 @@ public class LoadingMoreList<TItem, TMoreReference, TLoadedItems extends LoadedI
         DO_NOTHING,
         REMOVE_FROM_COLLECTION,
         REMOVE_FROM_LOADED_ITEMS,
-        REPLACE_SOURCE_ITEM_WITH_LOADED,
+        REPLACE_SOURCE_ITEM_WITH_LOADED
     }
 
     /**
@@ -354,7 +381,7 @@ public class LoadingMoreList<TItem, TMoreReference, TLoadedItems extends LoadedI
          * @return Action to do with items.
          */
         @NonNull
-        FilterAction decideFilterAction(@NonNull final TItem collectionObject, @NonNull final TItem loadedItemsObject);
+        FilterAction decideFilterAction(@NonNull TItem collectionObject, @NonNull TItem loadedItemsObject);
 
     }
 

src/main/java/ru/touchin/roboswag/core/observables/collections/loadable/LoadingRenewableList.java

@@ -1,241 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.collections.loadable;
-
-import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
-
-import java.util.NoSuchElementException;
-
-import ru.touchin.roboswag.core.log.Lc;
-import ru.touchin.roboswag.core.observables.collections.ObservableCollection;
-import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
-import rx.Observable;
-import rx.schedulers.Schedulers;
-import rx.subjects.BehaviorSubject;
-
-/**
- * Created by Gavriil Sitnikov on 23/05/16.
- * {@link ObservableCollection} which is loading items more and more by paging/limit-offset/reference-based mechanisms but also it is providing
- * interface to load newer items and info about it's loading availability.
- * To use this collection {@link MoreItemsLoader} and {@link NewerItemsLoader} should be created.
- *
- * @param <TItem>           Type of collection's items;
- * @param <TReference>      Type of reference object to help rightly loading next block of items;
- * @param <TNewerReference> Type of reference object to help rightly loading block of newer items;
- * @param <TLoadedItems>    Type of loading block of items.
- */
-public class LoadingRenewableList<TItem, TReference, TNewerReference,
-        TLoadedItems extends LoadedRenewableItems<TItem, TReference, TNewerReference>>
-        extends LoadingMoreList<TItem, TReference, TLoadedItems> {
-
-    @Nullable
-    private TNewerReference newerReference;
-    @NonNull
-    private final BehaviorSubject<Integer> newerItemsCount = BehaviorSubject.create(LoadedItems.UNKNOWN_ITEMS_COUNT);
-
-    @NonNull
-    private final Observable<TLoadedItems> loadingNewerObservable;
-    @NonNull
-    private final Observable<TLoadedItems> loadingNewestObservable;
-
-    public LoadingRenewableList(@NonNull final MoreItemsLoader<TItem, TReference, TLoadedItems> moreMoreItemsLoader,
-                                @NonNull final NewerItemsLoader<TItem, TReference, TNewerReference, TLoadedItems> newerItemsLoader) {
-        super(moreMoreItemsLoader);
-        this.loadingNewerObservable = createLoadingNewerObservable(newerItemsLoader, false);
-        this.loadingNewestObservable = createLoadingNewerObservable(newerItemsLoader, true);
-    }
-
-    public LoadingRenewableList(@NonNull final MoreItemsLoader<TItem, TReference, TLoadedItems> moreMoreItemsLoader,
-                                @NonNull final NewerItemsLoader<TItem, TReference, TNewerReference, TLoadedItems> newerItemsLoader,
-                                @Nullable final TLoadedItems initialItems) {
-        super(moreMoreItemsLoader, initialItems);
-        this.loadingNewerObservable = createLoadingNewerObservable(newerItemsLoader, false);
-        this.loadingNewestObservable = createLoadingNewerObservable(newerItemsLoader, true);
-        if (initialItems != null) {
-            updateNewerReference(initialItems);
-        }
-    }
-
-    @NonNull
-    private Observable<TLoadedItems> waitForInitialLoading(@NonNull final Observable<TLoadedItems> observable) {
-        return getLoadingMoreObservable().ignoreElements().concatWith(observable);
-    }
-
-    @NonNull
-    private NewerLoadRequest<TNewerReference> createActualRequest() {
-        return new NewerLoadRequest<>(newerReference, newerItemsCount.getValue());
-    }
-
-    @NonNull
-    private Observable<TLoadedItems> createLoadingNewerObservable(
-            @NonNull final NewerItemsLoader<TItem, TReference, TNewerReference, TLoadedItems> newerItemsLoader,
-            final boolean renew) {
-        return Observable
-                .switchOnNext(Observable.<Observable<TLoadedItems>>create(subscriber -> {
-                    if (!renew) {
-                        subscriber.onNext(Observable.concat(
-                                //we need non-empty list to start loading newer items or we need to wait any change (should be insertion)
-                                isEmpty() ? observeChanges().first().switchMap(ignored -> Observable.empty()) : Observable.empty(),
-                                createLoadRequestBasedObservable(this::createActualRequest,
-                                        loadRequest -> loadRequest.getNewerReference() == null && isEmpty()
-                                                ? waitForInitialLoading(newerItemsLoader.load(loadRequest))
-                                                : newerItemsLoader.load(loadRequest))));
-                    } else {
-                        subscriber.onNext(newerItemsLoader.load(new NewerLoadRequest<>(null, LoadedItems.UNKNOWN_ITEMS_COUNT))
-                                .subscribeOn(Schedulers.io())
-                                .observeOn(getLoaderScheduler()));
-                    }
-                    subscriber.onCompleted();
-                }))
-                .single()
-                .doOnError(throwable -> {
-                    if (throwable instanceof IllegalArgumentException || throwable instanceof NoSuchElementException) {
-                        Lc.assertion(new ShouldNotHappenException("Updates during loading not supported."
-                                + " NewerItemsLoader should emit only one result.", throwable));
-                    }
-                })
-                .doOnNext(loadedItems -> onNewerItemsLoaded(loadedItems, renew))
-                .replay(1)
-                .refCount();
-    }
-
-    /**
-     * Returns if there are new items to load.
-     *
-     * @return True if there are more items to load.
-     */
-    public boolean hasNewerItems() {
-        return newerItemsCount.getValue() != 0;
-    }
-
-    /**
-     * Update a new items count.
-     *
-     * @param count new items count
-     */
-    public void updateNewerItemsCount(final int count) {
-        newerItemsCount.onNext(count);
-    }
-
-    /**
-     * Returns {@link Observable} which is providing status of if is there are new items to load or not.
-     *
-     * @return {@link Observable} of more items availability status.
-     */
-    @NonNull
-    public Observable<Boolean> observeHasNewerItems() {
-        return newerItemsCount.map(count -> count != 0).distinctUntilChanged();
-    }
-
-    /**
-     * Returns {@link Observable} which is providing count of new items to load.
-     *
-     * @return {@link Observable} of new items availability status.
-     */
-    @NonNull
-    public Observable<Integer> observeNewerItemsCount() {
-        return newerItemsCount.distinctUntilChanged();
-    }
-
-    @Override
-    protected void onItemsLoaded(@NonNull final TLoadedItems loadedItems, final int insertPosition, final boolean reset) {
-        super.onItemsLoaded(loadedItems, insertPosition, reset);
-        if (newerReference == null) {
-            updateNewerReference(loadedItems);
-        }
-    }
-
-    /**
-     * Calls when newer items part loaded.
-     *
-     * @param loadedItems Loaded items;
-     * @param renew       Flag indicates is it loading just to load some new items (false) or to load totally new items (true).
-     */
-    protected void onNewerItemsLoaded(@NonNull final TLoadedItems loadedItems, final boolean renew) {
-        onItemsLoaded(loadedItems, 0, renew);
-        updateNewerReference(loadedItems);
-    }
-
-    @Override
-    protected void resetState() {
-        super.resetState();
-        newerReference = null;
-        newerItemsCount.onNext(LoadedItems.UNKNOWN_ITEMS_COUNT);
-    }
-
-    /**
-     * Returns {@link Observable} that will load newer items by count returned by last loaded items part.
-     *
-     * @return {@link Observable} to load newer items.
-     */
-    @NonNull
-    public Observable<TLoadedItems> loadNewer() {
-        return loadingNewerObservable.first();
-    }
-
-    /**
-     * Returns {@link Observable} that will load all newer items.
-     *
-     * @return Returns {@link Observable} to limited load newer items.
-     */
-    @NonNull
-    public Observable<?> loadToNewest() {
-        return loadToNewest(Integer.MAX_VALUE);
-    }
-
-    /**
-     * Returns {@link Observable} that will load some newer itemslimited by maximum pages loading results.
-     *
-     * @param maxPageDeep Limit to load pages;
-     * @return Returns {@link Observable} to limited load newer items.
-     */
-    @NonNull
-    public Observable<?> loadToNewest(final int maxPageDeep) {
-        return Observable
-                .switchOnNext(Observable
-                        .<Observable<?>>create(subscriber -> {
-                            subscriber.onNext(newerItemsCount.getValue() == 0
-                                    ? Observable.empty()
-                                    : loadingNewerObservable.switchMap(ignored -> Observable.error(new NotLoadedYetException())));
-                            subscriber.onCompleted();
-                        })
-                        .subscribeOn(getLoaderScheduler()))
-                .retry((number, throwable) -> number <= maxPageDeep && throwable instanceof NotLoadedYetException);
-    }
-
-    /**
-     * Returns {@link Observable} that tries to load some newer items even if there are no info about count of them.
-     *
-     * @return {@link Observable} to load newer items.
-     */
-    @NonNull
-    public Observable<TLoadedItems> renew() {
-        return loadingNewestObservable.first();
-    }
-
-    private void updateNewerReference(@NonNull final TLoadedItems loadedItems) {
-        if (loadedItems.getNewerReference() != null) {
-            newerReference = loadedItems.getNewerReference();
-        }
-        newerItemsCount.onNext(loadedItems.getNewerItemsCount());
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/collections/loadable/NewerItemsLoader.java

@@ -1,40 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.collections.loadable;
-
-import android.support.annotation.NonNull;
-
-import rx.Observable;
-
-/**
- * Created by Gavriil Sitnikov on 02/06/2016.
- * Object that is loading new part of items by reference.
- *
- * @param <TItem>           Type of items to be loaded;
- * @param <TNewerReference> Type of reference to be used to load new part of items;
- * @param <TLoadedItems>    Type of loaded items part.
- */
-public interface NewerItemsLoader<TItem, TReference, TNewerReference,
-        TLoadedItems extends LoadedRenewableItems<TItem, TReference, TNewerReference>> {
-
-    @NonNull
-    Observable<TLoadedItems> load(@NonNull final NewerLoadRequest<TNewerReference> newerLoadRequest);
-
-}

src/main/java/ru/touchin/roboswag/core/observables/collections/loadable/NewerLoadRequest.java

@@ -1,74 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.collections.loadable;
-
-import android.support.annotation.Nullable;
-
-import ru.touchin.roboswag.core.utils.ObjectUtils;
-
-/**
- * Created by Gavriil Sitnikov on 02/06/2016.
- * Request represents request to load new part of items.
- *
- * @param <TNewerReference> Type of reference to load new part of items.
- */
-public class NewerLoadRequest<TNewerReference> {
-
-    @Nullable
-    private final TNewerReference newerReference;
-    private final int newerItemsCount;
-
-    public NewerLoadRequest(@Nullable final TNewerReference newerReference, final int newerItemsCount) {
-        this.newerReference = newerReference;
-        this.newerItemsCount = newerItemsCount;
-    }
-
-    /**
-     * Returns reference to be used to load new part of items.
-     *
-     * @return Reference object.
-     */
-    @Nullable
-    public TNewerReference getNewerReference() {
-        return newerReference;
-    }
-
-    /**
-     * Count of newer items to load.
-     *
-     * @return Count of newer items to load.
-     */
-    public int getNewerItemsCount() {
-        return newerItemsCount;
-    }
-
-    @Override
-    public boolean equals(@Nullable final Object object) {
-        return object instanceof NewerLoadRequest
-                && ObjectUtils.equals(((NewerLoadRequest) object).newerReference, newerReference)
-                && ((NewerLoadRequest) object).newerItemsCount == newerItemsCount;
-    }
-
-    @Override
-    public int hashCode() {
-        return newerItemsCount + (newerReference != null ? newerReference.hashCode() : 0);
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/storable/BaseStorable.java

@@ -0,0 +1,482 @@
+/*
+ *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.observables.storable;
+
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+import java.lang.reflect.Type;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import ru.touchin.roboswag.core.log.Lc;
+import ru.touchin.roboswag.core.log.LcGroup;
+import ru.touchin.roboswag.core.observables.OnSubscribeRefCountWithCacheTime;
+import ru.touchin.roboswag.core.utils.ObjectUtils;
+import ru.touchin.roboswag.core.utils.Optional;
+import rx.Completable;
+import rx.Observable;
+import rx.Scheduler;
+import rx.Single;
+import rx.exceptions.OnErrorThrowable;
+import rx.functions.Actions;
+import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
+
+/**
+ * Created by Gavriil Sitnikov on 04/10/2015.
+ * Base class allows to async access to some store.
+ * Supports conversion between store and actual value. If it is not needed then use {@link SameTypesConverter}
+ * Supports migration from specific version to latest by {@link Migration} object.
+ * Allows to set default value which will be returned if actual value is null.
+ * Allows to declare specific {@link ObserveStrategy}.
+ * Also specific {@link Scheduler} could be specified to not create new scheduler per storable.
+ *
+ * @param <TKey>          Type of key to identify object;
+ * @param <TObject>       Type of actual object;
+ * @param <TStoreObject>  Type of store object. Could be same as {@link TObject};
+ * @param <TReturnObject> Type of actual value operating by Storable. Could be same as {@link TObject}.
+ */
+public abstract class BaseStorable<TKey, TObject, TStoreObject, TReturnObject> {
+
+    public static final LcGroup STORABLE_LC_GROUP = new LcGroup("STORABLE");
+
+    private static final long DEFAULT_CACHE_TIME_MILLIS = TimeUnit.SECONDS.toMillis(5);
+
+    @NonNull
+    private static ObserveStrategy getDefaultObserveStrategyFor(@NonNull final Type objectType, @NonNull final Type storeObjectType) {
+        if (objectType instanceof Class && ObjectUtils.isSimpleClass((Class) objectType)) {
+            return ObserveStrategy.CACHE_ACTUAL_VALUE;
+        }
+        if (objectType instanceof Class && ObjectUtils.isSimpleClass((Class) storeObjectType)) {
+            return ObserveStrategy.CACHE_STORE_VALUE;
+        }
+        return ObserveStrategy.NO_CACHE;
+    }
+
+    @NonNull
+    private final TKey key;
+    @NonNull
+    private final Type objectType;
+    @NonNull
+    private final Type storeObjectType;
+    @NonNull
+    private final Store<TKey, TStoreObject> store;
+    @NonNull
+    private final Converter<TObject, TStoreObject> converter;
+    @NonNull
+    private final PublishSubject<Optional<TStoreObject>> newStoreValueEvent = PublishSubject.create();
+    @NonNull
+    private final Observable<Optional<TStoreObject>> storeValueObservable;
+    @NonNull
+    private final Observable<Optional<TObject>> valueObservable;
+    @NonNull
+    private final Scheduler scheduler;
+
+    public BaseStorable(@NonNull final BuilderCore<TKey, TObject, TStoreObject> builderCore) {
+        this(builderCore.key, builderCore.objectType, builderCore.storeObjectType,
+                builderCore.store, builderCore.converter, builderCore.observeStrategy,
+                builderCore.migration, builderCore.defaultValue, builderCore.storeScheduler, builderCore.cacheTimeMillis);
+    }
+
+    @SuppressWarnings("PMD.ExcessiveParameterList")
+    //ExcessiveParameterList: that's why we are using builder to create it
+    private BaseStorable(@NonNull final TKey key,
+                         @NonNull final Type objectType,
+                         @NonNull final Type storeObjectType,
+                         @NonNull final Store<TKey, TStoreObject> store,
+                         @NonNull final Converter<TObject, TStoreObject> converter,
+                         @Nullable final ObserveStrategy observeStrategy,
+                         @Nullable final Migration<TKey> migration,
+                         @Nullable final TObject defaultValue,
+                         @Nullable final Scheduler storeScheduler,
+                         final long cacheTimeMillis) {
+        this.key = key;
+        this.objectType = objectType;
+        this.storeObjectType = storeObjectType;
+        this.store = store;
+        this.converter = converter;
+
+        final ObserveStrategy nonNullObserveStrategy
+                = observeStrategy != null ? observeStrategy : getDefaultObserveStrategyFor(objectType, storeObjectType);
+        scheduler = storeScheduler != null ? storeScheduler : Schedulers.from(Executors.newSingleThreadExecutor());
+        storeValueObservable
+                = createStoreValueObservable(nonNullObserveStrategy, migration, defaultValue, cacheTimeMillis);
+        valueObservable = createValueObservable(storeValueObservable, nonNullObserveStrategy, cacheTimeMillis);
+    }
+
+    @Nullable
+    private Optional<TStoreObject> returnDefaultValueIfNull(@NonNull final Optional<TStoreObject> storeObject, @Nullable final TObject defaultValue) {
+        if (storeObject.get() != null || defaultValue == null) {
+            return storeObject;
+        }
+
+        try {
+            return new Optional<>(converter.toStoreObject(objectType, storeObjectType, defaultValue));
+        } catch (final Converter.ConversionException exception) {
+            STORABLE_LC_GROUP.w(exception, "Exception while converting default value of '%s' from '%s' from store %s",
+                    key, defaultValue, store);
+            throw OnErrorThrowable.from(exception);
+        }
+    }
+
+    @NonNull
+    private Observable<Optional<TStoreObject>> createStoreInitialLoadingObservable(@Nullable final Migration<TKey> migration) {
+        final Single<Optional<TStoreObject>> loadObservable = store.loadObject(storeObjectType, key)
+                .doOnError(throwable -> STORABLE_LC_GROUP.w(throwable, "Exception while trying to load value of '%s' from store %s", key, store));
+        return (migration != null ? migration.migrateToLatestVersion(key).andThen(loadObservable) : loadObservable)
+                .subscribeOn(scheduler)
+                .observeOn(scheduler)
+                .toObservable()
+                .replay(1)
+                .refCount()
+                .take(1);
+    }
+
+    @NonNull
+    private Observable<Optional<TStoreObject>> createStoreValueObservable(@NonNull final ObserveStrategy observeStrategy,
+                                                                          @Nullable final Migration<TKey> migration,
+                                                                          @Nullable final TObject defaultValue,
+                                                                          final long cacheTimeMillis) {
+        final Observable<Optional<TStoreObject>> storeInitialLoadingObservable = createStoreInitialLoadingObservable(migration);
+        final Observable<Optional<TStoreObject>> result = storeInitialLoadingObservable
+                .concatWith(newStoreValueEvent)
+                .map(storeObject -> returnDefaultValueIfNull(storeObject, defaultValue));
+        return observeStrategy == ObserveStrategy.CACHE_STORE_VALUE || observeStrategy == ObserveStrategy.CACHE_STORE_AND_ACTUAL_VALUE
+                ? Observable.unsafeCreate(new OnSubscribeRefCountWithCacheTime<>(result.replay(1), cacheTimeMillis, TimeUnit.MILLISECONDS))
+                : result;
+    }
+
+    @NonNull
+    private Observable<Optional<TObject>> createValueObservable(@NonNull final Observable<Optional<TStoreObject>> storeValueObservable,
+                                                                @NonNull final ObserveStrategy observeStrategy,
+                                                                final long cacheTimeMillis) {
+        final Observable<Optional<TObject>> result = storeValueObservable
+                .map(storeObject -> {
+                    try {
+                        return new Optional<>(converter.toObject(objectType, storeObjectType, storeObject.get()));
+                    } catch (final Converter.ConversionException exception) {
+                        STORABLE_LC_GROUP.w(exception, "Exception while trying to converting value of '%s' from store %s by %s",
+                                key, storeObject, store, converter);
+                        throw OnErrorThrowable.from(exception);
+                    }
+                });
+        return observeStrategy == ObserveStrategy.CACHE_ACTUAL_VALUE || observeStrategy == ObserveStrategy.CACHE_STORE_AND_ACTUAL_VALUE
+                ? Observable.unsafeCreate(new OnSubscribeRefCountWithCacheTime<>(result.replay(1), cacheTimeMillis, TimeUnit.MILLISECONDS))
+                : result;
+    }
+
+    /**
+     * Returns key of value.
+     *
+     * @return Unique key.
+     */
+    @NonNull
+    public TKey getKey() {
+        return key;
+    }
+
+    /**
+     * Returns type of actual object.
+     *
+     * @return Type of actual object.
+     */
+    @NonNull
+    public Type getObjectType() {
+        return objectType;
+    }
+
+    /**
+     * Returns type of store object.
+     *
+     * @return Type of store object.
+     */
+    @NonNull
+    public Type getStoreObjectType() {
+        return storeObjectType;
+    }
+
+    /**
+     * Returns {@link Store} where store class representation of object is storing.
+     *
+     * @return Store.
+     */
+    @NonNull
+    public Store<TKey, TStoreObject> getStore() {
+        return store;
+    }
+
+    /**
+     * Returns {@link Converter} to convert values from store class to actual and back.
+     *
+     * @return Converter.
+     */
+    @NonNull
+    public Converter<TObject, TStoreObject> getConverter() {
+        return converter;
+    }
+
+    @NonNull
+    private Completable internalSet(@Nullable final TObject newValue, final boolean checkForEqualityBeforeSet) {
+        return (checkForEqualityBeforeSet ? storeValueObservable.take(1).toSingle() : Single.just(new Optional<>(null)))
+                .observeOn(scheduler)
+                .flatMapCompletable(oldStoreValue -> {
+                    final TStoreObject newStoreValue;
+                    try {
+                        newStoreValue = converter.toStoreObject(objectType, storeObjectType, newValue);
+                    } catch (final Converter.ConversionException exception) {
+                        STORABLE_LC_GROUP.w(exception, "Exception while trying to store value of '%s' from store %s by %s",
+                                key, newValue, store, converter);
+                        return Completable.error(exception);
+                    }
+                    if (checkForEqualityBeforeSet && ObjectUtils.equals(newStoreValue, oldStoreValue.get())) {
+                        return Completable.complete();
+                    }
+                    return store.storeObject(storeObjectType, key, newStoreValue)
+                            .doOnError(throwable -> STORABLE_LC_GROUP.w(throwable,
+                                    "Exception while trying to store value of '%s' from store %s by %s",
+                                    key, newValue, store, converter))
+                            .observeOn(scheduler)
+                            .andThen(Completable.fromAction(() -> {
+                                newStoreValueEvent.onNext(new Optional<>(newStoreValue));
+                                if (checkForEqualityBeforeSet) {
+                                    STORABLE_LC_GROUP.i("Value of '%s' changed from '%s' to '%s'", key, oldStoreValue, newStoreValue);
+                                } else {
+                                    STORABLE_LC_GROUP.i("Value of '%s' force changed to '%s'", key, newStoreValue);
+                                }
+                            }));
+                });
+    }
+
+    /**
+     * Creates observable which is async setting value to store.
+     * It is not checking if stored value equals new value.
+     * In result it will be faster to not get value from store and compare but it will emit item to {@link #observe()} subscribers.
+     * NOTE: It could emit ONLY completed and errors events. It is not providing onNext event!
+     *
+     * @param newValue Value to set;
+     * @return Observable of setting process.
+     */
+    //COMPATIBILITY NOTE: it is not Completable to prevent migration of old code
+    @NonNull
+    public Observable<?> forceSet(@Nullable final TObject newValue) {
+        return internalSet(newValue, false).toObservable();
+    }
+
+    /**
+     * Creates observable which is async setting value to store.
+     * It is checking if stored value equals new value.
+     * In result it will take time to get value from store and compare
+     * but it won't emit item to {@link #observe()} subscribers if stored value equals new value.
+     * NOTE: It could emit ONLY completed and errors events. It is not providing onNext event!
+     *
+     * @param newValue Value to set;
+     * @return Observable of setting process.
+     */
+    //COMPATIBILITY NOTE: it is not Completable to prevent migration of old code
+    @NonNull
+    public Observable<?> set(@Nullable final TObject newValue) {
+        return internalSet(newValue, true).toObservable();
+    }
+
+    @Deprecated
+    //COMPATIBILITY NOTE: it is deprecated as it's execution not bound to Android lifecycle objects
+    public void setCalm(@Nullable final TObject newValue) {
+        set(newValue).subscribe(Actions.empty(), Lc::assertion);
+    }
+
+    /**
+     * Sets value synchronously. You should NOT use this method normally. Use {@link #set(Object)} asynchronously instead.
+     *
+     * @param newValue Value to set;
+     */
+    @Deprecated
+    //deprecation: it should be used for debug only and in very rare cases.
+    public void setSync(@Nullable final TObject newValue) {
+        set(newValue).toBlocking().subscribe();
+    }
+
+    @NonNull
+    protected Observable<Optional<TObject>> observeOptionalValue() {
+        return valueObservable;
+    }
+
+    /**
+     * Returns Observable which is emitting item on subscribe and every time when someone have changed value.
+     * It could emit next and error events but not completed.
+     *
+     * @return Returns observable of value.
+     */
+    @NonNull
+    public abstract Observable<TReturnObject> observe();
+
+    /**
+     * Returns Observable which is emitting only one item on subscribe.
+     * It could emit next and error events but not completed.
+     *
+     * @return Returns observable of value.
+     */
+    @NonNull
+    //COMPATIBILITY NOTE: it is not Single to prevent migration of old code
+    public Observable<TReturnObject> get() {
+        return observe().take(1);
+    }
+
+    /**
+     * Gets value synchronously. You should NOT use this method normally. Use {@link #get()} or {@link #observe()} asynchronously instead.
+     *
+     * @return Returns value;
+     */
+    @Deprecated
+    //deprecation: it should be used for debug only and in very rare cases.
+    @Nullable
+    public TReturnObject getSync() {
+        return get().toBlocking().first();
+    }
+
+    /**
+     * Enum that is representing strategy of observing item from store.
+     */
+    public enum ObserveStrategy {
+
+        /**
+         * Not caching value so on every {@link #get()} emit it will get value from {@link #getStore()} and converts it with {@link #getConverter()}.
+         */
+        NO_CACHE,
+        /**
+         * Caching only store value so on every {@link #get()} emit it will converts it with {@link #getConverter()}.
+         * Do not use such strategy if store object could be big (like byte-array of file).
+         */
+        CACHE_STORE_VALUE,
+        /**
+         * Caching value so it won't spend time for getting value from {@link #getStore()} and converts it by {@link #getConverter()}.
+         * But it will take time for getting value from {@link #getStore()} to set value.
+         * Do not use such strategy if object could be big (like Bitmap or long string).
+         * Do not use such strategy if object is mutable because multiple subscribers could then change it's state.
+         */
+        CACHE_ACTUAL_VALUE,
+        /**
+         * Caching value so it won't spend time for getting value from {@link #getStore()} and converts it by {@link #getConverter()}.
+         * It won't take time or getting value from {@link #getStore()} to set value.
+         * Do not use such strategy if store object could be big (like byte-array of file).
+         * Do not use such strategy if object could be big (like Bitmap or long string).
+         * Do not use such strategy if object is mutable because multiple subscribers could then change it's state.
+         */
+        CACHE_STORE_AND_ACTUAL_VALUE
+
+    }
+
+    /**
+     * Helper class to create various builders.
+     *
+     * @param <TKey>         Type of key to identify object;
+     * @param <TObject>      Type of actual object;
+     * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
+     */
+    public static class BuilderCore<TKey, TObject, TStoreObject> {
+
+        @NonNull
+        protected final TKey key;
+        @NonNull
+        protected final Type objectType;
+        @NonNull
+        private final Type storeObjectType;
+        @NonNull
+        private final Store<TKey, TStoreObject> store;
+        @NonNull
+        private final Converter<TObject, TStoreObject> converter;
+        @Nullable
+        private ObserveStrategy observeStrategy;
+        @Nullable
+        private Migration<TKey> migration;
+        @Nullable
+        private TObject defaultValue;
+        @Nullable
+        private Scheduler storeScheduler;
+        private long cacheTimeMillis;
+
+        protected BuilderCore(@NonNull final TKey key,
+                              @NonNull final Type objectType,
+                              @NonNull final Type storeObjectType,
+                              @NonNull final Store<TKey, TStoreObject> store,
+                              @NonNull final Converter<TObject, TStoreObject> converter) {
+            this(key, objectType, storeObjectType, store, converter, null, null, null, null, DEFAULT_CACHE_TIME_MILLIS);
+        }
+
+        protected BuilderCore(@NonNull final BuilderCore<TKey, TObject, TStoreObject> sourceBuilder) {
+            this(sourceBuilder.key, sourceBuilder.objectType, sourceBuilder.storeObjectType,
+                    sourceBuilder.store, sourceBuilder.converter, sourceBuilder.observeStrategy,
+                    sourceBuilder.migration, sourceBuilder.defaultValue, sourceBuilder.storeScheduler, sourceBuilder.cacheTimeMillis);
+        }
+
+        @SuppressWarnings({"PMD.ExcessiveParameterList", "CPD-START"})
+        //CPD: it is same code as constructor of Storable
+        //ExcessiveParameterList: that's why we are using builder to create it
+        private BuilderCore(@NonNull final TKey key,
+                            @NonNull final Type objectType,
+                            @NonNull final Type storeObjectType,
+                            @NonNull final Store<TKey, TStoreObject> store,
+                            @NonNull final Converter<TObject, TStoreObject> converter,
+                            @Nullable final ObserveStrategy observeStrategy,
+                            @Nullable final Migration<TKey> migration,
+                            @Nullable final TObject defaultValue,
+                            @Nullable final Scheduler storeScheduler,
+                            final long cacheTimeMillis) {
+            this.key = key;
+            this.objectType = objectType;
+            this.storeObjectType = storeObjectType;
+            this.store = store;
+            this.converter = converter;
+            this.observeStrategy = observeStrategy;
+            this.migration = migration;
+            this.defaultValue = defaultValue;
+            this.storeScheduler = storeScheduler;
+            this.cacheTimeMillis = cacheTimeMillis;
+        }
+
+        @SuppressWarnings("CPD-END")
+        protected void setStoreSchedulerInternal(@Nullable final Scheduler storeScheduler) {
+            this.storeScheduler = storeScheduler;
+        }
+
+        protected void setObserveStrategyInternal(@Nullable final ObserveStrategy observeStrategy) {
+            this.observeStrategy = observeStrategy;
+        }
+
+        protected void setMigrationInternal(@NonNull final Migration<TKey> migration) {
+            this.migration = migration;
+        }
+
+        protected void setCacheTimeInternal(final long cacheTime, @NonNull final TimeUnit timeUnit) {
+            this.cacheTimeMillis = timeUnit.toMillis(cacheTime);
+        }
+
+        @Nullable
+        protected TObject getDefaultValue() {
+            return defaultValue;
+        }
+
+        protected void setDefaultValueInternal(@NonNull final TObject defaultValue) {
+            this.defaultValue = defaultValue;
+        }
+
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/observables/storable/Converter.java

@@ -38,7 +38,7 @@ public interface Converter<TObject, TStoreObject> {
      *
      * @param objectType      Type of object;
      * @param storeObjectType Type of store object allowed to store;
-     * @param object           Object to be converted to store object;
+     * @param object          Object to be converted to store object;
      * @return Object that is allowed to store into specific {@link Store};
      * @throws ConversionException Exception during conversion. Usually it indicates illegal state.
      */
@@ -51,7 +51,7 @@ public interface Converter<TObject, TStoreObject> {
      *
      * @param objectType      Type of object;
      * @param storeObjectType Type of store object allowed to store;
-     * @param storeObject      Object from specific {@link Store};
+     * @param storeObject     Object from specific {@link Store};
      * @return Object converted from store object;
      * @throws ConversionException Exception during conversion. Usually it indicates illegal state.
      */

src/main/java/ru/touchin/roboswag/core/observables/storable/Migration.java

@@ -58,7 +58,7 @@ public class Migration<TKey> {
     @NonNull
     private Single<Long> loadCurrentVersion(@NonNull final TKey key) {
         return versionsStore.loadObject(Long.class, key)
-                .map(version -> version != null ? version : DEFAULT_VERSION)
+                .map(version -> version.get() != null ? version.get() : DEFAULT_VERSION)
                 .onErrorResumeNext(throwable
                         -> Single.error(new MigrationException(String.format("Can't get version of '%s' from %s", key, versionsStore), throwable)));
     }

src/main/java/ru/touchin/roboswag/core/observables/storable/Storable.java

@@ -23,22 +23,12 @@ import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 
 import java.lang.reflect.Type;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import ru.touchin.roboswag.core.log.Lc;
-import ru.touchin.roboswag.core.log.LcGroup;
-import ru.touchin.roboswag.core.observables.OnSubscribeRefCountWithCacheTime;
-import ru.touchin.roboswag.core.observables.storable.builders.NonNullStorableBuilder;
-import ru.touchin.roboswag.core.utils.ObjectUtils;
-import rx.Completable;
+import ru.touchin.roboswag.core.observables.storable.concrete.NonNullStorable;
+import ru.touchin.roboswag.core.utils.Optional;
 import rx.Observable;
 import rx.Scheduler;
-import rx.Single;
-import rx.exceptions.OnErrorThrowable;
-import rx.functions.Actions;
-import rx.schedulers.Schedulers;
-import rx.subjects.PublishSubject;
 
 /**
  * Created by Gavriil Sitnikov on 04/10/2015.
@@ -53,421 +43,17 @@ import rx.subjects.PublishSubject;
  * @param <TObject>      Type of actual object;
  * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
  */
-public class Storable<TKey, TObject, TStoreObject> {
-
-    public static final LcGroup STORABLE_LC_GROUP = new LcGroup("STORABLE");
-
-    private static final long DEFAULT_CACHE_TIME_MILLIS = TimeUnit.SECONDS.toMillis(5);
-
-    @NonNull
-    private static ObserveStrategy getDefaultObserveStrategyFor(@NonNull final Type objectType, @NonNull final Type storeObjectType) {
-        if (objectType instanceof Class && ObjectUtils.isSimpleClass((Class) objectType)) {
-            return ObserveStrategy.CACHE_ACTUAL_VALUE;
-        }
-        if (objectType instanceof Class && ObjectUtils.isSimpleClass((Class) storeObjectType)) {
-            return ObserveStrategy.CACHE_STORE_VALUE;
-        }
-        return ObserveStrategy.NO_CACHE;
-    }
-
-    @NonNull
-    private final TKey key;
-    @NonNull
-    private final Type objectType;
-    @NonNull
-    private final Type storeObjectType;
-    @NonNull
-    private final Store<TKey, TStoreObject> store;
-    @NonNull
-    private final Converter<TObject, TStoreObject> converter;
-    @NonNull
-    private final PublishSubject<TStoreObject> newStoreValueEvent = PublishSubject.create();
-    @NonNull
-    private final Observable<TStoreObject> storeValueObservable;
-    @NonNull
-    private final Observable<TObject> valueObservable;
-    @NonNull
-    private final Scheduler scheduler;
+//COMPATIBILITY NOTE: in RxJava2 it should extends BaseStorable<TKey, TObject, TStoreObject, Optional<TObject>>
+public class Storable<TKey, TObject, TStoreObject> extends BaseStorable<TKey, TObject, TStoreObject, TObject> {
 
     public Storable(@NonNull final BuilderCore<TKey, TObject, TStoreObject> builderCore) {
-        this(builderCore.key, builderCore.objectType, builderCore.storeObjectType,
-                builderCore.store, builderCore.converter, builderCore.observeStrategy,
-                builderCore.migration, builderCore.defaultValue, builderCore.storeScheduler, builderCore.cacheTimeMillis);
-    }
-
-    @SuppressWarnings("PMD.ExcessiveParameterList")
-    //ExcessiveParameterList: that's why we are using builder to create it
-    private Storable(@NonNull final TKey key,
-                     @NonNull final Type objectType,
-                     @NonNull final Type storeObjectType,
-                     @NonNull final Store<TKey, TStoreObject> store,
-                     @NonNull final Converter<TObject, TStoreObject> converter,
-                     @Nullable final ObserveStrategy observeStrategy,
-                     @Nullable final Migration<TKey> migration,
-                     @Nullable final TObject defaultValue,
-                     @Nullable final Scheduler storeScheduler,
-                     final long cacheTimeMillis) {
-        this.key = key;
-        this.objectType = objectType;
-        this.storeObjectType = storeObjectType;
-        this.store = store;
-        this.converter = converter;
-
-        final ObserveStrategy nonNullObserveStrategy
-                = observeStrategy != null ? observeStrategy : getDefaultObserveStrategyFor(objectType, storeObjectType);
-        scheduler = storeScheduler != null ? storeScheduler : Schedulers.from(Executors.newSingleThreadExecutor());
-        storeValueObservable
-                = createStoreValueObservable(nonNullObserveStrategy, migration, defaultValue, cacheTimeMillis);
-        valueObservable = createValueObservable(storeValueObservable, nonNullObserveStrategy, cacheTimeMillis);
-    }
-
-    @Nullable
-    private TStoreObject returnDefaultValueIfNull(@Nullable final TStoreObject storeObject, @Nullable final TObject defaultValue) {
-        if (storeObject != null || defaultValue == null) {
-            return storeObject;
-        }
-
-        try {
-            return converter.toStoreObject(objectType, storeObjectType, defaultValue);
-        } catch (final Converter.ConversionException exception) {
-            STORABLE_LC_GROUP.w(exception, "Exception while converting default value of '%s' from '%s' from store %s",
-                    key, defaultValue, store);
-            throw OnErrorThrowable.from(exception);
-        }
+        super(builderCore);
     }
 
     @NonNull
-    private Observable<TStoreObject> createStoreInitialLoadingObservable(@Nullable final Migration<TKey> migration) {
-        final Single<TStoreObject> loadObservable = store.loadObject(storeObjectType, key)
-                .doOnError(throwable -> STORABLE_LC_GROUP.w(throwable, "Exception while trying to load value of '%s' from store %s", key, store));
-        return (migration != null ? migration.migrateToLatestVersion(key).andThen(loadObservable) : loadObservable)
-                .subscribeOn(scheduler)
-                .observeOn(scheduler)
-                .toObservable()
-                .replay(1)
-                .refCount()
-                .take(1);
-    }
-
-    @NonNull
-    private Observable<TStoreObject> createStoreValueObservable(@NonNull final ObserveStrategy observeStrategy,
-                                                                @Nullable final Migration<TKey> migration,
-                                                                @Nullable final TObject defaultValue,
-                                                                final long cacheTimeMillis) {
-        final Observable<TStoreObject> storeInitialLoadingObservable = createStoreInitialLoadingObservable(migration);
-        final Observable<TStoreObject> result = storeInitialLoadingObservable
-                .concatWith(newStoreValueEvent)
-                .map(storeObject -> returnDefaultValueIfNull(storeObject, defaultValue));
-        return observeStrategy == ObserveStrategy.CACHE_STORE_VALUE || observeStrategy == ObserveStrategy.CACHE_STORE_AND_ACTUAL_VALUE
-                ? Observable.create(new OnSubscribeRefCountWithCacheTime<>(result.replay(1), cacheTimeMillis, TimeUnit.MILLISECONDS))
-                : result;
-    }
-
-    @NonNull
-    private Observable<TObject> createValueObservable(@NonNull final Observable<TStoreObject> storeValueObservable,
-                                                      @NonNull final ObserveStrategy observeStrategy,
-                                                      final long cacheTimeMillis) {
-        final Observable<TObject> result = storeValueObservable
-                .map(storeObject -> {
-                    try {
-                        return converter.toObject(objectType, storeObjectType, storeObject);
-                    } catch (final Converter.ConversionException exception) {
-                        STORABLE_LC_GROUP.w(exception, "Exception while trying to converting value of '%s' from store %s by %s",
-                                key, storeObject, store, converter);
-                        throw OnErrorThrowable.from(exception);
-                    }
-                });
-        return observeStrategy == ObserveStrategy.CACHE_ACTUAL_VALUE || observeStrategy == ObserveStrategy.CACHE_STORE_AND_ACTUAL_VALUE
-                ? Observable.create(new OnSubscribeRefCountWithCacheTime<>(result.replay(1), cacheTimeMillis, TimeUnit.MILLISECONDS))
-                : result;
-    }
-
-    /**
-     * Returns key of value.
-     *
-     * @return Unique key.
-     */
-    @NonNull
-    public TKey getKey() {
-        return key;
-    }
-
-    /**
-     * Returns type of actual object.
-     *
-     * @return Type of actual object.
-     */
-    @NonNull
-    public Type getObjectType() {
-        return objectType;
-    }
-
-    /**
-     * Returns type of store object.
-     *
-     * @return Type of store object.
-     */
-    @NonNull
-    public Type getStoreObjectType() {
-        return storeObjectType;
-    }
-
-    /**
-     * Returns {@link Store} where store class representation of object is storing.
-     *
-     * @return Store.
-     */
-    @NonNull
-    public Store<TKey, TStoreObject> getStore() {
-        return store;
-    }
-
-    /**
-     * Returns {@link Converter} to convert values from store class to actual and back.
-     *
-     * @return Converter.
-     */
-    @NonNull
-    public Converter<TObject, TStoreObject> getConverter() {
-        return converter;
-    }
-
-    @NonNull
-    private Completable internalSet(@Nullable final TObject newValue, final boolean checkForEqualityBeforeSet) {
-        return (checkForEqualityBeforeSet ? storeValueObservable.take(1) : Observable.just(null))
-                .observeOn(scheduler)
-                .switchMap(oldStoreValue -> {
-                    final TStoreObject newStoreValue;
-                    try {
-                        newStoreValue = converter.toStoreObject(objectType, storeObjectType, newValue);
-                    } catch (final Converter.ConversionException exception) {
-                        STORABLE_LC_GROUP.w(exception, "Exception while trying to store value of '%s' from store %s by %s",
-                                key, newValue, store, converter);
-                        return Observable.error(exception);
-                    }
-                    if (checkForEqualityBeforeSet && ObjectUtils.equals(newStoreValue, oldStoreValue)) {
-                        return Observable.empty();
-                    }
-                    return store.storeObject(storeObjectType, key, newStoreValue)
-                            .doOnError(throwable -> STORABLE_LC_GROUP.w(throwable,
-                                    "Exception while trying to store value of '%s' from store %s by %s",
-                                    key, newValue, store, converter))
-                            .observeOn(scheduler)
-                            .andThen(Completable.fromAction(() -> {
-                                newStoreValueEvent.onNext(newStoreValue);
-                                if (checkForEqualityBeforeSet) {
-                                    STORABLE_LC_GROUP.i("Value of '%s' changed from '%s' to '%s'", key, oldStoreValue, newStoreValue);
-                                } else {
-                                    STORABLE_LC_GROUP.i("Value of '%s' force changed to '%s'", key, newStoreValue);
-                                }
-                            }))
-                            .toObservable();
-                })
-                .toCompletable();
-    }
-
-    /**
-     * Creates observable which is async setting value to store.
-     * It is not checking if stored value equals new value.
-     * In result it will be faster to not get value from store and compare but it will emit item to {@link #observe()} subscribers.
-     * NOTE: It could emit ONLY completed and errors events. It is not providing onNext event! //TODO: it's Completable :(
-     *
-     * @param newValue Value to set;
-     * @return Observable of setting process.
-     */
-    @NonNull
-    public Observable<?> forceSet(@Nullable final TObject newValue) {
-        return internalSet(newValue, false).toObservable();
-    }
-
-    /**
-     * Creates observable which is async setting value to store.
-     * It is checking if stored value equals new value.
-     * In result it will take time to get value from store and compare
-     * but it won't emit item to {@link #observe()} subscribers if stored value equals new value.
-     * NOTE: It could emit ONLY completed and errors events. It is not providing onNext event! //TODO: it's Completable :(
-     *
-     * @param newValue Value to set;
-     * @return Observable of setting process.
-     */
-    @NonNull
-    public Observable<?> set(@Nullable final TObject newValue) {
-        return internalSet(newValue, true).toObservable();
-    }
-
-    public void setCalm(@Nullable final TObject value) {
-        set(value).subscribe(Actions.empty(), Lc::assertion);
-    }
-
-    /**
-     * Sets value synchronously. You should NOT use this method normally. Use {@link #set(Object)} asynchronously instead.
-     *
-     * @param newValue Value to set;
-     */
-    //deprecation: it should be used for debug only and in very rare cases.
-    public void setSync(@Nullable final TObject newValue) {
-        set(newValue).toBlocking().subscribe();
-    }
-
-    /**
-     * Returns Observable which is emitting item on subscribe and every time when someone have changed value.
-     * It could emit next and error events but not completed.
-     *
-     * @return Returns observable of value.
-     */
-    @NonNull
+    @Override
     public Observable<TObject> observe() {
-        return valueObservable;
-    }
-
-    /**
-     * Returns Observable which is emitting only one item on subscribe. //TODO: it's Single :(
-     * It could emit next and error events but not completed.
-     *
-     * @return Returns observable of value.
-     */
-    @NonNull
-    public Observable<TObject> get() {
-        return valueObservable.take(1);
-    }
-
-    /**
-     * Gets value synchronously. You should NOT use this method normally. Use {@link #get()} or {@link #observe()} asynchronously instead.
-     *
-     * @return Returns value;
-     */
-    //deprecation: it should be used for debug only and in very rare cases.
-    @Nullable
-    public TObject getSync() {
-        return get().toBlocking().first();
-    }
-
-    /**
-     * Enum that is representing strategy of observing item from store.
-     */
-    public enum ObserveStrategy {
-
-        /**
-         * Not caching value so on every {@link #get()} emit it will get value from {@link #getStore()} and converts it with {@link #getConverter()}.
-         */
-        NO_CACHE,
-        /**
-         * Caching only store value so on every {@link #get()} emit it will converts it with {@link #getConverter()}.
-         * Do not use such strategy if store object could be big (like byte-array of file).
-         */
-        CACHE_STORE_VALUE,
-        /**
-         * Caching value so it won't spend time for getting value from {@link #getStore()} and converts it by {@link #getConverter()}.
-         * But it will take time for getting value from {@link #getStore()} to set value.
-         * Do not use such strategy if object could be big (like Bitmap or long string).
-         * Do not use such strategy if object is mutable because multiple subscribers could then change it's state.
-         */
-        CACHE_ACTUAL_VALUE,
-        /**
-         * Caching value so it won't spend time for getting value from {@link #getStore()} and converts it by {@link #getConverter()}.
-         * It won't take time or getting value from {@link #getStore()} to set value.
-         * Do not use such strategy if store object could be big (like byte-array of file).
-         * Do not use such strategy if object could be big (like Bitmap or long string).
-         * Do not use such strategy if object is mutable because multiple subscribers could then change it's state.
-         */
-        CACHE_STORE_AND_ACTUAL_VALUE
-
-    }
-
-    /**
-     * Helper class to create various builders.
-     *
-     * @param <TKey>         Type of key to identify object;
-     * @param <TObject>      Type of actual object;
-     * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
-     */
-    public static class BuilderCore<TKey, TObject, TStoreObject> {
-
-        @NonNull
-        protected final TKey key;
-        @NonNull
-        protected final Type objectType;
-        @NonNull
-        private final Type storeObjectType;
-        @NonNull
-        private final Store<TKey, TStoreObject> store;
-        @NonNull
-        private final Converter<TObject, TStoreObject> converter;
-        @Nullable
-        private ObserveStrategy observeStrategy;
-        @Nullable
-        private Migration<TKey> migration;
-        @Nullable
-        private TObject defaultValue;
-        @Nullable
-        private Scheduler storeScheduler;
-        private long cacheTimeMillis;
-
-        protected BuilderCore(@NonNull final TKey key,
-                              @NonNull final Type objectType,
-                              @NonNull final Type storeObjectType,
-                              @NonNull final Store<TKey, TStoreObject> store,
-                              @NonNull final Converter<TObject, TStoreObject> converter) {
-            this(key, objectType, storeObjectType, store, converter, null, null, null, null, DEFAULT_CACHE_TIME_MILLIS);
-        }
-
-        protected BuilderCore(@NonNull final BuilderCore<TKey, TObject, TStoreObject> sourceBuilder) {
-            this(sourceBuilder.key, sourceBuilder.objectType, sourceBuilder.storeObjectType,
-                    sourceBuilder.store, sourceBuilder.converter, sourceBuilder.observeStrategy,
-                    sourceBuilder.migration, sourceBuilder.defaultValue, sourceBuilder.storeScheduler, sourceBuilder.cacheTimeMillis);
-        }
-
-        @SuppressWarnings({"PMD.ExcessiveParameterList", "CPD-START"})
-        //CPD: it is same code as constructor of Storable
-        //ExcessiveParameterList: that's why we are using builder to create it
-        private BuilderCore(@NonNull final TKey key,
-                            @NonNull final Type objectType,
-                            @NonNull final Type storeObjectType,
-                            @NonNull final Store<TKey, TStoreObject> store,
-                            @NonNull final Converter<TObject, TStoreObject> converter,
-                            @Nullable final ObserveStrategy observeStrategy,
-                            @Nullable final Migration<TKey> migration,
-                            @Nullable final TObject defaultValue,
-                            @Nullable final Scheduler storeScheduler,
-                            final long cacheTimeMillis) {
-            this.key = key;
-            this.objectType = objectType;
-            this.storeObjectType = storeObjectType;
-            this.store = store;
-            this.converter = converter;
-            this.observeStrategy = observeStrategy;
-            this.migration = migration;
-            this.defaultValue = defaultValue;
-            this.storeScheduler = storeScheduler;
-            this.cacheTimeMillis = cacheTimeMillis;
-        }
-
-        @SuppressWarnings("CPD-END")
-        protected void setStoreSchedulerInternal(@Nullable final Scheduler storeScheduler) {
-            this.storeScheduler = storeScheduler;
-        }
-
-        protected void setObserveStrategyInternal(@Nullable final ObserveStrategy observeStrategy) {
-            this.observeStrategy = observeStrategy;
-        }
-
-        protected void setMigrationInternal(@NonNull final Migration<TKey> migration) {
-            this.migration = migration;
-        }
-
-        protected void setCacheTimeInternal(final long cacheTime, @NonNull final TimeUnit timeUnit) {
-            this.cacheTimeMillis = timeUnit.toMillis(cacheTime);
-        }
-
-        @Nullable
-        protected TObject getDefaultValue() {
-            return defaultValue;
-        }
-
-        protected void setDefaultValueInternal(@NonNull final TObject defaultValue) {
-            this.defaultValue = defaultValue;
-        }
-
+        return observeOptionalValue().map(Optional::get);
     }
 
     /**
@@ -544,8 +130,8 @@ public class Storable<TKey, TObject, TStoreObject> {
          * @return Builder that allows to specify other fields.
          */
         @NonNull
-        public NonNullStorableBuilder<TKey, TObject, TStoreObject> setDefaultValue(@NonNull final TObject defaultValue) {
-            return new NonNullStorableBuilder<>(this, defaultValue);
+        public NonNullStorable.Builder<TKey, TObject, TStoreObject> setDefaultValue(@NonNull final TObject defaultValue) {
+            return new NonNullStorable.Builder<>(this, defaultValue);
         }
 
         /**

src/main/java/ru/touchin/roboswag/core/observables/storable/Store.java

@@ -24,6 +24,7 @@ import android.support.annotation.Nullable;
 
 import java.lang.reflect.Type;
 
+import ru.touchin.roboswag.core.utils.Optional;
 import rx.Completable;
 import rx.Single;
 
@@ -64,6 +65,6 @@ public interface Store<TKey, TStoreObject> {
      * @return Object from store found by key;
      */
     @NonNull
-    Single<TStoreObject> loadObject(@NonNull Type storeObjectType, @NonNull TKey key);
+    Single<Optional<TStoreObject>> loadObject(@NonNull Type storeObjectType, @NonNull TKey key);
 
 }

src/main/java/ru/touchin/roboswag/core/observables/storable/builders/NonNullStorableBuilder.java

@@ -1,113 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.observables.storable.builders;
-
-import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
-
-import java.util.concurrent.TimeUnit;
-
-import ru.touchin.roboswag.core.observables.storable.Migration;
-import ru.touchin.roboswag.core.observables.storable.Storable;
-import ru.touchin.roboswag.core.observables.storable.concrete.NonNullStorable;
-import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
-import rx.Scheduler;
-
-/**
- * Created by Gavriil Sitnikov on 15/05/2016.
- * Builder that is already contains not null default value.
- *
- * @param <TKey>         Type of key to identify object;
- * @param <TObject>      Type of actual object;
- * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
- */
-public class NonNullStorableBuilder<TKey, TObject, TStoreObject> extends Storable.BuilderCore<TKey, TObject, TStoreObject> {
-
-    public NonNullStorableBuilder(@NonNull final Storable.Builder<TKey, TObject, TStoreObject> sourceBuilder,
-                                  @NonNull final TObject defaultValue) {
-        super(sourceBuilder);
-        setDefaultValueInternal(defaultValue);
-    }
-
-    /**
-     * Sets specific {@link Scheduler} to store/load/convert values on it.
-     *
-     * @param storeScheduler Scheduler;
-     * @return Builder that allows to specify other fields.
-     */
-    @NonNull
-    public NonNullStorableBuilder<TKey, TObject, TStoreObject> setStoreScheduler(@Nullable final Scheduler storeScheduler) {
-        setStoreSchedulerInternal(storeScheduler);
-        return this;
-    }
-
-    /**
-     * Sets specific {@link Storable.ObserveStrategy} to cache value in memory in specific way.
-     *
-     * @param observeStrategy ObserveStrategy;
-     * @return Builder that allows to specify other fields.
-     */
-    @NonNull
-    public NonNullStorableBuilder<TKey, TObject, TStoreObject> setObserveStrategy(@Nullable final Storable.ObserveStrategy observeStrategy) {
-        setObserveStrategyInternal(observeStrategy);
-        return this;
-    }
-
-    /**
-     * Sets cache time for while value that cached by {@link #setObserveStrategy(Storable.ObserveStrategy)}
-     * will be in memory after everyone unsubscribe.
-     * It is important for example for cases when user switches between screens and hide/open app very fast.
-     *
-     * @param cacheTime Cache time value;
-     * @param timeUnit  Cache time units.
-     * @return Builder that allows to specify other fields.
-     */
-    @NonNull
-    public NonNullStorableBuilder<TKey, TObject, TStoreObject> setCacheTime(final long cacheTime, @NonNull final TimeUnit timeUnit) {
-        setCacheTimeInternal(cacheTime, timeUnit);
-        return this;
-    }
-
-    /**
-     * Sets specific {@link Migration} to migrate values from specific version to latest version.
-     *
-     * @param migration Migration;
-     * @return Builder that allows to specify other fields.
-     */
-    @NonNull
-    public NonNullStorableBuilder<TKey, TObject, TStoreObject> setMigration(@NonNull final Migration<TKey> migration) {
-        setMigrationInternal(migration);
-        return this;
-    }
-
-    /**
-     * Building {@link NonNullStorable} object.
-     *
-     * @return New {@link NonNullStorable}.
-     */
-    @NonNull
-    public NonNullStorable<TKey, TObject, TStoreObject> build() {
-        if (getDefaultValue() == null) {
-            throw new ShouldNotHappenException();
-        }
-        return new NonNullStorable<>(this);
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/observables/storable/concrete/NonNullStorable.java

@@ -20,10 +20,16 @@
 package ru.touchin.roboswag.core.observables.storable.concrete;
 
 import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
 
+import java.util.concurrent.TimeUnit;
+
+import ru.touchin.roboswag.core.observables.storable.BaseStorable;
+import ru.touchin.roboswag.core.observables.storable.Migration;
 import ru.touchin.roboswag.core.observables.storable.Storable;
-import ru.touchin.roboswag.core.observables.storable.builders.NonNullStorableBuilder;
 import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
+import rx.Observable;
+import rx.Scheduler;
 
 /**
  * Created by Gavriil Sitnikov on 04/10/2015.
@@ -34,20 +40,109 @@ import ru.touchin.roboswag.core.utils.ShouldNotHappenException;
  * @param <TObject>      Type of actual object;
  * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
  */
-public class NonNullStorable<TKey, TObject, TStoreObject> extends Storable<TKey, TObject, TStoreObject> {
+public class NonNullStorable<TKey, TObject, TStoreObject> extends BaseStorable<TKey, TObject, TStoreObject, TObject> {
 
-    public NonNullStorable(@NonNull final NonNullStorableBuilder<TKey, TObject, TStoreObject> builderCore) {
+    public NonNullStorable(@NonNull final Builder<TKey, TObject, TStoreObject> builderCore) {
         super(builderCore);
     }
 
     @NonNull
     @Override
-    public TObject getSync() {
-        final TObject result = super.getSync();
-        if (result == null) {
-            throw new ShouldNotHappenException();
-        }
-        return result;
+    public Observable<TObject> observe() {
+        return observeOptionalValue()
+                .map(optional -> {
+                    if (optional.get() == null) {
+                        throw new ShouldNotHappenException();
+                    }
+                    return optional.get();
+                });
     }
 
+    /**
+     * Created by Gavriil Sitnikov on 15/05/2016.
+     * Builder that is already contains not null default value.
+     *
+     * @param <TKey>         Type of key to identify object;
+     * @param <TObject>      Type of actual object;
+     * @param <TStoreObject> Type of store object. Could be same as {@link TObject}.
+     */
+    @SuppressWarnings("CPD-START")
+    //CPD: it is same code as Builder of Storable because it's methods returning this and can't be inherited
+    public static class Builder<TKey, TObject, TStoreObject> extends BuilderCore<TKey, TObject, TStoreObject> {
+
+        public Builder(@NonNull final Storable.Builder<TKey, TObject, TStoreObject> sourceBuilder,
+                       @NonNull final TObject defaultValue) {
+            super(sourceBuilder);
+            if (defaultValue == null) {
+                throw new ShouldNotHappenException();
+            }
+            setDefaultValueInternal(defaultValue);
+        }
+
+        /**
+         * Sets specific {@link Scheduler} to store/load/convert values on it.
+         *
+         * @param storeScheduler Scheduler;
+         * @return Builder that allows to specify other fields.
+         */
+        @NonNull
+        public Builder<TKey, TObject, TStoreObject> setStoreScheduler(@Nullable final Scheduler storeScheduler) {
+            setStoreSchedulerInternal(storeScheduler);
+            return this;
+        }
+
+        /**
+         * Sets specific {@link ObserveStrategy} to cache value in memory in specific way.
+         *
+         * @param observeStrategy ObserveStrategy;
+         * @return Builder that allows to specify other fields.
+         */
+        @NonNull
+        public Builder<TKey, TObject, TStoreObject> setObserveStrategy(@Nullable final ObserveStrategy observeStrategy) {
+            setObserveStrategyInternal(observeStrategy);
+            return this;
+        }
+
+        /**
+         * Sets cache time for while value that cached by {@link #setObserveStrategy(ObserveStrategy)}
+         * will be in memory after everyone unsubscribe.
+         * It is important for example for cases when user switches between screens and hide/open app very fast.
+         *
+         * @param cacheTime Cache time value;
+         * @param timeUnit  Cache time units.
+         * @return Builder that allows to specify other fields.
+         */
+        @NonNull
+        public Builder<TKey, TObject, TStoreObject> setCacheTime(final long cacheTime, @NonNull final TimeUnit timeUnit) {
+            setCacheTimeInternal(cacheTime, timeUnit);
+            return this;
+        }
+
+        /**
+         * Sets specific {@link Migration} to migrate values from specific version to latest version.
+         *
+         * @param migration Migration;
+         * @return Builder that allows to specify other fields.
+         */
+        @NonNull
+        public Builder<TKey, TObject, TStoreObject> setMigration(@NonNull final Migration<TKey> migration) {
+            setMigrationInternal(migration);
+            return this;
+        }
+
+        /**
+         * Building {@link NonNullStorable} object.
+         *
+         * @return New {@link NonNullStorable}.
+         */
+        @NonNull
+        @SuppressWarnings("CPD-END")
+        public NonNullStorable<TKey, TObject, TStoreObject> build() {
+            if (getDefaultValue() == null) {
+                throw new ShouldNotHappenException();
+            }
+            return new NonNullStorable<>(this);
+        }
+
+    }
 }

src/main/java/ru/touchin/roboswag/core/utils/Optional.java

@@ -0,0 +1,71 @@
+/*
+ *  Copyright (c) 2017 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
+ *
+ *  This file is part of RoboSwag library.
+ *
+ *  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
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package ru.touchin.roboswag.core.utils;
+
+import android.support.annotation.Nullable;
+
+import java.io.Serializable;
+
+/**
+ * Created by Gavriil Sitnikov on 16/04/2017.
+ * Holds nullable objects inside. It is needed to implement RxJava2 non-null emitting logic.
+ *
+ * @param <T> Type of object.
+ */
+public class Optional<T> implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @Nullable
+    private final T value;
+
+    public Optional(@Nullable final T value) {
+        this.value = value;
+    }
+
+    /**
+     * Returns holding nullable object.
+     *
+     * @return Holding object.
+     */
+    @Nullable
+    public T get() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+
+        final Optional<?> that = (Optional<?>) object;
+        return ObjectUtils.equals(value, that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return value != null ? value.hashCode() : 0;
+    }
+
+}

src/main/java/ru/touchin/roboswag/core/utils/ProcessPriorityThreadFactory.java

@@ -1,52 +0,0 @@
-/*
- *  Copyright (c) 2015 RoboSwag (Gavriil Sitnikov, Vsevolod Ivanov)
- *
- *  This file is part of RoboSwag library.
- *
- *  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
- *
- *  http://www.apache.org/licenses/LICENSE-2.0
- *
- *  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.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- */
-
-package ru.touchin.roboswag.core.utils;
-
-import android.support.annotation.NonNull;
-
-import java.util.concurrent.ThreadFactory;
-
-/**
- * Created by Ilia Kurtov on 25/05/2017 with a help from https://stackoverflow.com/a/21187003/4312184
- * ThreadFactory that can change default thread priority. Suitable for creating Schedulers fo RxJava like this:
- * final Scheduler scheduler = Schedulers.from(Executors.newSingleThreadExecutor(new ProcessPriorityThreadFactory(Thread.MIN_PRIORITY)));
- */
-public final class ProcessPriorityThreadFactory implements ThreadFactory {
-
-    private final int threadPriority;
-
-    /**
-     * threadPriority can be in a range from {@link Thread#MIN_PRIORITY} to {@link Thread#MAX_PRIORITY}
-     *
-     * @param threadPriority priority for the Thread.
-     */
-    public ProcessPriorityThreadFactory(final int threadPriority) {
-        this.threadPriority = threadPriority;
-    }
-
-    @Override
-    @NonNull
-    public Thread newThread(@NonNull final Runnable runnable) {
-        final Thread thread = new Thread(runnable);
-        thread.setPriority(threadPriority);
-        return thread;
-    }
-
-}

src/main/java/ru/touchin/roboswag/core/utils/ServiceBinder.java

@@ -22,6 +22,7 @@ package ru.touchin.roboswag.core.utils;
 import android.app.Service;
 import android.os.Binder;
 import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
 
 /**
  * Created by Gavriil Sitnikov on 03/10/2015.
@@ -47,4 +48,23 @@ public class ServiceBinder<TService extends Service> extends Binder {
         return service;
     }
 
+    @Override
+    public boolean equals(@Nullable final Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+
+        final ServiceBinder that = (ServiceBinder) object;
+
+        return ObjectUtils.equals(service, that.service);
+    }
+
+    @Override
+    public int hashCode() {
+        return service.hashCode();
+    }
+
 }

src/main/java/ru/touchin/roboswag/core/utils/ThreadLocalValue.java

@@ -21,8 +21,6 @@ package ru.touchin.roboswag.core.utils;
 
 import android.support.annotation.NonNull;
 
-import rx.functions.Func0;
-
 /**
  * Created by Gavriil Sitnikov on 13/11/2015.
  * Thread local value with specified creator of value per thread.
@@ -30,24 +28,33 @@ import rx.functions.Func0;
 public class ThreadLocalValue<T> extends ThreadLocal<T> {
 
     @NonNull
-    private final Func0<T> creator;
+    private final Fabric<T> fabric;
 
-    public ThreadLocalValue(@NonNull final NonNullFunc<T> creator) {
+    public ThreadLocalValue(@NonNull final Fabric<T> fabric) {
         super();
-        this.creator = creator;
+        this.fabric = fabric;
     }
 
     @NonNull
     @Override
     protected T initialValue() {
-        return creator.call();
+        return fabric.create();
     }
 
-    public interface NonNullFunc<T> extends Func0<T> {
+    /**
+     * Fabric of thread-local objects.
+     *
+     * @param <T> Type of objects.
+     */
+    public interface Fabric<T> {
 
+        /**
+         * Creates object.
+         *
+         * @return new instance of object.
+         */
         @NonNull
-        @Override
-        T call();
+        T create();
 
     }
 

src/main/java/ru/touchin/roboswag/core/utils/pairs/HalfNullablePair.java

@@ -81,9 +81,7 @@ public class HalfNullablePair<TFirst, TSecond> implements Serializable {
 
     @Override
     public int hashCode() {
-        int result = first.hashCode();
-        result = 31 * result + (second != null ? second.hashCode() : 0);
-        return result;
+        return ObjectUtils.hashCode(first, second);
     }
 
 }

src/main/java/ru/touchin/roboswag/core/utils/pairs/NonNullPair.java

@@ -32,7 +32,7 @@ import ru.touchin.roboswag.core.utils.ObjectUtils;
  * Both arguments are not null.
  * Note that if you want to save this pair in state, you need make TFirst and TSecond Serializable too.
  *
- * @param <TFirst> type of the first nonnull argument.
+ * @param <TFirst>  type of the first nonnull argument.
  * @param <TSecond> type of the second nonnull argument.
  */
 public class NonNullPair<TFirst, TSecond> implements Serializable {
@@ -81,9 +81,7 @@ public class NonNullPair<TFirst, TSecond> implements Serializable {
 
     @Override
     public int hashCode() {
-        int result = first.hashCode();
-        result = 31 * result + second.hashCode();
-        return result;
+        return ObjectUtils.hashCode(first, second);
     }
 
 }

src/main/java/ru/touchin/roboswag/core/utils/pairs/NullablePair.java

@@ -85,9 +85,7 @@ public class NullablePair<TFirst, TSecond> implements Serializable { //todo: mb
 
     @Override
     public int hashCode() {
-        int result = first.hashCode();
-        result = 31 * result + second.hashCode();
-        return result;
+        return ObjectUtils.hashCode(first, second);
     }
 
 }