Periodically refresh Kotlin flow (polling)-CodePudding

How shoud I peridically "refresh" kotlin flow, to get fresh data.

I have list of book id's stored in room database. From network api I get the actual books using the stored ids.

My goal is to listen database changes and map the book ids to actual books. That works fine.

The problem is, how shoud I re-fetch periodically book data from network api but still keep listening the database all the time? So changes in book ids in database shoud allways trigger network api call (booksApi.get()).


fun getBookesFlow(): Flow<List<Book>> {
  return booksDao.all().map { ids -> 
    booksApi.get(ids)
  }
}

fun getPollingFlow(): Flow<List<Book>> {
  return flow {
    while(true) {
      // Reset the flow every 30 seconds
      getBookesFlow().collect { bookes -> // This is just idea, not working code
        emit(books)
      }
      delay(30.seconds)
    }
  }
}

fun getBooks(): {
  viewModelScope.launch {
   getPollingFlow().collect { books ->
    _uiState.update { it.copy( books = books) }
   }
  }
}

CodePudding user response：

You can model this by treating both the database updates and also a 30 second timer as the sources of updates. To combine two flows, use combine.

private val timerFlow = flow {
    while (currentCoroutineContext().isActive)
    {
        emit(Unit)
        delay(30.seconds)
    }
}

fun getPollingFlow(): Flow<List<Book>> =
    timerFlow.combine(booksDao.all()) { _, ids ->
        booksApi.get(ids)
    }.distinctUntilChanged()

CodePudding user response：

We can't force flows to provide values, so polling doesn't really apply to them. Instead, we create flows on top of other flows to provide the behavior we need.

In your case we need a flow that will observe booksDao.all() and then it will re-emit the same data periodically. We can do this easily with transformLatest():

fun getBookesFlow(): Flow<List<Book>> {
  return booksDao.all()
    .transformLatest {
      while (true) {
        emit(it)
        delay(30.seconds)
      }
    }.map { ids -> 
      booksApi.get(ids)
    }
}

Whenever there is a new item to booksDao.all(), we start an infinite loop that emits the same data to the downstream flow. With each new item the previous loop is cancelled, so we only emit the latest.

Even better, we can create a flow operator that does the above:

fun getBookesFlow(): Flow<List<Book>> {
  return booksDao.all()
    .emitLatestPeriodically(30.seconds)
    .map { ids -> 
      booksApi.get(ids)
    }
}

fun <T> Flow<T>.emitLatestPeriodically(interval: Duration): Flow<T> = transformLatest {
    while (true) {
        emit(it)
        delay(interval)
    }
}

If you prefer to not use experimental functions (they're pretty stable though) then you can use flatMapLatest() instead.