diff --git a/paging/common/src/main/java/android/arch/paging/ContiguousPagedList.java b/paging/common/src/main/java/android/arch/paging/ContiguousPagedList.java
index 2a5cd42..cdff391 100644
--- a/paging/common/src/main/java/android/arch/paging/ContiguousPagedList.java
+++ b/paging/common/src/main/java/android/arch/paging/ContiguousPagedList.java
@@ -47,7 +47,9 @@
             });
         }
 
-        @MainThread
+        // Creation thread for initial synchronous load, otherwise main thread
+        // Safe to access main thread only state - no other thread has reference during construction
+        @AnyThread
         @Override
         public void onPageResult(@NonNull PageResult<K, V> pageResult) {
             if (pageResult.page == null) {
@@ -70,6 +72,17 @@
             } else if (pageResult.type == PageResult.PREPEND) {
                 mKeyedStorage.prependPage(page, ContiguousPagedList.this);
             }
+
+            if (mBoundaryCallback != null) {
+                boolean deferEmpty = mStorage.size() == 0;
+                boolean deferBegin = !deferEmpty
+                        && pageResult.type == PageResult.PREPEND
+                        && pageResult.page.items.size() == 0;
+                boolean deferEnd = !deferEmpty
+                        && pageResult.type == PageResult.APPEND
+                        && pageResult.page.items.size() == 0;
+                deferBoundaryCallbacks(deferEmpty, deferBegin, deferEnd);
+            }
         }
     };
 
@@ -77,9 +90,11 @@
             @NonNull ContiguousDataSource<K, V> dataSource,
             @NonNull Executor mainThreadExecutor,
             @NonNull Executor backgroundThreadExecutor,
+            @Nullable BoundaryCallback<V> boundaryCallback,
             @NonNull Config config,
             final @Nullable K key) {
-        super(new PagedStorage<K, V>(), mainThreadExecutor, backgroundThreadExecutor, config);
+        super(new PagedStorage<K, V>(), mainThreadExecutor, backgroundThreadExecutor,
+                boundaryCallback, config);
         mDataSource = dataSource;
 
         // blocking init just triggers the initial load on the construction thread -
@@ -168,7 +183,7 @@
         final int position = mStorage.getLeadingNullCount() + mStorage.getPositionOffset();
 
         // safe to access first item here - mStorage can't be empty if we're prepending
-        final V item = mStorage.getFirstContiguousItem();
+        final V item = mStorage.getFirstLoadedItem();
         mBackgroundThreadExecutor.execute(new Runnable() {
             @Override
             public void run() {
@@ -191,7 +206,7 @@
                 + mStorage.getStorageCount() - 1 + mStorage.getPositionOffset();
 
         // safe to access first item here - mStorage can't be empty if we're appending
-        final V item = mStorage.getLastContiguousItem();
+        final V item = mStorage.getLastLoadedItem();
         mBackgroundThreadExecutor.execute(new Runnable() {
             @Override
             public void run() {
@@ -234,6 +249,8 @@
         // finally dispatch callbacks, after prepend may have already been scheduled
         notifyChanged(leadingNulls, changedCount);
         notifyInserted(0, addedCount);
+
+        offsetBoundaryAccessIndices(addedCount);
     }
 
     @MainThread
diff --git a/paging/common/src/main/java/android/arch/paging/DataSource.java b/paging/common/src/main/java/android/arch/paging/DataSource.java
index 524e570..ff44521 100644
--- a/paging/common/src/main/java/android/arch/paging/DataSource.java
+++ b/paging/common/src/main/java/android/arch/paging/DataSource.java
@@ -48,6 +48,10 @@
 @SuppressWarnings("unused") // suppress warning to remove Key/Value, needed for subclass type safety
 public abstract class DataSource<Key, Value> {
 
+    public interface Factory<Key, Value> {
+        DataSource<Key, Value> create();
+    }
+
     // Since we currently rely on implementation details of two implementations,
     // prevent external subclassing, except through exposed subclasses
     DataSource() {
diff --git a/paging/common/src/main/java/android/arch/paging/ListDataSource.java b/paging/common/src/main/java/android/arch/paging/ListDataSource.java
new file mode 100644
index 0000000..d3a171e
--- /dev/null
+++ b/paging/common/src/main/java/android/arch/paging/ListDataSource.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2017 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 android.arch.paging;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ListDataSource<T> extends TiledDataSource<T> {
+    private final List<T> mList;
+
+    public ListDataSource(List<T> list) {
+        mList = new ArrayList<>(list);
+    }
+
+    @Override
+    public int countItems() {
+        return mList.size();
+    }
+
+    @Override
+    public List<T> loadRange(int startPosition, int count) {
+        int endExclusive = Math.min(mList.size(), startPosition + count);
+        return mList.subList(startPosition, endExclusive);
+    }
+}
diff --git a/paging/common/src/main/java/android/arch/paging/PagedList.java b/paging/common/src/main/java/android/arch/paging/PagedList.java
index 51f524a..f18e108 100644
--- a/paging/common/src/main/java/android/arch/paging/PagedList.java
+++ b/paging/common/src/main/java/android/arch/paging/PagedList.java
@@ -16,8 +16,10 @@
 
 package android.arch.paging;
 
+import android.support.annotation.AnyThread;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
+import android.support.annotation.RestrictTo;
 import android.support.annotation.WorkerThread;
 
 import java.lang.ref.WeakReference;
@@ -97,6 +99,8 @@
     final Executor mMainThreadExecutor;
     @NonNull
     final Executor mBackgroundThreadExecutor;
+    @Nullable
+    final BoundaryCallback<T> mBoundaryCallback;
     @NonNull
     final Config mConfig;
     @NonNull
@@ -105,6 +109,16 @@
     int mLastLoad = 0;
     T mLastItem = null;
 
+    // if set to true, mBoundaryCallback is non-null, and should
+    // be dispatched when nearby load has occurred
+    private boolean mBoundaryCallbackBeginDeferred = false;
+    private boolean mBoundaryCallbackEndDeferred = false;
+
+    // lowest and highest index accessed by loadAround. Used to
+    // decide when mBoundaryCallback should be dispatched
+    private int mLowestIndexAccessed = Integer.MAX_VALUE;
+    private int mHighestIndexAccessed = Integer.MIN_VALUE;
+
     private final AtomicBoolean mDetached = new AtomicBoolean(false);
 
     protected final ArrayList<WeakReference<Callback>> mCallbacks = new ArrayList<>();
@@ -112,10 +126,12 @@
     PagedList(@NonNull PagedStorage<?, T> storage,
             @NonNull Executor mainThreadExecutor,
             @NonNull Executor backgroundThreadExecutor,
+            @Nullable BoundaryCallback<T> boundaryCallback,
             @NonNull Config config) {
         mStorage = storage;
         mMainThreadExecutor = mainThreadExecutor;
         mBackgroundThreadExecutor = backgroundThreadExecutor;
+        mBoundaryCallback = boundaryCallback;
         mConfig = config;
     }
 
@@ -129,6 +145,7 @@
      *                           Generally, this is the UI/main thread.
      * @param backgroundThreadExecutor Data loading will be done via this executor - should be a
      *                                 background thread.
+     * @param boundaryCallback Optional boundary callback to attach to the list.
      * @param config PagedList Config, which defines how the PagedList will load data.
      * @param <K> Key type that indicates to the DataSource what data to load.
      * @param <T> Type of items to be held and loaded by the PagedList.
@@ -139,6 +156,7 @@
     private static <K, T> PagedList<T> create(@NonNull DataSource<K, T> dataSource,
             @NonNull Executor mainThreadExecutor,
             @NonNull Executor backgroundThreadExecutor,
+            @Nullable BoundaryCallback<T> boundaryCallback,
             @NonNull Config config,
             @Nullable K key) {
         if (dataSource.isContiguous() || !config.enablePlaceholders) {
@@ -150,12 +168,14 @@
             return new ContiguousPagedList<>(contigDataSource,
                     mainThreadExecutor,
                     backgroundThreadExecutor,
+                    boundaryCallback,
                     config,
                     key);
         } else {
             return new TiledPagedList<>((TiledDataSource<T>) dataSource,
                     mainThreadExecutor,
                     backgroundThreadExecutor,
+                    boundaryCallback,
                     config,
                     (key != null) ? (Integer) key : 0);
         }
@@ -186,6 +206,7 @@
         private DataSource<Key, Value> mDataSource;
         private Executor mMainThreadExecutor;
         private Executor mBackgroundThreadExecutor;
+        private BoundaryCallback mBoundaryCallback;
         private Config mConfig;
         private Key mInitialKey;
 
@@ -229,6 +250,14 @@
             return this;
         }
 
+        @NonNull
+        public Builder<Key, Value> setBoundaryCallback(
+                @Nullable BoundaryCallback boundaryCallback) {
+            mBoundaryCallback = boundaryCallback;
+            return this;
+        }
+
+
         /**
          * The Config defining how the PagedList should load from the DataSource.
          *
@@ -284,10 +313,12 @@
                 throw new IllegalArgumentException("Config required");
             }
 
+            //noinspection unchecked
             return PagedList.create(
                     mDataSource,
                     mMainThreadExecutor,
                     mBackgroundThreadExecutor,
+                    mBoundaryCallback,
                     mConfig,
                     mInitialKey);
         }
@@ -312,7 +343,6 @@
         return item;
     }
 
-
     /**
      * Load adjacent items to passed index.
      *
@@ -321,8 +351,122 @@
     public void loadAround(int index) {
         mLastLoad = index + getPositionOffset();
         loadAroundInternal(index);
+
+        mLowestIndexAccessed = Math.min(mLowestIndexAccessed, index);
+        mHighestIndexAccessed = Math.max(mHighestIndexAccessed, index);
+
+        /*
+         * mLowestIndexAccessed / mHighestIndexAccessed have been updated, so check if we need to
+         * dispatch boundary callbacks. Boundary callbacks are deferred until last items are loaded,
+         * and accesses happen near the boundaries.
+         *
+         * Note: we post here, since RecyclerView may want to add items in response, and this
+         * call occurs in PagedListAdapter bind.
+         */
+        tryDispatchBoundaryCallbacks(true);
     }
 
+    // Creation thread for initial synchronous load, otherwise main thread
+    // Safe to access main thread only state - no other thread has reference during construction
+    @AnyThread
+    void deferBoundaryCallbacks(final boolean deferEmpty,
+            final boolean deferBegin, final boolean deferEnd) {
+        if (mBoundaryCallback == null) {
+            throw new IllegalStateException("Computing boundary");
+        }
+
+        /*
+         * If lowest/highest haven't been initialized, set them to storage size,
+         * since placeholders must already be computed by this point.
+         *
+         * This is just a minor optimization so that BoundaryCallback callbacks are sent immediately
+         * if the initial load size is smaller than the prefetch window (see
+         * TiledPagedListTest#boundaryCallback_immediate())
+         */
+        if (mLowestIndexAccessed == Integer.MAX_VALUE) {
+            mLowestIndexAccessed = mStorage.size();
+        }
+        if (mHighestIndexAccessed == Integer.MIN_VALUE) {
+            mHighestIndexAccessed = 0;
+        }
+
+        if (deferEmpty || deferBegin || deferEnd) {
+            // Post to the main thread, since we may be on creation thread currently
+            mMainThreadExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    // on is dispatched immediately, since items won't be accessed
+                    //noinspection ConstantConditions
+                    if (deferEmpty) {
+                        mBoundaryCallback.onZeroItemsLoaded();
+                    }
+
+                    // for other callbacks, mark deferred, and only dispatch if loadAround
+                    // has been called near to the position
+                    if (deferBegin) {
+                        mBoundaryCallbackBeginDeferred = true;
+                    }
+                    if (deferEnd) {
+                        mBoundaryCallbackEndDeferred = true;
+                    }
+                    tryDispatchBoundaryCallbacks(false);
+                }
+            });
+        }
+    }
+
+    /**
+     * Call this when mLowest/HighestIndexAccessed are changed, or
+     * mBoundaryCallbackBegin/EndDeferred is set.
+     */
+    private void tryDispatchBoundaryCallbacks(boolean post) {
+        final boolean dispatchBegin = mBoundaryCallbackBeginDeferred
+                && mLowestIndexAccessed <= mConfig.prefetchDistance;
+        final boolean dispatchEnd = mBoundaryCallbackEndDeferred
+                && mHighestIndexAccessed >= size() - mConfig.prefetchDistance;
+
+        if (!dispatchBegin && !dispatchEnd) {
+            return;
+        }
+
+        if (dispatchBegin) {
+            mBoundaryCallbackBeginDeferred = false;
+        }
+        if (dispatchEnd) {
+            mBoundaryCallbackEndDeferred = false;
+        }
+        if (post) {
+            mMainThreadExecutor.execute(new Runnable() {
+                @Override
+                public void run() {
+                    dispatchBoundaryCallbacks(dispatchBegin, dispatchEnd);
+                }
+            });
+        } else {
+            dispatchBoundaryCallbacks(dispatchBegin, dispatchEnd);
+        }
+    }
+
+    private void dispatchBoundaryCallbacks(boolean begin, boolean end) {
+        // safe to deref mBoundaryCallback here, since we only defer if mBoundaryCallback present
+        if (begin) {
+            //noinspection ConstantConditions
+            mBoundaryCallback.onItemAtFrontLoaded(
+                    snapshot(), mStorage.getFirstLoadedItem(), mStorage.size());
+        }
+        if (end) {
+            //noinspection ConstantConditions
+            mBoundaryCallback.onItemAtEndLoaded(
+                    snapshot(), mStorage.getLastLoadedItem(), mStorage.size());
+        }
+    }
+
+    /** @hide */
+    @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
+    void offsetBoundaryAccessIndices(int offset) {
+        mLowestIndexAccessed += offset;
+        mHighestIndexAccessed += offset;
+    }
 
     /**
      * Returns size of the list, including any not-yet-loaded null padding.
@@ -351,6 +495,7 @@
      *
      * @return Immutable snapshot of PagedList data.
      */
+    @SuppressWarnings("WeakerAccess")
     @NonNull
     public List<T> snapshot() {
         if (isImmutable()) {
@@ -726,4 +871,15 @@
             }
         }
     }
+
+    /**
+     * WIP API for load-more-into-local-storage callbacks
+     */
+    public abstract static class BoundaryCallback<T> {
+        public abstract void onZeroItemsLoaded();
+        public abstract void onItemAtFrontLoaded(@NonNull List<T> pagedListSnapshot,
+                @NonNull T itemAtFront, int pagedListSize);
+        public abstract void onItemAtEndLoaded(@NonNull List<T> pagedListSnapshot,
+                @NonNull T itemAtEnd, int pagedListSize);
+    }
 }
diff --git a/paging/common/src/main/java/android/arch/paging/PagedStorage.java b/paging/common/src/main/java/android/arch/paging/PagedStorage.java
index 7f91290..b857462 100644
--- a/paging/common/src/main/java/android/arch/paging/PagedStorage.java
+++ b/paging/common/src/main/java/android/arch/paging/PagedStorage.java
@@ -230,13 +230,13 @@
 
     // ---------------- Contiguous API -------------------
 
-    V getFirstContiguousItem() {
+    V getFirstLoadedItem() {
         // safe to access first page's first item here:
         // If contiguous, mPages can't be empty, can't hold null Pages, and items can't be empty
         return mPages.get(0).items.get(0);
     }
 
-    V getLastContiguousItem() {
+    V getLastLoadedItem() {
         // safe to access last page's last item here:
         // If contiguous, mPages can't be empty, can't hold null Pages, and items can't be empty
         Page<K, V> page = mPages.get(mPages.size() - 1);
diff --git a/paging/common/src/main/java/android/arch/paging/SnapshotPagedList.java b/paging/common/src/main/java/android/arch/paging/SnapshotPagedList.java
index 7e965a0..6a8a748 100644
--- a/paging/common/src/main/java/android/arch/paging/SnapshotPagedList.java
+++ b/paging/common/src/main/java/android/arch/paging/SnapshotPagedList.java
@@ -27,6 +27,7 @@
         super(pagedList.mStorage.snapshot(),
                 pagedList.mMainThreadExecutor,
                 pagedList.mBackgroundThreadExecutor,
+                null,
                 pagedList.mConfig);
         mContiguous = pagedList.isContiguous();
         mLastKey = pagedList.getLastKey();
diff --git a/paging/common/src/main/java/android/arch/paging/TiledDataSource.java b/paging/common/src/main/java/android/arch/paging/TiledDataSource.java
index 61dead3..0ea9428 100644
--- a/paging/common/src/main/java/android/arch/paging/TiledDataSource.java
+++ b/paging/common/src/main/java/android/arch/paging/TiledDataSource.java
@@ -87,6 +87,8 @@
  */
 public abstract class TiledDataSource<Type> extends DataSource<Integer, Type> {
 
+    private int mItemCount;
+
     /**
      * Number of items that this DataSource can provide in total.
      *
@@ -123,6 +125,7 @@
      */
     void loadRangeInitial(int startPosition, int count, int pageSize, int itemCount,
             PageResult.Receiver<Integer, Type> receiver) {
+        mItemCount = itemCount;
 
         if (itemCount == 0) {
             // no data to load, just immediately return empty
@@ -132,7 +135,6 @@
             return;
         }
 
-
         List<Type> list = loadRangeWrapper(startPosition, count);
 
         count = Math.min(count, itemCount - startPosition);
@@ -167,9 +169,15 @@
     void loadRange(int startPosition, int count, PageResult.Receiver<Integer, Type> receiver) {
         List<Type> list = loadRangeWrapper(startPosition, count);
 
-        Page<Integer, Type> page = list != null ? new Page<Integer, Type>(list) : null;
+        Page<Integer, Type> page = null;
+        int trailingNulls = mItemCount - startPosition;
+
+        if (list != null) {
+            page = new Page<Integer, Type>(list);
+            trailingNulls -= list.size();
+        }
         receiver.postOnPageResult(new PageResult<>(
-                PageResult.TILE, page, startPosition, 0, 0));
+                PageResult.TILE, page, startPosition, trailingNulls, 0));
     }
 
     private List<Type> loadRangeWrapper(int startPosition, int count) {
diff --git a/paging/common/src/main/java/android/arch/paging/TiledPagedList.java b/paging/common/src/main/java/android/arch/paging/TiledPagedList.java
index 934a0dd..76bb682 100644
--- a/paging/common/src/main/java/android/arch/paging/TiledPagedList.java
+++ b/paging/common/src/main/java/android/arch/paging/TiledPagedList.java
@@ -17,7 +17,6 @@
 package android.arch.paging;
 
 import android.support.annotation.AnyThread;
-import android.support.annotation.MainThread;
 import android.support.annotation.NonNull;
 import android.support.annotation.Nullable;
 import android.support.annotation.WorkerThread;
@@ -46,7 +45,9 @@
             });
         }
 
-        @MainThread
+        // Creation thread for initial synchronous load, otherwise main thread
+        // Safe to access main thread only state - no other thread has reference during construction
+        @AnyThread
         @Override
         public void onPageResult(@NonNull PageResult<Integer, T> pageResult) {
             if (pageResult.page == null) {
@@ -67,6 +68,13 @@
                 mKeyedStorage.insertPage(pageResult.leadingNulls, pageResult.page,
                         TiledPagedList.this);
             }
+
+            if (mBoundaryCallback != null) {
+                boolean deferEmpty = mStorage.size() == 0;
+                boolean deferBegin = !deferEmpty && pageResult.leadingNulls == 0;
+                boolean deferEnd = !deferEmpty && pageResult.trailingNulls == 0;
+                deferBoundaryCallbacks(deferEmpty, deferBegin, deferEnd);
+            }
         }
     };
 
@@ -74,15 +82,17 @@
     TiledPagedList(@NonNull TiledDataSource<T> dataSource,
             @NonNull Executor mainThreadExecutor,
             @NonNull Executor backgroundThreadExecutor,
+            @Nullable BoundaryCallback<T> boundaryCallback,
             @NonNull Config config,
             int position) {
-        super(new PagedStorage<Integer, T>(),
-                mainThreadExecutor, backgroundThreadExecutor, config);
+        super(new PagedStorage<Integer, T>(), mainThreadExecutor, backgroundThreadExecutor,
+                boundaryCallback, config);
         mDataSource = dataSource;
 
         final int pageSize = mConfig.pageSize;
 
         final int itemCount = mDataSource.countItems();
+
         final int firstLoadSize = Math.min(itemCount,
                 (Math.max(mConfig.initialLoadSizeHint / pageSize, 2)) * pageSize);
         final int firstLoadPosition = computeFirstLoadPosition(
diff --git a/paging/common/src/test/java/android/arch/paging/ContiguousPagedListTest.kt b/paging/common/src/test/java/android/arch/paging/ContiguousPagedListTest.kt
index 1777cf2..03d553a 100644
--- a/paging/common/src/test/java/android/arch/paging/ContiguousPagedListTest.kt
+++ b/paging/common/src/test/java/android/arch/paging/ContiguousPagedListTest.kt
@@ -22,6 +22,8 @@
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
+import org.mockito.Mockito.any
+import org.mockito.Mockito.eq
 import org.mockito.Mockito.mock
 import org.mockito.Mockito.verify
 import org.mockito.Mockito.verifyNoMoreInteractions
@@ -41,17 +43,18 @@
         }
     }
 
-    private inner class TestSource : PositionalDataSource<Item>() {
+    private inner class TestSource(val listData: List<Item> = ITEMS)
+            : PositionalDataSource<Item>() {
         override fun countItems(): Int {
             return if (mCounted) {
-                ITEMS.size
+                listData.size
             } else {
                 DataSource.COUNT_UNDEFINED
             }
         }
 
         private fun getClampedRange(startInc: Int, endExc: Int, reverse: Boolean): List<Item> {
-            val list = ITEMS.subList(Math.max(0, startInc), Math.min(ITEMS.size, endExc))
+            val list = listData.subList(Math.max(0, startInc), Math.min(listData.size, endExc))
             if (reverse) {
                 Collections.reverse(list)
             }
@@ -140,21 +143,20 @@
         verifyRange(90, 10, TestSource().loadInitial(95, 10, true)!!)
     }
 
-
     private fun createCountedPagedList(
-            config: PagedList.Config, initialPosition: Int): ContiguousPagedList<Int, Item> {
+            initialPosition: Int,
+            pageSize: Int = 20,
+            initLoadSize: Int = 40,
+            prefetchDistance: Int = 20,
+            listData: List<Item> = ITEMS,
+            boundaryCallback: PagedList.BoundaryCallback<Item>? = null)
+            : ContiguousPagedList<Int, Item> {
         return ContiguousPagedList(
-                TestSource(), mMainThread, mBackgroundThread,
-                config,
-                initialPosition)
-    }
-
-    private fun createCountedPagedList(initialPosition: Int): ContiguousPagedList<Int, Item> {
-        return createCountedPagedList(
+                TestSource(listData), mMainThread, mBackgroundThread, boundaryCallback,
                 PagedList.Config.Builder()
-                        .setInitialLoadSizeHint(40)
-                        .setPageSize(20)
-                        .setPrefetchDistance(20)
+                        .setInitialLoadSizeHint(initLoadSize)
+                        .setPageSize(pageSize)
+                        .setPrefetchDistance(prefetchDistance)
                         .build(),
                 initialPosition)
     }
@@ -240,13 +242,8 @@
 
     @Test
     fun distantPrefetch() {
-        val pagedList = createCountedPagedList(
-                PagedList.Config.Builder()
-                        .setInitialLoadSizeHint(10)
-                        .setPageSize(10)
-                        .setPrefetchDistance(30)
-                        .build(),
-                0)
+        val pagedList = createCountedPagedList(0,
+                initLoadSize = 10, pageSize = 10, prefetchDistance = 30)
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(null, callback)
         verifyRange(0, 10, pagedList)
@@ -303,7 +300,6 @@
         val snapshot = pagedList.snapshot() as PagedList<Item>
         verifyRange(40, 60, snapshot)
 
-
         pagedList.loadAround(if (mCounted) 45 else 5)
         drain()
         verifyRange(20, 80, pagedList)
@@ -315,6 +311,71 @@
         verifyNoMoreInteractions(callback)
     }
 
+    @Test
+    fun boundaryCallback_empty() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createCountedPagedList(0,
+                listData = ArrayList(), boundaryCallback = boundaryCallback)
+        assertEquals(0, pagedList.size)
+
+        // nothing yet
+        verifyNoMoreInteractions(boundaryCallback)
+
+        // onZeroItemsLoaded posted, since creation often happens on BG thread
+        drain()
+        verify(boundaryCallback).onZeroItemsLoaded()
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
+    @Test
+    fun boundaryCallback_delayed() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createCountedPagedList(90,
+                initLoadSize = 20, prefetchDistance = 5, boundaryCallback = boundaryCallback)
+        verifyRange(80, 20, pagedList)
+
+
+        // nothing yet
+        verifyZeroInteractions(boundaryCallback)
+        drain()
+        verifyZeroInteractions(boundaryCallback)
+
+        // loading around last item causes onItemAtEndLoaded
+        pagedList.loadAround(if (mCounted) 99 else 19)
+        drain()
+        verifyRange(80, 20, pagedList)
+        verify(boundaryCallback).onItemAtEndLoaded(
+                any(), eq(ITEMS.last()), eq(if (mCounted) 100 else 20))
+        verifyNoMoreInteractions(boundaryCallback)
+
+
+        // prepending doesn't trigger callback...
+        pagedList.loadAround(if (mCounted) 80 else 0)
+        drain()
+        verifyRange(60, 40, pagedList)
+        verifyZeroInteractions(boundaryCallback)
+
+        // ...load rest of data, still no dispatch...
+        pagedList.loadAround(if (mCounted) 60 else 0)
+        drain()
+        pagedList.loadAround(if (mCounted) 40 else 0)
+        drain()
+        pagedList.loadAround(if (mCounted) 20 else 0)
+        drain()
+        verifyRange(0, 100, pagedList)
+        verifyZeroInteractions(boundaryCallback)
+
+        // ... finally try prepend, see 0 items, which will dispatch front callback
+        pagedList.loadAround(0)
+        drain()
+        verify(boundaryCallback).onItemAtFrontLoaded(any(), eq(ITEMS.first()), eq(100))
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
     private fun drain() {
         var executed: Boolean
         do {
diff --git a/paging/common/src/test/java/android/arch/paging/TiledPagedListTest.kt b/paging/common/src/test/java/android/arch/paging/TiledPagedListTest.kt
index 27b80ab..6524008 100644
--- a/paging/common/src/test/java/android/arch/paging/TiledPagedListTest.kt
+++ b/paging/common/src/test/java/android/arch/paging/TiledPagedListTest.kt
@@ -23,6 +23,8 @@
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.JUnit4
+import org.mockito.Mockito.any
+import org.mockito.Mockito.eq
 import org.mockito.Mockito.mock
 import org.mockito.Mockito.verify
 import org.mockito.Mockito.verifyNoMoreInteractions
@@ -41,37 +43,27 @@
         }
     }
 
-    private class TestTiledSource : TiledDataSource<Item>() {
-        override fun countItems(): Int {
-            return ITEMS.size
-        }
-
-        override fun loadRange(startPosition: Int, count: Int): List<Item> {
-            val endPosition = Math.min(ITEMS.size, startPosition + count)
-            return ITEMS.subList(startPosition, endPosition)
-        }
-    }
-
-    private fun verifyRange(list: List<Item>, vararg loadedPages: Int) {
+    private fun verifyLoadedPages(list: List<Item>, vararg loadedPages: Int, expected: List<Item> = ITEMS) {
         val loadedPageList = loadedPages.asList()
-        assertEquals(ITEMS.size, list.size)
+        assertEquals(expected.size, list.size)
         for (i in list.indices) {
             if (loadedPageList.contains(i / PAGE_SIZE)) {
-                assertSame("Index $i", ITEMS[i], list[i])
+                assertSame("Index $i", expected[i], list[i])
             } else {
                 assertNull("Index $i", list[i])
             }
         }
     }
 
-    private fun createTiledPagedList(loadPosition: Int, initPages: Int,
-            prefetchDistance: Int = PAGE_SIZE): TiledPagedList<Item> {
-        val source = TestTiledSource()
+    private fun createTiledPagedList(loadPosition: Int, initPageCount: Int,
+            prefetchDistance: Int = PAGE_SIZE,
+            listData: List<Item> = ITEMS,
+            boundaryCallback: PagedList.BoundaryCallback<Item>? = null): TiledPagedList<Item> {
         return TiledPagedList(
-                source, mMainThread, mBackgroundThread,
+                ListDataSource(listData), mMainThread, mBackgroundThread, boundaryCallback,
                 PagedList.Config.Builder()
                         .setPageSize(PAGE_SIZE)
-                        .setInitialLoadSizeHint(PAGE_SIZE * initPages)
+                        .setInitialLoadSizeHint(PAGE_SIZE * initPageCount)
                         .setPrefetchDistance(prefetchDistance)
                         .build(),
                 loadPosition)
@@ -94,87 +86,87 @@
 
     @Test
     fun initialLoad_onePage() {
-        val pagedList = createTiledPagedList(0, 1)
-        verifyRange(pagedList, 0, 1)
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 1)
+        verifyLoadedPages(pagedList, 0, 1)
     }
 
     @Test
     fun initialLoad_onePageOffset() {
-        val pagedList = createTiledPagedList(10, 1)
-        verifyRange(pagedList, 0, 1)
+        val pagedList = createTiledPagedList(loadPosition = 10, initPageCount = 1)
+        verifyLoadedPages(pagedList, 0, 1)
     }
 
     @Test
     fun initialLoad_full() {
-        val pagedList = createTiledPagedList(0, 100)
-        verifyRange(pagedList, 0, 1, 2, 3, 4)
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 100)
+        verifyLoadedPages(pagedList, 0, 1, 2, 3, 4)
     }
 
     @Test
     fun initialLoad_end() {
-        val pagedList = createTiledPagedList(44, 2)
-        verifyRange(pagedList, 3, 4)
+        val pagedList = createTiledPagedList(loadPosition = 44, initPageCount = 2)
+        verifyLoadedPages(pagedList, 3, 4)
     }
 
     @Test
     fun initialLoad_multiple() {
-        val pagedList = createTiledPagedList(9, 2)
-        verifyRange(pagedList, 0, 1)
+        val pagedList = createTiledPagedList(loadPosition = 9, initPageCount = 2)
+        verifyLoadedPages(pagedList, 0, 1)
     }
 
     @Test
     fun initialLoad_offset() {
-        val pagedList = createTiledPagedList(41, 2)
-        verifyRange(pagedList, 3, 4)
+        val pagedList = createTiledPagedList(loadPosition = 41, initPageCount = 2)
+        verifyLoadedPages(pagedList, 3, 4)
     }
 
     @Test
     fun append() {
-        val pagedList = createTiledPagedList(0, 1)
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 1)
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(null, callback)
-        verifyRange(pagedList, 0, 1)
+        verifyLoadedPages(pagedList, 0, 1)
         verifyZeroInteractions(callback)
 
         pagedList.loadAround(15)
 
-        verifyRange(pagedList, 0, 1)
+        verifyLoadedPages(pagedList, 0, 1)
 
         drain()
 
-        verifyRange(pagedList, 0, 1, 2)
+        verifyLoadedPages(pagedList, 0, 1, 2)
         verify(callback).onChanged(20, 10)
         verifyNoMoreInteractions(callback)
     }
 
     @Test
     fun prepend() {
-        val pagedList = createTiledPagedList(44, 2)
+        val pagedList = createTiledPagedList(loadPosition = 44, initPageCount = 2)
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(null, callback)
-        verifyRange(pagedList, 3, 4)
+        verifyLoadedPages(pagedList, 3, 4)
         verifyZeroInteractions(callback)
 
         pagedList.loadAround(35)
         drain()
 
-        verifyRange(pagedList, 2, 3, 4)
+        verifyLoadedPages(pagedList, 2, 3, 4)
         verify<PagedList.Callback>(callback).onChanged(20, 10)
         verifyNoMoreInteractions(callback)
     }
 
     @Test
     fun loadWithGap() {
-        val pagedList = createTiledPagedList(0, 1)
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 1)
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(null, callback)
-        verifyRange(pagedList, 0, 1)
+        verifyLoadedPages(pagedList, 0, 1)
         verifyZeroInteractions(callback)
 
         pagedList.loadAround(44)
         drain()
 
-        verifyRange(pagedList, 0, 1, 3, 4)
+        verifyLoadedPages(pagedList, 0, 1, 3, 4)
         verify(callback).onChanged(30, 10)
         verify(callback).onChanged(40, 5)
         verifyNoMoreInteractions(callback)
@@ -182,45 +174,47 @@
 
     @Test
     fun tinyPrefetchTest() {
-        val pagedList = createTiledPagedList(0, 1, 1)
+        val pagedList = createTiledPagedList(
+                loadPosition = 0, initPageCount = 1, prefetchDistance = 1)
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(null, callback)
-        verifyRange(pagedList, 0, 1)
+        verifyLoadedPages(pagedList, 0, 1)
         verifyZeroInteractions(callback)
 
         pagedList.loadAround(33)
         drain()
 
-        verifyRange(pagedList, 0, 1, 3)
+        verifyLoadedPages(pagedList, 0, 1, 3)
         verify(callback).onChanged(30, 10)
         verifyNoMoreInteractions(callback)
 
         pagedList.loadAround(44)
         drain()
 
-        verifyRange(pagedList, 0, 1, 3, 4)
+        verifyLoadedPages(pagedList, 0, 1, 3, 4)
         verify(callback).onChanged(40, 5)
         verifyNoMoreInteractions(callback)
     }
 
     @Test
     fun appendCallbackAddedLate() {
-        val pagedList = createTiledPagedList(0, 1, 0)
-        verifyRange(pagedList, 0, 1)
+        val pagedList = createTiledPagedList(
+                loadPosition = 0, initPageCount = 1, prefetchDistance = 0)
+        verifyLoadedPages(pagedList, 0, 1)
 
         pagedList.loadAround(25)
         drain()
-        verifyRange(pagedList, 0, 1, 2)
+        verifyLoadedPages(pagedList, 0, 1, 2)
 
         // snapshot at 30 items
         val snapshot = pagedList.snapshot()
-        verifyRange(snapshot, 0, 1, 2)
+        verifyLoadedPages(snapshot, 0, 1, 2)
 
         pagedList.loadAround(35)
         pagedList.loadAround(44)
         drain()
-        verifyRange(pagedList, 0, 1, 2, 3, 4)
-        verifyRange(snapshot, 0, 1, 2)
+        verifyLoadedPages(pagedList, 0, 1, 2, 3, 4)
+        verifyLoadedPages(snapshot, 0, 1, 2)
 
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(snapshot, callback)
@@ -230,22 +224,23 @@
 
     @Test
     fun prependCallbackAddedLate() {
-        val pagedList = createTiledPagedList(44, 2, 0)
-        verifyRange(pagedList, 3, 4)
+        val pagedList = createTiledPagedList(
+                loadPosition = 44, initPageCount = 2, prefetchDistance = 0)
+        verifyLoadedPages(pagedList, 3, 4)
 
         pagedList.loadAround(25)
         drain()
-        verifyRange(pagedList, 2, 3, 4)
+        verifyLoadedPages(pagedList, 2, 3, 4)
 
         // snapshot at 30 items
         val snapshot = pagedList.snapshot()
-        verifyRange(snapshot, 2, 3, 4)
+        verifyLoadedPages(snapshot, 2, 3, 4)
 
         pagedList.loadAround(15)
         pagedList.loadAround(5)
         drain()
-        verifyRange(pagedList, 0, 1, 2, 3, 4)
-        verifyRange(snapshot, 2, 3, 4)
+        verifyLoadedPages(pagedList, 0, 1, 2, 3, 4)
+        verifyLoadedPages(snapshot, 2, 3, 4)
 
         val callback = mock(PagedList.Callback::class.java)
         pagedList.addWeakCallback(snapshot, callback)
@@ -257,7 +252,7 @@
     fun placeholdersDisabled() {
         // disable placeholders with config, so we create a contiguous version of the pagedlist
         val pagedList = PagedList.Builder<Int, Item>()
-                .setDataSource(TestTiledSource())
+                .setDataSource(ListDataSource(ITEMS))
                 .setMainThreadExecutor(mMainThread)
                 .setBackgroundThreadExecutor(mBackgroundThread)
                 .setConfig(PagedList.Config.Builder()
@@ -278,6 +273,98 @@
         assertEquals(0, contiguousPagedList.mStorage.trailingNullCount)
     }
 
+    @Test
+    fun boundaryCallback_empty() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 1,
+                listData = ArrayList(), boundaryCallback = boundaryCallback)
+        assertEquals(0, pagedList.size)
+
+        // nothing yet
+        verifyNoMoreInteractions(boundaryCallback)
+
+        // onZeroItemsLoaded posted, since creation often happens on BG thread
+        drain()
+        verify(boundaryCallback).onZeroItemsLoaded()
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
+    @Test
+    fun boundaryCallback_immediate() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 1,
+                listData = ITEMS.subList(0, 2), boundaryCallback = boundaryCallback)
+        assertEquals(2, pagedList.size)
+
+        // nothing yet
+        verifyZeroInteractions(boundaryCallback)
+
+        // callbacks posted, since creation often happens on BG thread
+        drain()
+        verify(boundaryCallback).onItemAtFrontLoaded(any(), eq(ITEMS[0]), eq(2))
+        verify(boundaryCallback).onItemAtEndLoaded(any(), eq(ITEMS[1]), eq(2))
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
+    @Test
+    fun boundaryCallback_delayedUntilLoaded() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createTiledPagedList(loadPosition = 20, initPageCount = 1,
+                boundaryCallback = boundaryCallback)
+        verifyLoadedPages(pagedList, 1, 2) // 0, 3, and 4 not loaded yet
+
+        // nothing yet, even after drain
+        verifyZeroInteractions(boundaryCallback)
+        drain()
+        verifyZeroInteractions(boundaryCallback)
+
+        pagedList.loadAround(0)
+        pagedList.loadAround(44)
+
+        // still nothing, since items aren't loaded...
+        verifyZeroInteractions(boundaryCallback)
+
+        drain()
+        // first/last items loaded now, so callbacks dispatched
+        verify(boundaryCallback).onItemAtFrontLoaded(any(), eq(ITEMS.first()), eq(45))
+        verify(boundaryCallback).onItemAtEndLoaded(any(), eq(ITEMS.last()), eq(45))
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
+    @Test
+    fun boundaryCallback_delayedUntilNearbyAccess() {
+        @Suppress("UNCHECKED_CAST")
+        val boundaryCallback =
+                mock(PagedList.BoundaryCallback::class.java) as PagedList.BoundaryCallback<Item>
+        val pagedList = createTiledPagedList(loadPosition = 0, initPageCount = 5,
+                prefetchDistance = 2, boundaryCallback = boundaryCallback)
+        verifyLoadedPages(pagedList, 0, 1, 2, 3, 4)
+
+        // all items loaded, but no access near ends, so no callbacks
+        verifyZeroInteractions(boundaryCallback)
+        drain()
+        verifyZeroInteractions(boundaryCallback)
+
+        pagedList.loadAround(0)
+        pagedList.loadAround(44)
+
+        // callbacks not posted immediately
+        verifyZeroInteractions(boundaryCallback)
+
+        drain()
+
+        // items accessed, so now posted callbacks are run
+        verify(boundaryCallback).onItemAtFrontLoaded(any(), eq(ITEMS.first()), eq(45))
+        verify(boundaryCallback).onItemAtEndLoaded(any(), eq(ITEMS.last()), eq(45))
+        verifyNoMoreInteractions(boundaryCallback)
+    }
+
     private fun drain() {
         var executed: Boolean
         do {
diff --git a/paging/integration-tests/testapp/src/main/java/android/arch/paging/integration/testapp/PagedListItemViewModel.java b/paging/integration-tests/testapp/src/main/java/android/arch/paging/integration/testapp/PagedListItemViewModel.java
index 237cc14..974eab9 100644
--- a/paging/integration-tests/testapp/src/main/java/android/arch/paging/integration/testapp/PagedListItemViewModel.java
+++ b/paging/integration-tests/testapp/src/main/java/android/arch/paging/integration/testapp/PagedListItemViewModel.java
@@ -19,7 +19,7 @@
 import android.arch.lifecycle.LiveData;
 import android.arch.lifecycle.ViewModel;
 import android.arch.paging.DataSource;
-import android.arch.paging.LivePagedListProvider;
+import android.arch.paging.LivePagedListBuilder;
 import android.arch.paging.PagedList;
 
 /**
@@ -41,16 +41,19 @@
 
     LiveData<PagedList<Item>> getLivePagedList() {
         if (mLivePagedList == null) {
-            mLivePagedList = new LivePagedListProvider<Integer, Item>() {
-                @Override
-                protected DataSource<Integer, Item> createDataSource() {
-                    ItemDataSource newDataSource = new ItemDataSource();
-                    synchronized (mDataSourceLock) {
-                        mDataSource = newDataSource;
-                        return mDataSource;
-                    }
-                }
-            }.create(0, 20);
+            mLivePagedList = new LivePagedListBuilder<Integer, Item>()
+                    .setPagingConfig(20)
+                    .setDataSourceFactory(new DataSource.Factory<Integer, Item>() {
+                        @Override
+                        public DataSource<Integer, Item> create() {
+                            ItemDataSource newDataSource = new ItemDataSource();
+                            synchronized (mDataSourceLock) {
+                                mDataSource = newDataSource;
+                                return mDataSource;
+                            }
+                        }
+                    })
+                    .build();
         }
 
         return mLivePagedList;
diff --git a/paging/runtime/src/androidTest/java/android/arch/paging/ListDataSource.kt b/paging/runtime/src/androidTest/java/android/arch/paging/ListDataSource.kt
deleted file mode 100644
index b206d2e..0000000
--- a/paging/runtime/src/androidTest/java/android/arch/paging/ListDataSource.kt
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Copyright (C) 2017 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 android.arch.paging
-
-class ListDataSource<T> constructor(private val list: List<T>) : TiledDataSource<T>() {
-
-    override fun countItems(): Int {
-        return list.size
-    }
-
-    override fun loadRange(startPosition: Int, count: Int): List<T> {
-        val endExclusive = Math.min(list.size, startPosition + count)
-        return list.subList(startPosition, endExclusive)
-    }
-}
diff --git a/paging/runtime/src/androidTest/java/android/arch/paging/PagedListAdapterHelperTest.kt b/paging/runtime/src/androidTest/java/android/arch/paging/PagedListAdapterHelperTest.kt
index 9e95316..735a61f 100644
--- a/paging/runtime/src/androidTest/java/android/arch/paging/PagedListAdapterHelperTest.kt
+++ b/paging/runtime/src/androidTest/java/android/arch/paging/PagedListAdapterHelperTest.kt
@@ -20,12 +20,12 @@
 import android.support.v7.recyclerview.extensions.DiffCallback
 import android.support.v7.recyclerview.extensions.ListAdapterConfig
 import android.support.v7.util.ListUpdateCallback
-import junit.framework.Assert.assertEquals
-import junit.framework.Assert.assertFalse
-import junit.framework.Assert.assertNotNull
-import junit.framework.Assert.assertNull
-import junit.framework.Assert.assertTrue
-import junit.framework.Assert.fail
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Assert.fail
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.JUnit4
diff --git a/paging/runtime/src/androidTest/java/android/arch/paging/StringPagedList.kt b/paging/runtime/src/androidTest/java/android/arch/paging/StringPagedList.kt
index 86a52ab..c2e5ec7 100644
--- a/paging/runtime/src/androidTest/java/android/arch/paging/StringPagedList.kt
+++ b/paging/runtime/src/androidTest/java/android/arch/paging/StringPagedList.kt
@@ -17,7 +17,7 @@
 package android.arch.paging
 
 class StringPagedList constructor(leadingNulls: Int, trailingNulls: Int, vararg items: String)
-        : PagedList<String>(PagedStorage<Int, String>(), TestExecutor(), TestExecutor(),
+        : PagedList<String>(PagedStorage<Int, String>(), TestExecutor(), TestExecutor(), null,
                 PagedList.Config.Builder().setPageSize(1).build()), PagedStorage.Callback {
     init {
         @Suppress("UNCHECKED_CAST")
diff --git a/paging/runtime/src/main/java/android/arch/paging/LivePagedListBuilder.java b/paging/runtime/src/main/java/android/arch/paging/LivePagedListBuilder.java
new file mode 100644
index 0000000..ee1810b
--- /dev/null
+++ b/paging/runtime/src/main/java/android/arch/paging/LivePagedListBuilder.java
@@ -0,0 +1,159 @@
+/*
+ * Copyright (C) 2017 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 android.arch.paging;
+
+import android.arch.core.executor.ArchTaskExecutor;
+import android.arch.lifecycle.ComputableLiveData;
+import android.arch.lifecycle.LiveData;
+import android.support.annotation.AnyThread;
+import android.support.annotation.NonNull;
+import android.support.annotation.Nullable;
+
+import java.util.concurrent.Executor;
+
+public class LivePagedListBuilder<Key, Value> {
+    private Key mInitialLoadKey;
+    private PagedList.Config mConfig;
+    private DataSource.Factory<Key, Value> mDataSourceFactory;
+    private PagedList.BoundaryCallback mBoundaryCallback;
+    private Executor mMainThreadExecutor;
+    private Executor mBackgroundThreadExecutor;
+
+    @SuppressWarnings("WeakerAccess")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setInitialLoadKey(@Nullable Key key) {
+        mInitialLoadKey = key;
+        return this;
+    }
+
+    @SuppressWarnings("WeakerAccess")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setPagingConfig(@NonNull PagedList.Config config) {
+        mConfig = config;
+        return this;
+    }
+
+    @SuppressWarnings("WeakerAccess")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setPagingConfig(int pageSize) {
+        mConfig = new PagedList.Config.Builder().setPageSize(pageSize).build();
+        return this;
+    }
+
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setDataSourceFactory(
+            @NonNull DataSource.Factory<Key, Value> dataSourceFactory) {
+        mDataSourceFactory = dataSourceFactory;
+        return this;
+    }
+
+    @SuppressWarnings("unused")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setBoundaryCallback(
+            @Nullable PagedList.BoundaryCallback<Value> boundaryCallback) {
+        mBoundaryCallback = boundaryCallback;
+        return this;
+    }
+
+    @SuppressWarnings("unused")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setMainThreadExecutor(
+            @NonNull Executor mainThreadExecutor) {
+        mMainThreadExecutor = mainThreadExecutor;
+        return this;
+    }
+
+    @SuppressWarnings("unused")
+    @NonNull
+    public LivePagedListBuilder<Key, Value> setBackgroundThreadExecutor(
+            @NonNull Executor backgroundThreadExecutor) {
+        mBackgroundThreadExecutor = backgroundThreadExecutor;
+        return this;
+    }
+
+    @NonNull
+    public LiveData<PagedList<Value>> build() {
+        if (mConfig == null) {
+            throw new IllegalArgumentException("PagedList.Config must be provided");
+        }
+        if (mDataSourceFactory == null) {
+            throw new IllegalArgumentException("DataSource.Factory must be provided");
+        }
+        if (mMainThreadExecutor == null) {
+            mMainThreadExecutor = ArchTaskExecutor.getMainThreadExecutor();
+        }
+        if (mBackgroundThreadExecutor == null) {
+            mBackgroundThreadExecutor = ArchTaskExecutor.getIOThreadExecutor();
+        }
+
+        return create(mInitialLoadKey, mConfig, mBoundaryCallback, mDataSourceFactory,
+                mMainThreadExecutor, mBackgroundThreadExecutor);
+    }
+
+    @AnyThread
+    @NonNull
+    public static <Key, Value> LiveData<PagedList<Value>> create(
+            @Nullable final Key initialLoadKey,
+            @NonNull final PagedList.Config config,
+            @Nullable final PagedList.BoundaryCallback boundaryCallback,
+            @NonNull final DataSource.Factory<Key, Value> dataSourceFactory,
+            @NonNull final Executor mainThreadExecutor,
+            @NonNull final Executor backgroundThreadExecutor) {
+        return new ComputableLiveData<PagedList<Value>>() {
+            @Nullable
+            private PagedList<Value> mList;
+            @Nullable
+            private DataSource<Key, Value> mDataSource;
+
+            private final DataSource.InvalidatedCallback mCallback =
+                    new DataSource.InvalidatedCallback() {
+                        @Override
+                        public void onInvalidated() {
+                            invalidate();
+                        }
+                    };
+
+            @Override
+            protected PagedList<Value> compute() {
+                @Nullable Key initializeKey = initialLoadKey;
+                if (mList != null) {
+                    //noinspection unchecked
+                    initializeKey = (Key) mList.getLastKey();
+                }
+
+                do {
+                    if (mDataSource != null) {
+                        mDataSource.removeInvalidatedCallback(mCallback);
+                    }
+
+                    mDataSource = dataSourceFactory.create();
+                    mDataSource.addInvalidatedCallback(mCallback);
+
+                    mList = new PagedList.Builder<Key, Value>()
+                            .setDataSource(mDataSource)
+                            .setMainThreadExecutor(mainThreadExecutor)
+                            .setBackgroundThreadExecutor(backgroundThreadExecutor)
+                            .setBoundaryCallback(boundaryCallback)
+                            .setConfig(config)
+                            .setInitialKey(initializeKey)
+                            .build();
+                } while (mList.isDetached());
+                return mList;
+            }
+        }.getLiveData();
+    }
+}
diff --git a/paging/runtime/src/main/java/android/arch/paging/LivePagedListProvider.java b/paging/runtime/src/main/java/android/arch/paging/LivePagedListProvider.java
index 07dd84b..e0a03cb 100644
--- a/paging/runtime/src/main/java/android/arch/paging/LivePagedListProvider.java
+++ b/paging/runtime/src/main/java/android/arch/paging/LivePagedListProvider.java
@@ -16,8 +16,6 @@
 
 package android.arch.paging;
 
-import android.arch.core.executor.ArchTaskExecutor;
-import android.arch.lifecycle.ComputableLiveData;
 import android.arch.lifecycle.LiveData;
 import android.support.annotation.AnyThread;
 import android.support.annotation.NonNull;
@@ -53,8 +51,23 @@
  * @see PagedListAdapter
  * @see DataSource
  * @see PagedList
+ *
+ * @deprecated To construct a {@code LiveData<PagedList>}, use {@link LivePagedListBuilder}, which
+ * provides the same construction capability with more customization, and better defaults. The role
+ * of DataSource construction has been separated out to {@link DataSource.Factory} to access or
+ * provide a self-invalidating sequence of DataSources. If you were acquiring this from Room, you
+ * can switch to having your Dao return a {@link DataSource.Factory} instead, and create a LiveData
+ * of PagedList with a {@link LivePagedListBuilder}.
  */
-public abstract class LivePagedListProvider<Key, Value> {
+// NOTE: Room 1.0 depends on this class, so it should not be removed
+// until Room switches to using DataSource.Factory directly
+@Deprecated
+public abstract class LivePagedListProvider<Key, Value> implements DataSource.Factory<Key, Value> {
+
+    @Override
+    public DataSource<Key, Value> create() {
+        return createDataSource();
+    }
 
     /**
      * Construct a new data source to be wrapped in a new PagedList, which will be returned
@@ -80,10 +93,11 @@
     @AnyThread
     @NonNull
     public LiveData<PagedList<Value>> create(@Nullable Key initialLoadKey, int pageSize) {
-        return create(initialLoadKey,
-                new PagedList.Config.Builder()
-                        .setPageSize(pageSize)
-                        .build());
+        return new LivePagedListBuilder<Key, Value>()
+                .setInitialLoadKey(initialLoadKey)
+                .setPagingConfig(pageSize)
+                .setDataSourceFactory(this)
+                .build();
     }
 
     /**
@@ -100,49 +114,12 @@
      */
     @AnyThread
     @NonNull
-    public LiveData<PagedList<Value>> create(@Nullable final Key initialLoadKey,
-            final PagedList.Config config) {
-        return new ComputableLiveData<PagedList<Value>>() {
-            @Nullable
-            private PagedList<Value> mList;
-            @Nullable
-            private DataSource<Key, Value> mDataSource;
-
-            private final DataSource.InvalidatedCallback mCallback =
-                    new DataSource.InvalidatedCallback() {
-                @Override
-                public void onInvalidated() {
-                    invalidate();
-                }
-            };
-
-            @Override
-            protected PagedList<Value> compute() {
-                @Nullable Key initializeKey = initialLoadKey;
-                if (mList != null) {
-                    //noinspection unchecked
-                    initializeKey = (Key) mList.getLastKey();
-                }
-
-                do {
-                    if (mDataSource != null) {
-                        mDataSource.removeInvalidatedCallback(mCallback);
-                    }
-
-                    mDataSource = createDataSource();
-                    mDataSource.addInvalidatedCallback(mCallback);
-
-                    mList = new PagedList.Builder<Key, Value>()
-                            .setDataSource(mDataSource)
-                            .setMainThreadExecutor(ArchTaskExecutor.getMainThreadExecutor())
-                            .setBackgroundThreadExecutor(
-                                    ArchTaskExecutor.getIOThreadExecutor())
-                            .setConfig(config)
-                            .setInitialKey(initializeKey)
-                            .build();
-                } while (mList.isDetached());
-                return mList;
-            }
-        }.getLiveData();
+    public LiveData<PagedList<Value>> create(@Nullable Key initialLoadKey,
+            @NonNull PagedList.Config config) {
+        return new LivePagedListBuilder<Key, Value>()
+                .setInitialLoadKey(initialLoadKey)
+                .setPagingConfig(config)
+                .setDataSourceFactory(this)
+                .build();
     }
 }
diff --git a/paging/runtime/src/main/java/android/arch/paging/PagedListAdapter.java b/paging/runtime/src/main/java/android/arch/paging/PagedListAdapter.java
index 93c02ea..89b9c2e 100644
--- a/paging/runtime/src/main/java/android/arch/paging/PagedListAdapter.java
+++ b/paging/runtime/src/main/java/android/arch/paging/PagedListAdapter.java
@@ -113,6 +113,13 @@
 public abstract class PagedListAdapter<T, VH extends RecyclerView.ViewHolder>
         extends RecyclerView.Adapter<VH> {
     private final PagedListAdapterHelper<T> mHelper;
+    private final PagedListAdapterHelper.PagedListListener<T> mListener =
+            new PagedListAdapterHelper.PagedListListener<T>() {
+        @Override
+        public void onCurrentListChanged(@Nullable PagedList<T> currentList) {
+            PagedListAdapter.this.onCurrentListChanged(currentList);
+        }
+    };
 
     /**
      * Creates a PagedListAdapter with default threading and
@@ -125,11 +132,13 @@
      */
     protected PagedListAdapter(@NonNull DiffCallback<T> diffCallback) {
         mHelper = new PagedListAdapterHelper<>(this, diffCallback);
+        mHelper.mListener = mListener;
     }
 
     @SuppressWarnings("unused, WeakerAccess")
     protected PagedListAdapter(@NonNull ListAdapterConfig<T> config) {
         mHelper = new PagedListAdapterHelper<>(new ListAdapterHelper.AdapterCallback(this), config);
+        mHelper.mListener = mListener;
     }
 
     /**
@@ -167,4 +176,22 @@
     public PagedList<T> getCurrentList() {
         return mHelper.getCurrentList();
     }
+
+    /**
+     * Called when the current PagedList is updated.
+     * <p>
+     * This may be dispatched as part of {@link #setList(PagedList)} if a background diff isn't
+     * needed (such as when the first list is passed, or the list is cleared). In either case,
+     * PagedListAdapter will simply call
+     * {@link #notifyItemRangeInserted(int, int) notifyItemRangeInserted/Removed(0, mPreviousSize)}.
+     * <p>
+     * This method will <em>not</em>be called when the Adapter switches from presenting a PagedList
+     * to a snapshot version of the PagedList during a diff. This means you cannot observe each
+     * PagedList via this method.
+     *
+     * @param currentList new PagedList being displayed, may be null.
+     */
+    @SuppressWarnings("WeakerAccess")
+    public void onCurrentListChanged(@Nullable PagedList<T> currentList) {
+    }
 }
diff --git a/paging/runtime/src/main/java/android/arch/paging/PagedListAdapterHelper.java b/paging/runtime/src/main/java/android/arch/paging/PagedListAdapterHelper.java
index abcff41..51a6e37 100644
--- a/paging/runtime/src/main/java/android/arch/paging/PagedListAdapterHelper.java
+++ b/paging/runtime/src/main/java/android/arch/paging/PagedListAdapterHelper.java
@@ -123,6 +123,14 @@
     private final ListUpdateCallback mUpdateCallback;
     private final ListAdapterConfig<T> mConfig;
 
+    // TODO: REAL API
+    interface PagedListListener<T> {
+        void onCurrentListChanged(@Nullable PagedList<T> currentList);
+    }
+
+    @Nullable
+    PagedListListener<T> mListener;
+
     private boolean mIsContiguous;
 
     private PagedList<T> mPagedList;
@@ -247,6 +255,9 @@
             }
             // dispatch update callback after updating mPagedList/mSnapshot
             mUpdateCallback.onRemoved(0, removedCount);
+            if (mListener != null) {
+                mListener.onCurrentListChanged(null);
+            }
             return;
         }
 
@@ -257,6 +268,10 @@
 
             // dispatch update callback after updating mPagedList/mSnapshot
             mUpdateCallback.onInserted(0, pagedList.size());
+
+            if (mListener != null) {
+                mListener.onCurrentListChanged(pagedList);
+            }
             return;
         }
 
@@ -311,6 +326,9 @@
                 previousSnapshot.mStorage, newList.mStorage, diffResult);
 
         newList.addWeakCallback(diffSnapshot, mPagedListCallback);
+        if (mListener != null) {
+            mListener.onCurrentListChanged(mPagedList);
+        }
     }
 
     /**
diff --git a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/dao/UserDao.java b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/dao/UserDao.java
index 665a1ae..7e7a333 100644
--- a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/dao/UserDao.java
+++ b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/dao/UserDao.java
@@ -17,6 +17,7 @@
 package android.arch.persistence.room.integration.testapp.dao;
 
 import android.arch.lifecycle.LiveData;
+import android.arch.paging.DataSource;
 import android.arch.paging.LivePagedListProvider;
 import android.arch.paging.TiledDataSource;
 import android.arch.persistence.room.Dao;
@@ -184,7 +185,10 @@
     }
 
     @Query("SELECT * FROM user where mAge > :age")
-    public abstract LivePagedListProvider<Integer, User> loadPagedByAge(int age);
+    public abstract DataSource.Factory<Integer, User> loadPagedByAge(int age);
+
+    @Query("SELECT * FROM user where mAge > :age")
+    public abstract LivePagedListProvider<Integer, User> loadPagedByAge_legacy(int age);
 
     @Query("SELECT * FROM user ORDER BY mAge DESC")
     public abstract TiledDataSource<User> loadUsersByAgeDesc();
diff --git a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/LivePagedListProviderTest.java b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/DataSourceFactoryTest.java
similarity index 75%
rename from room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/LivePagedListProviderTest.java
rename to room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/DataSourceFactoryTest.java
index df70a17..c546531 100644
--- a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/LivePagedListProviderTest.java
+++ b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/paging/DataSourceFactoryTest.java
@@ -28,6 +28,7 @@
 import android.arch.lifecycle.LifecycleRegistry;
 import android.arch.lifecycle.LiveData;
 import android.arch.lifecycle.Observer;
+import android.arch.paging.LivePagedListBuilder;
 import android.arch.paging.PagedList;
 import android.arch.persistence.room.integration.testapp.test.TestDatabaseTest;
 import android.arch.persistence.room.integration.testapp.test.TestUtil;
@@ -46,15 +47,54 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+@LargeTest
 @RunWith(AndroidJUnit4.class)
-public class LivePagedListProviderTest extends TestDatabaseTest {
+public class DataSourceFactoryTest extends TestDatabaseTest {
     @Rule
     public CountingTaskExecutorRule mExecutorRule = new CountingTaskExecutorRule();
 
+    private interface LivePagedListFactory {
+        LiveData<PagedList<User>> create();
+    }
+
     @Test
-    @LargeTest
     public void getUsersAsPagedList()
             throws InterruptedException, ExecutionException, TimeoutException {
+        validateUsersAsPagedList(new LivePagedListFactory() {
+            @Override
+            public LiveData<PagedList<User>> create() {
+                return new LivePagedListBuilder<Integer, User>()
+                        .setPagingConfig(new PagedList.Config.Builder()
+                                .setPageSize(10)
+                                .setPrefetchDistance(1)
+                                .setInitialLoadSizeHint(10).build())
+                        .setDataSourceFactory(mUserDao.loadPagedByAge(3))
+                        .build();
+            }
+        });
+    }
+
+
+    // TODO: delete this and factory abstraction when LivePagedListProvider is removed
+    @Test
+    public void getUsersAsPagedList_legacyLivePagedListProvider()
+            throws InterruptedException, ExecutionException, TimeoutException {
+        validateUsersAsPagedList(new LivePagedListFactory() {
+            @Override
+            public LiveData<PagedList<User>> create() {
+                return mUserDao.loadPagedByAge_legacy(3).create(
+                        0,
+                        new PagedList.Config.Builder()
+                                .setPageSize(10)
+                                .setPrefetchDistance(1)
+                                .setInitialLoadSizeHint(10)
+                                .build());
+            }
+        });
+    }
+
+    private void validateUsersAsPagedList(LivePagedListFactory factory)
+            throws InterruptedException, ExecutionException, TimeoutException {
         mDatabase.beginTransaction();
         try {
             for (int i = 0; i < 100; i++) {
@@ -67,12 +107,8 @@
             mDatabase.endTransaction();
         }
         assertThat(mUserDao.count(), is(100));
-        final LiveData<PagedList<User>> livePagedUsers = mUserDao.loadPagedByAge(3).create(
-                0,
-                new PagedList.Config.Builder()
-                        .setPageSize(10)
-                        .setPrefetchDistance(1)
-                        .setInitialLoadSizeHint(10).build());
+
+        final LiveData<PagedList<User>> livePagedUsers = factory.create();
 
         final TestLifecycleOwner testOwner = new TestLifecycleOwner();
         testOwner.handleEvent(Lifecycle.Event.ON_CREATE);
diff --git a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/test/QueryTransactionTest.java b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/test/QueryTransactionTest.java
index 854c862..291cfd6 100644
--- a/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/test/QueryTransactionTest.java
+++ b/room/integration-tests/testapp/src/androidTest/java/android/arch/persistence/room/integration/testapp/test/QueryTransactionTest.java
@@ -25,7 +25,8 @@
 import android.arch.lifecycle.Lifecycle;
 import android.arch.lifecycle.LiveData;
 import android.arch.lifecycle.Observer;
-import android.arch.paging.LivePagedListProvider;
+import android.arch.paging.DataSource;
+import android.arch.paging.LivePagedListBuilder;
 import android.arch.paging.PagedList;
 import android.arch.paging.TiledDataSource;
 import android.arch.persistence.room.Dao;
@@ -202,7 +203,10 @@
 
     @Test
     public void pagedList() {
-        LiveData<PagedList<Entity1>> pagedList = mDao.pagedList().create(null, 10);
+        LiveData<PagedList<Entity1>> pagedList = new LivePagedListBuilder<Integer, Entity1>()
+                .setDataSourceFactory(mDao.pagedList())
+                .setPagingConfig(10)
+                .build();
         observeForever(pagedList);
         drain();
         assertThat(sStartedTransactionCount.get(), is(mUseTransactionDao ? 0 : 0));
@@ -366,7 +370,7 @@
 
         List<Entity1WithChildren> withRelation();
 
-        LivePagedListProvider<Integer, Entity1> pagedList();
+        DataSource.Factory<Integer, Entity1> pagedList();
 
         TiledDataSource<Entity1> dataSource();
 
@@ -406,7 +410,7 @@
 
         @Override
         @Query(SELECT_ALL)
-        LivePagedListProvider<Integer, Entity1> pagedList();
+        DataSource.Factory<Integer, Entity1> pagedList();
 
         @Override
         @Query(SELECT_ALL)
@@ -448,7 +452,7 @@
         @Override
         @Transaction
         @Query(SELECT_ALL)
-        LivePagedListProvider<Integer, Entity1> pagedList();
+        DataSource.Factory<Integer, Entity1> pagedList();
 
         @Override
         @Transaction
diff --git a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/CustomerViewModel.java b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/CustomerViewModel.java
index 320b2cd..89d16b7 100644
--- a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/CustomerViewModel.java
+++ b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/CustomerViewModel.java
@@ -21,7 +21,7 @@
 import android.arch.lifecycle.AndroidViewModel;
 import android.arch.lifecycle.LiveData;
 import android.arch.paging.DataSource;
-import android.arch.paging.LivePagedListProvider;
+import android.arch.paging.LivePagedListBuilder;
 import android.arch.paging.PagedList;
 import android.arch.persistence.room.Room;
 import android.arch.persistence.room.integration.testapp.database.Customer;
@@ -81,30 +81,30 @@
         });
     }
 
+    private static <K> LiveData<PagedList<Customer>> getLivePagedList(
+            K initialLoadKey, DataSource.Factory<K, Customer> dataSourceFactory) {
+        return new LivePagedListBuilder<K, Customer>()
+                .setInitialLoadKey(initialLoadKey)
+                .setPagingConfig(new PagedList.Config.Builder()
+                        .setPageSize(10)
+                        .setEnablePlaceholders(false)
+                        .build())
+                .setDataSourceFactory(dataSourceFactory)
+                .build();
+    }
+
     LiveData<PagedList<Customer>> getLivePagedList(int position) {
         if (mLiveCustomerList == null) {
-            mLiveCustomerList = mDatabase.getCustomerDao()
-                    .loadPagedAgeOrder().create(position,
-                            new PagedList.Config.Builder()
-                                    .setPageSize(10)
-                                    .setEnablePlaceholders(false)
-                                    .build());
+            mLiveCustomerList =
+                    getLivePagedList(position, mDatabase.getCustomerDao().loadPagedAgeOrder());
         }
         return mLiveCustomerList;
     }
 
     LiveData<PagedList<Customer>> getLivePagedList(String key) {
         if (mLiveCustomerList == null) {
-            mLiveCustomerList = new LivePagedListProvider<String, Customer>() {
-                @Override
-                protected DataSource<String, Customer> createDataSource() {
-                    return new LastNameAscCustomerDataSource(mDatabase);
-                }
-            }.create(key,
-                    new PagedList.Config.Builder()
-                            .setPageSize(10)
-                            .setEnablePlaceholders(false)
-                            .build());
+            mLiveCustomerList =
+                    getLivePagedList(key, LastNameAscCustomerDataSource.factory(mDatabase));
         }
         return mLiveCustomerList;
     }
diff --git a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/CustomerDao.java b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/CustomerDao.java
index b5df914..db45dc4 100644
--- a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/CustomerDao.java
+++ b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/CustomerDao.java
@@ -16,7 +16,7 @@
 
 package android.arch.persistence.room.integration.testapp.database;
 
-import android.arch.paging.LivePagedListProvider;
+import android.arch.paging.DataSource;
 import android.arch.persistence.room.Dao;
 import android.arch.persistence.room.Insert;
 import android.arch.persistence.room.Query;
@@ -44,12 +44,11 @@
     void insertAll(Customer[] customers);
 
     /**
-     * @return LivePagedListProvider of customers, ordered by last name. Call
-     * {@link LivePagedListProvider#create(Object, android.arch.paging.PagedList.Config)} to
-     * get a LiveData of PagedLists.
+     * @return DataSource.Factory of customers, ordered by last name. Use
+     * {@link android.arch.paging.LivePagedListBuilder} to get a LiveData of PagedLists.
      */
     @Query("SELECT * FROM customer ORDER BY mLastName ASC")
-    LivePagedListProvider<Integer, Customer> loadPagedAgeOrder();
+    DataSource.Factory<Integer, Customer> loadPagedAgeOrder();
 
     /**
      * @return number of customers
diff --git a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/LastNameAscCustomerDataSource.java b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/LastNameAscCustomerDataSource.java
index 1bc731a..a38d6ae 100644
--- a/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/LastNameAscCustomerDataSource.java
+++ b/room/integration-tests/testapp/src/main/java/android/arch/persistence/room/integration/testapp/database/LastNameAscCustomerDataSource.java
@@ -15,6 +15,7 @@
  */
 package android.arch.persistence.room.integration.testapp.database;
 
+import android.arch.paging.DataSource;
 import android.arch.paging.KeyedDataSource;
 import android.arch.persistence.room.InvalidationTracker;
 import android.support.annotation.NonNull;
@@ -32,10 +33,19 @@
     private final InvalidationTracker.Observer mObserver;
     private SampleDatabase mDb;
 
+    public static Factory<String, Customer> factory(final SampleDatabase db) {
+        return new Factory<String, Customer>() {
+            @Override
+            public DataSource<String, Customer> create() {
+                return new LastNameAscCustomerDataSource(db);
+            }
+        };
+    }
+
     /**
      * Create a DataSource from the customer table of the given database
      */
-    public LastNameAscCustomerDataSource(SampleDatabase db) {
+    private LastNameAscCustomerDataSource(SampleDatabase db) {
         mDb = db;
         mCustomerDao = db.getCustomerDao();
         mObserver = new InvalidationTracker.Observer("customer") {