Fancy Kotlin coroutine usage – Advanced Paging 3 source code

More than technology, the article has material, concerned about the public number nine heart said, a high quality article every week, and nine heart in Dachang road side by side.

preface

In my previous article, I briefly discussed the usage of Paging 3. Some students immediately said that Paging 3 is a pagination library.

In fact, Paging 3 appears to be a Paging library, but in reality you only see the appearance. Removing the cloak of Paging, Paging can help us control the life cycle of data, from acquisition to display, providing an automated and standardized control process. Plus other components in Android Jetpack:

• Such asLiveDataCan solve our component life cycle problems
• Such asRoomThe provided PagingSource is refreshed in the UI as soon as new data is added to the database

Don’t ask, ask a word, sweet ~

As we all know, the best way to study source code is to study it with questions in mind.

So let’s start with Paging 3 and de-cloak it to see how it handles Paging:

1. What does the whole process of paging look like?
2. How do PagingSource and RemoteMediator work together?
3. How to implement its state management?

directory

First, look at the structure from the perspective of use

The code used will not be described, but you can read Android Jetpack-Paging 3 to learn how to Use it.

From the previous article, we learned that there are several important classes:

class	instructions
PagingSource	Data source for Paing.
RemoteMediator	If you have both local and remote data sources, this is where you can use themPagingSourceAct as a local data source,RemoteMediatorAct as a remote data source.
Pager	An entry point for data, which can be provided externallyFlowIt is a responsive data source that can be retrieved by the receiverPagingData.
PagingData	We’re not going to get to that, but it’s worth knowing. The official note saysContainer for Paged data from a single generation of loads.single generationI understand that when the data source does not change, it is a generation, that is, there is no increase, deletion or change.
RecyclerView	Our old friend needs no introduction
PagingDataAdapter	The Paging to 3RecyclerViewTailor-made adapters

From these classes, first establish a general structure diagram:

Notice that this PagingAdapter is connected to the Flow process. As I said earlier, PagingData provides a generation of data, you can think of it as a snapshot of the data, and when the data source changes, It will generate a new PagingSource.

Before reading this article, you should have some knowledge of coroutines. I recommend reading my article, Kotlin – Coroutines as You Learn.

Second, analysis preparation

Before moving on, I feel it is important to discuss the state management mechanism and event consumption mechanism in Paging 3.

1 Status management and event management

1.1 Status Management

Status management scenarios:

data class LoadStates(
    /** [LoadState] corresponding to [LoadType.REFRESH] loads. */
    val refresh: LoadState,
    /** [LoadState] corresponding to [LoadType.PREPEND] loads. */
    val prepend: LoadState,
    /** [LoadState] corresponding to [LoadType.APPEND] loads. */
    val append: LoadState
)
Copy the code

There are three status management scenarios:

1. refresh: Refresh state, as the name impliesPagerThe state of the initialization data.
2. append: The state of backloading, common in loading more scenarios.
3. prepend: A state in which data is loaded forward, usually starting at a point where data has not been loaded before.

So given these three scenarios, what are the states of it? The corresponding class for state is LoadState, which provides us with three states:

1. NotLoading: not loaded, and not loaded is divided into load completed and load unfinished, by member variableendOfPaginationReachedControl.
2. Loading: in the load
3. ErrorError:

With these states, we can do more things like interact with the UI and manage request events, for example.

1.2 Event Management

In addition to state management, we also need event management, so for example, if the data comes back, I need to notify an Insert event and include the state change, so event management actually includes state management.

There are also three types of event management:

internal sealed class PageEvent<T : Any> { // Intentional to prefer Refresh, Prepend, Append constructors from Companion. @Suppress("DataClassPrivateConstructor") data class Insert<T : Any> private constructor( val loadType: LoadType, val pages: List<TransformablePage<T>>, val placeholdersBefore: Int, val placeholdersAfter: Int, val combinedLoadStates: CombinedLoadStates ) : PageEvent<T>() { // ... } data class Drop<T : Any>( val loadType: LoadType, val minPageOffset: Int, val maxPageOffset: Int, val placeholdersRemaining: Int ) : PageEvent<T>() { // ... } data class LoadStateUpdate<T : Any>( val loadType: LoadType, val fromMediator: Boolean, val loadState: LoadState // TODO: consider using full state object here ) : PageEvent<T>() { //... }}Copy the code

Concrete is:

1. Insert: Insert events, including specific datapagesRefresh\Append\Prepend; Refresh\Append\PrependloadTypeAnd combined statecombinedLoadStates(including Refresh\Append\Prepend loading state).
2. Drop: Delete event
3. LoadStateUpdate: Events that load state changes.

Third, data generation

With that in mind, let’s go through the process.

The entry is when the PagingAdapter establishes a relationship with the Pager:

lifecycleScope.launch {
    // Note: collectLatest is required. If you only use collect, the filter will not take effect
    viewModel.shoes.collectLatest {
        adapter.submitData(it)
    }
}
Copy the code

Let’s talk about how the data is generated.

Viewmodel. shoes is a Flow provided by Pager:

// Build the Flow code when used
Pager(config = PagingConfig(
    pageSize = 20,
    enablePlaceholders = false,
    initialLoadSize = 20), pagingSourceFactory = { brand? .let { shoeRepository.getShoesByBrandPagingSource(it) } ? : shoeRepository.getAllShoesPagingSource() }).flowclass Pager<Key : Any, Value : Any>
@JvmOverloads constructor(
    config: PagingConfig,
    initialKey: Key? = null.@OptIn(ExperimentalPagingApi::class)
    remoteMediator: RemoteMediator<Key, Value>? = null,
    pagingSourceFactory: () -> PagingSource<Key, Value>
) {
    /** * A cold [Flow] of [PagingData], which emits new instances of [PagingData] once they become * invalidated by [PagingSource.invalidate] or calls to [AsyncPagingDataDiffer.refresh] or * [PagingDataAdapter.refresh]. */
    val flow: Flow<PagingData<Value>> = PageFetcher(
        pagingSourceFactory,
        initialKey,
        config,
        remoteMediator
    ).flow
}
Copy the code

The first paragraph is the code in which we use Paging 3. Without further ado, it is listed just so you know that it has this process.

The second paragraph is the source code for Pager, which is really just a shell, and in the constructor, the last argument is a closure that returns a PagingSource. In addition, Pager provides a Flow of type Flow > from the Flow in PageFetcher.

1. PageFetcher

The Flow section of Pager looks like this:

internal class PageFetcher<Key : Any, Value : Any>(
    / /... Construction parameter omission
) {
    // The Channel used to control the refresh
    private val refreshChannel = ConflatedBroadcastChannel<Boolean> ()// Failed retry Channel
    private val retryChannel = ConflatedBroadcastChannel<Unit> ()// The object built by paging builder can maintain the scope so that on rotation we don't stop
    // the paging.
    val flow: Flow<PagingData<Value>> = channelFlow {
      	Build RemoteMediatorAccessor
        valremoteMediatorAccessor = remoteMediator? .let { RemoteMediatorAccessor(this, it)
        }
        // 2. Change refreshChannel to Flow
        refreshChannel.asFlow()
            .onStart {
        		// 3. The operation triggered before collect
                @OptIn(ExperimentalPagingApi::class)emit(remoteMediatorAccessor? .initialize() == LAUNCH_INITIAL_REFRESH) } .scan(null) {
                // 4. We can deal with the old result before calculating the new one
                // ...
            }
            .filterNotNull()
            .mapLatest { generation ->
              	// 5. Build PagingData by processing only the latest values
                // ...
            }
            .collect { send(it) }
    }
  	// ... 
}
Copy the code

In view of the long, omitted a lot of code, the specific method to put the code.

1.1 Two Attributes

So let’s look at the two properties refreshChannel and retryChannel, which are actually used to send signals, refreshChannel is really important, it sends a refresh signal, and retryChannel sends a rerequest signal.

1.2 the outer

The returned Flow is wrapped in a layer of channelFlow, which is used either to transfer data across multiple coroutines or to have an indeterminate amount of data, which we’ll see later.

1.3 Building remote Data sources

Build a RemoteMediator Accessor that wraps the RemoteMediator for the remote data source.

Later, we will take each extension method of Flow as a part. When it comes to the specific extension method, we will not talk about its principle, but only its function. Interested students can have a look at its implementation.

1.4 create a Flow

RefreshChannel is converted to Flow, and the Flow#onStart method is called, which is called before the Flow can collect operation. What does this method do? Just one line of code:

remoteMediatorAccessor? .initialize() == LAUNCH_INITIAL_REFRESHCopy the code

RemoteMediator Accessor is a shell of RemoteMediator.

// The first call point
valremoteMediatorAccessor = remoteMediator? .let { RemoteMediatorAccessor(this, it)
}

/ / RemoteMediatorAccessor method
internal fun <Key : Any, Value : Any> RemoteMediatorAccessor(
    scope: CoroutineScope,
    delegate: RemoteMediator<Key, Value>
): RemoteMediatorAccessor<Key, Value> = RemoteMediatorAccessImpl(scope, delegate)

private class RemoteMediatorAccessImpl<Key : Any, Value : Any>(
    private val scope: CoroutineScope,
    private val remoteMediator: RemoteMediator<Key, Value>
) : RemoteMediatorAccessor<Key, Value> {
    // ...

    override suspend fun initialize(a): RemoteMediator.InitializeAction {
        return remoteMediator.initialize().also { action ->
            if (action == RemoteMediator.InitializeAction.LAUNCH_INITIAL_REFRESH) {
  				If RemoteMediator has default initialization behavior before collect, set the status
                accessorState.use {
                    it.setBlockState(LoadType.APPEND, REQUIRES_REFRESH)
                    it.setBlockState(LoadType.PREPEND, REQUIRES_REFRESH)
                }
            }
        }
    }

    // ...
}
Copy the code

From the code I’ve listed, RemoteMediatorAccessor wraps RemoteMediator, Remotemediatoraccessimply #initialize also calls the RemoteMediator#initialize method, which returns an enumeration InitializeAction of two types:

1. LAUNCH_INITIAL_REFRESH: Emits a refresh signal during initialization
2. SKIP_INITIAL_REFRESH: does not emit a refresh signal during initialization, waiting for the UI request to send

Going back to the flow in PageFetcher, you can see that when you return to the onStart method, it has two cases:

1. If you haveRemoteMediatorBy default, it will launchtrue.
2. There is noRemoteMediatorOr initialize the remote data source by default without requesting itfalse.

It can be interpreted as whether it wants to refresh the remote data source at initialization.

1.5 Scan

The function of this method is that every time the flow emits a new signal, you can get the new signal and calculate the new result. Before that, you can also get the old result for processing the old result.

As you can see from the parameters of this method:

• previousGeneration: The result of the last calculation
• triggerRemoteRefresh: Mentioned aboveonStartMethod, or called elsewhererefreshChannelThe emitted signal triggers a refresh of the remote data source.

internal class PageFetcher<Key : Any, Value : Any>(
    private val pagingSourceFactory: () -> PagingSource<Key, Value>,
    private valinitialKey: Key? .private val config: PagingConfig,
    @OptIn(ExperimentalPagingApi::class)
    private val remoteMediator: RemoteMediator<Key, Value>? = null
) {
    // ... 

    // The object built by paging builder can maintain the scope so that on rotation we don't stop
    // the paging.
    val flow: Flow<PagingData<Value>> = channelFlow {
        valremoteMediatorAccessor = remoteMediator? .let { RemoteMediatorAccessor(this, it)
        }
        refreshChannel.asFlow()
            .onStart {
                @OptIn(ExperimentalPagingApi::class)emit(remoteMediatorAccessor? .initialize() == LAUNCH_INITIAL_REFRESH) } .scan(null) { previousGeneration: PageFetcherSnapshot<Key, Value>? , triggerRemoteRefresh ->1. Generate a new data source
                varpagingSource = generateNewPagingSource(previousGeneration? .pagingSource)while(pagingSource.invalid) { pagingSource = generateNewPagingSource(previousGeneration? .pagingSource) }@OptIn(ExperimentalPagingApi::class)
                valinitialKey: Key? = previousGeneration? .refreshKeyInfo() ? .let { pagingSource.getRefreshKey(it) } ? : initialKey// 2. Release the old data sourcepreviousGeneration? .close()// 3. Generate new PageFetcherSnapshot
                PageFetcherSnapshot<Key, Value>(
                    initialKey = initialKey,
                    pagingSource = pagingSource,
                    config = config,
                    retryFlow = retryChannel.asFlow(),
                    // Only trigger remote refresh on refresh signals that do not originate from
                    // initialization or PagingSource invalidation.
                    triggerRemoteRefresh = triggerRemoteRefresh,
                    remoteMediatorConnection = remoteMediatorAccessor,
                    invalidate = this@PageFetcher::refresh
                )
            }
            .filterNotNull()
            .mapLatest { generation ->
                // ...
            }
            .collect { send(it) }
    }
    / /...
}
Copy the code

This method does three things:

1. Generate a new data sourcePageSource:PageFetcher#generateNewPagingSourceThis method is calledPageFetcherConstructorpagingSourceFactoryA new data source is created and some listening is done.
2. Release old data sources.
3. Return a new onePageFetcherSnapshotObject, which is the holding class for the data snapshot.

1.6 Filter null values

The Flow#filterNotNull method filters incoming empty values.

1.7 Processing the latest value

Flow#mapLatest only processes the latest value. While this method is working, a new value is sent upstream, and it stops working and processes the new value.

internal class PageFetcher<Key : Any, Value : Any>(
    private val pagingSourceFactory: () -> PagingSource<Key, Value>,
    private valinitialKey: Key? .private val config: PagingConfig,
    @OptIn(ExperimentalPagingApi::class)
    private val remoteMediator: RemoteMediator<Key, Value>? = null
) {
    // ... 

    // The object built by paging builder can maintain the scope so that on rotation we don't stop
    // the paging.
    val flow: Flow<PagingData<Value>> = channelFlow {
        valremoteMediatorAccessor = remoteMediator? .let { RemoteMediatorAccessor(this, it)
        }
        refreshChannel.asFlow()
            .onStart {
                @OptIn(ExperimentalPagingApi::class)emit(remoteMediatorAccessor? .initialize() == LAUNCH_INITIAL_REFRESH) } .scan(null) {
                // ...
            }
            .filterNotNull()
            .mapLatest { generation ->
                val downstreamFlow = if (remoteMediatorAccessor == null) {
                    generation.pageEventFlow
                } else {
                    generation.injectRemoteEvents(remoteMediatorAccessor)
                }
                PagingData(
                    flow = downstreamFlow,
                    receiver = PagerUiReceiver(generation, retryChannel)
                )
            }
            .collect { send(it) }
    }
    / /...
}
Copy the code

In the Flow#mapLatest method, it does two things:

1. You get a stream of eventspageEventFlow.
2. We’re going to encapsulate this stream of eventsPagingData.

1.8 send PagingData

Send the PagingData obtained above, which will eventually be consumed by the PagingDataAdapter, back to the code we wrote at the beginning:

// viewmodel. shoes is Flow
      
       >
      
viewModel.shoes.collectLatest {
	adapter.submitData(it)
}
Copy the code

To sum up, although the above process is many, in fact, the purpose is:

1. getPagingDataAnd thePagingDataThe most important thing is the flow of eventsFlow<PageEvent<T>>, it comes fromPageFetcherSnapshot.
2. Depending on whether the code is enabledRemoteMediator.

2 PagingData

From 1, we learned that the event Flow > comes from PageFetcherSnapshot, which is the core code related to data.

Boy, another big chunk of code, and the most important one is pageEventFlow:

internal class PageFetcherSnapshot<Key : Any, Value : Any>(
    internal valinitialKey: Key? .internal val pagingSource: PagingSource<Key, Value>,
    private val config: PagingConfig,
    private val retryFlow: Flow<Unit>,
    private val triggerRemoteRefresh: Boolean = false.val remoteMediatorConnection: RemoteMediatorConnection<Key, Value>? = null.private val invalidate: () -> Unit= {{})// ...
    @OptIn(ExperimentalCoroutinesApi::class)
    private val pageEventChCollected = AtomicBoolean(false)
    private val pageEventCh = Channel<PageEvent<Value>>(Channel.BUFFERED)
    private val stateLock = Mutex()
    private val state = PageFetcherSnapshotState<Key, Value>(
        config = config
    )
    private val pageEventChannelFlowJob = Job()

    @OptIn(ExperimentalCoroutinesApi::class)
    val pageEventFlow: Flow<PageEvent<Value>> = cancelableChannelFlow(pageEventChannelFlowJob) {
        // 1. Create a coroutine pageEventCh to send the received event
        launch {
            pageEventCh.consumeAsFlow().collect {
                // Protect against races where a subsequent call to submitData invoked close(),
                // but a pageEvent arrives after closing causing ClosedSendChannelException.
                try {
                    send(it)
                } catch (e: ClosedSendChannelException) {
                    // Safe to drop PageEvent here, since collection has been cancelled.}}}// 2. Accept the retry information, is not for caching
        val retryChannel = Channel<Unit>(Channel.RENDEZVOUS)
        launch { retryFlow.collect { retryChannel.offer(it) } }

        // 3. Retry action
        launch {
            retryChannel.consumeAsFlow()
                .collect {
                    // Process the corresponding status after retry
                    // ...}}// 4. If you need to update remotely while refreshing, let remoteMediator load the data
        if(triggerRemoteRefresh) { remoteMediatorConnection? .let {val pagingState = stateLock.withLock { state.currentPagingState(null) }
                it.requestLoad(LoadType.REFRESH, pagingState)
            }
        }

        // 5. PageSource initializes data
        doInitialLoad(state)

        // 6. Consumer hint
        if (stateLock.withLock { state.sourceLoadStates.get(LoadType.REFRESH) } !is LoadState.Error) {
            startConsumingHints()
        }
    }

    // ...
}
Copy the code

PageEventFlow is divided into 6 parts. We focus on 1, 4, 5 and 6.

2.1 launch PageEvent

Create a coroutine to forward PageEvent

received by pageEventCh.

2.2 Requesting a Remote data source

If you create a remote data source and need to load remote data at initialization, start requesting remote data,

2.3 PagingSource initialization

There’s the first data initialization of the PagingSource, so what happens?

internal class PageFetcherSnapshot<Key : Any, Value : Any>(
    // ...
) {
    // ...

    private suspend fun doInitialLoad(
        state: PageFetcherSnapshotState<Key, Value>
    ) {
        // 1. Set the current loading status - refresh
        stateLock.withLock { state.setLoading(LoadType.REFRESH) }

        // Build parameters
        val params = loadParams(LoadType.REFRESH, initialKey)
        // 2. Load data and get result
        when (val result = pagingSource.load(params)) {
            is PagingSource.LoadResult.Page<Key, Value> -> {
                // 3
                val insertApplied = stateLock.withLock { state.insert(0, LoadType.REFRESH, result) }

                // 4. Handle the various states
                stateLock.withLock {
                    state.setSourceLoadState(LoadType.REFRESH, LoadState.NotLoading.Incomplete)
                    if (result.prevKey == null) {
                        state.setSourceLoadState(
                            type = PREPEND,
                            newState = when (remoteMediatorConnection) {
                                null -> LoadState.NotLoading.Complete
                                else -> LoadState.NotLoading.Incomplete
                            }
                        )
                    }
                    if (result.nextKey == null) {
                        state.setSourceLoadState(
                            type = APPEND,
                            newState = when (remoteMediatorConnection) {
                                null -> LoadState.NotLoading.Complete
                                else -> LoadState.NotLoading.Incomplete
                            }
                        )
                    }
                }

                // 5. Send PageEvent
                if (insertApplied) {
                    stateLock.withLock {
                        with(state) {
                            pageEventCh.send(result.toPageEvent(LoadType.REFRESH))
                        }
                    }
                }

                // 6. Whether the remote data request is necessary
                if(remoteMediatorConnection ! =null) {
                    if (result.prevKey == null || result.nextKey == null) {
                        val pagingState =
                            stateLock.withLock { state.currentPagingState(lastHint) }

                        if (result.prevKey == null) {
                            remoteMediatorConnection.requestLoad(PREPEND, pagingState)
                        }

                        if (result.nextKey == null) {
                            remoteMediatorConnection.requestLoad(APPEND, pagingState)
                        }
                    }
                }
            }
            is PagingSource.LoadResult.Error -> stateLock.withLock {
                // Error status request
                val loadState = LoadState.Error(result.throwable)
                if (state.setSourceLoadState(LoadType.REFRESH, loadState)) {
                    pageEventCh.send(PageEvent.LoadStateUpdate(LoadType.REFRESH, false, loadState))
                }
            }
        }
    }
}
Copy the code

From the first time the data is initialized, you can see many things:

1. Changes in data loading status: Refresh scenarioIncomplete — Loading– Set based on the returned resultComplete 和 IncompleteAnd some of the states will pass through the first partpageEventChSends a status update event.
2. inRefreshThe settingLoadingAfter the state, the loaded parameters are built into thepageSourceMake a data request and finally seepagingSource.
3. becausepagingSource.load(params)You can get one of two things, if it’s an error you just handle the error.
4. If it is a normal result, it will process the result first. Change the state again, and then launch one at a timeInsertEvents.
5. Because sometimespageSourceFailed to get result, set againremoteMediatorAt this time, you need to use it againremoteMediatorProceed to the next data request

This is the time to answer the first question:

If a remoteMediator exists, the data request will be made using the remoteMediator when the pageSource does not get the result.

2.4 How Can I Load More Data

If the first flush doesn’t go wrong, that is, the first flush doesn’t go wrong, the startConsumingHints method is called next:

internal class PageFetcherSnapshot<Key : Any, Value : Any>( // ... ) {/ /... private val state = PageFetcherSnapshotState<Key, Value>( config = config ) @OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class) private fun CoroutineScope.startConsumingHints() { // ... / / to monitor the Prepend news launch {state. ConsumePrependGenerationIdAsFlow () collectAsGenerationalViewportHints (the Prepend)} / / Listen to Append the message launch {state. ConsumeAppendGenerationIdAsFlow () collectAsGenerationalViewportHints (Append)}} private suspend fun Flow<Int>.collectAsGenerationalViewportHints( loadType: LoadType) = flatMapLatest {generationId -> statelock. withLock {//... } @OptIn(FlowPreview::class) hintChannel.asFlow() .drop(if (generationId == 0) 0 else 1) .map { hint -> GenerationalViewportHint(generationId, hint) } } .runningReduce { previous, next -> if (next.shouldPrioritizeOver(previous, LoadType)) next else previous}.conflate().collect {generationalHint -> Change status // 2. Request using PageSource // 3. 4. Determine whether to use RemoteMediator doLoad(state, loadType, generationalHint)}}Copy the code

The last part of doLoad is similar to the third part of doInitialLoad, with some differences. For example, when loading data, it determines whether the current data position is less than a threshold (set when PagerConfig is created) from the end or header of the loaded data. More will be loaded when this condition is true.

You might be a little dizzy here, but let’s use a picture to summarize what we talked about earlier:

After reading this section, it seems that there are still some questions that are not clear:

1. How does the Ui drive the loading of more data than the initial part?

Fourth, data consumption

1. Specific consumption behavior

From the data returned upstream, we get PagingData

. Let’s see how the adapter PagingDataAdapter handles this data:

abstract class PagingDataAdapter<T : Any, VH : RecyclerView.ViewHolder> @JvmOverloads constructor(
    diffCallback: DiffUtil.ItemCallback<T>,
    mainDispatcher: CoroutineDispatcher = Dispatchers.Main,
    workerDispatcher: CoroutineDispatcher = Dispatchers.Default
) : RecyclerView.Adapter<VH>() {
    private val differ = AsyncPagingDataDiffer(
        diffCallback = diffCallback,
        updateCallback = AdapterListUpdateCallback(this),
        mainDispatcher = mainDispatcher,
        workerDispatcher = workerDispatcher
    )

    // ...

    suspend fun submitData(pagingData: PagingData<T>) {
        differ.submitData(pagingData)
    }

 	// ...
}
Copy the code

PagingDataAdapter also doesn’t handle it itself, but hands it over to AsyncPagingDataDiffer, as the PagingDataAdapter annotation says, The PagingDataAdapter is just a shell for AsyncPagingDataDiffer, and the AsyncPagingDataDiffer is a hot potato for PagingDataDiffer.

abstract class PagingDataDiffer<T : Any>(
    private val differCallback: DifferCallback,
    private val mainDispatcher: CoroutineDispatcher = Dispatchers.Main
) {
    // ...

    suspend fun collectFrom(pagingData: PagingData<T>) = collectFromRunner.runInIsolation {
        // 1. Assign values to the current receiver
        receiver = pagingData.receiver

        pagingData.flow.collect { event ->
            withContext<Unit>(mainDispatcher) {
                // Switch to main thread
                if (event is PageEvent.Insert && event.loadType == LoadType.REFRESH) {
                    // 1. The current is the insert event and the current is the refresh scenario
                    lastAccessedIndexUnfulfilled = false

                    // 2. Create a new PagePresenter that manages local data
                    val newPresenter = PagePresenter(event)
                    // 3. Recalculate the loading location of the data
                    val transformedLastAccessedIndex = presentNewList(
                        previousList = presenter,
                        newList = newPresenter,
                        newCombinedLoadStates = event.combinedLoadStates,
                        lastAccessedIndex = lastAccessedIndex
                    )
                    presenter = newPresenter

                    // Dispatch LoadState + DataRefresh updates as soon as we are done diffing,
                    // but after setting presenter.
                    dataRefreshedListeners.forEach { listener ->
                        listener(event.pages.all { page -> page.data.isEmpty() })
                    }
                    // 4. Notification status change
                    dispatchLoadStates(event.combinedLoadStates)

                    // 5. If the loading location of data changes, use receiver to send notificationtransformedLastAccessedIndex? .let { newIndex -> lastAccessedIndex = newIndex receiver? .accessHint( newPresenter.viewportHintForPresenterIndex(newIndex) ) } }else {
                    // ...
                }
            }
        }
    }
}
Copy the code

Using Refresh as an example, let’s take a quick look at how to consume data:

1.1 Cache data and notify UI of updates

In the case of Refresh, a data manager is built first, in this case PagePresenter.

Next comes notification data refresh, where the presentNewList method is handed to subclasses to implement:

private val differBase = object : PagingDataDiffer<T>(differCallback, mainDispatcher) { override suspend fun presentNewList( previousList: NullPaddedList<T>, newList: NullPaddedList<T>, newCombinedLoadStates: CombinedLoadStates, lastAccessedIndex: Int) = when {// fast path for no items -> some items previousList. Size == 0 -> {// Notification of new data insertion for the first time differCallback.onInserted(0, newList.size) null } // fast path for some items -> no items newList.size == 0 -> { differCallback.onRemoved(0, previousList.size) null } else -> { val diffResult = withContext(workerDispatcher) { previousList.computeDiff(newList, diffCallback) } previousList.dispatchDiff(updateCallback, newList, diffResult) previousList.transformAnchorIndex( diffResult = diffResult, newList = newList, oldPosition = lastAccessedIndex ) } }Copy the code

The first case is if there is data coming back from the first refresh. Use differCallback directly to notify that data has been added. Of course, this will notify our adapter PagingAdapter to call the corresponding method.

You might wonder a little bit here, but the adapter PagingAdapter doesn’t hold any data, so how does it get the data?

The PagingAdapter overwrites the getItem method, removing layers of nesting, and finally uses PagingDataDiffer:

abstract class PagingDataDiffer<T : Any>( private val differCallback: DifferCallback, private val mainDispatcher: CoroutineDispatcher = Dispatchers.Main ) { // ... operator fun get(@IntRange(from = 0) index: Int): T? { lastAccessedIndexUnfulfilled = true lastAccessedIndex = index receiver? .accessHint(presenter.viewportHintForPresenterIndex(index)) return presenter.get(index) } }Copy the code

Therefore, the getItem method is implemented through the data manager PagePresenter. Otherwise, UiReceiver#accessHint is called every time the data is retrieved. When you still have data to load and the current display location is less than a certain threshold from the end of the data, the doLoad method is triggered, which causes Pager to load more data.

1.2 Notification of data status updates

Back in the PagingDataDiffer#collect method, after processing the above, the status is updated:

abstract class PagingDataDiffer<T : Any>(
    private val differCallback: DifferCallback,
    private val mainDispatcher: CoroutineDispatcher = Dispatchers.Main
) {
    // ...
    private fun dispatchLoadStates(states: CombinedLoadStates) {
        if (combinedLoadStates.snapshot() == states) return

        combinedLoadStates.set(states)
        loadStateListeners.forEach { it(states) }
    }
}
Copy the code

So what can you do with these states? Can handle state with the user, for example, refresh error can switch wrong interface and so on.

1.3 Calculate whether the loading position of data changes

If it is not the first refresh and some data source has changed, such as deleting or adding data, some of the original location information is not accurate, you need to call the receiver. AccessHint method to send notifications.

The non-REFRESH case ends up doing something similar to Refresh without further elaboration.

To summarize the consumption process, use a picture to describe it:

The first and third questions we started with are clear:

The entire Paging 3 process is around the Flow >, through which state and data changes are sent, the UI monitors data through it, and finally notifies the observers of the data and status.

conclusion

After reading the source code of Paging 3, I have two feelings:

• First point: the original coroutine can be used so amazing, the first time I feel that the coroutine I use is not the same thing as the big guy use coroutine, I cry ~

• Second point: message passing through the entire coroutineChannelIn the realization of, combinedFlowTo write it, it’s really more concise and fluid.

In the next article, I will discuss how I used Paging 3 in my starting business.

Thanks for reading, and if you have another comment, feel free to leave a comment below!

If you think this article is good, it is the biggest encouragement for the author ~