raulh82vlc

fun getSearchTweets(query: String)
: Observable<List<Tweet>> =
flow<List<Tweet>> {
      val accessToken = tokenDBDataSource.getAccessToken()?: throw BadTokenException("No token was found")
      val token = accessToken.token
      // retrieve old values from DB
      emit(tweetsDBDataSource.retrieveAllTweetsForTweetsIds(tweetIds))
      // get fresh values from network & saved them into DB
      val eitherResult = networkDataSource.search(token, query)
      val tweets = mapperTweets.map(eitherResult)

raulh82vlc

suspend fun getSearchTweets(query: String): List<Tweet> {
    val accessToken = tokenDBDataSource.getAccessToken()?: throw BadTokenException("No token was found")
    val token = accessToken.token
    val eitherResult = networkDataSource.search(token, query)
    val tweets = mapperTweets.map(eitherResult)
    tweetQueryJoinDBDataSource.deleteTweets(query)
    tweetsDBDataSource.insertTweets(tweets)
    insertQuery(query)
    tweetQueryJoinDBDataSource.insertQueryAndTweetsId(tweetQueryJoinDBDataSource.convert(tweets, query))
    val tweetIds = tweetQueryJoinDBDataSource.retrieveAllTweetsIdAQuery(query)

raulh82vlc

@ExperimentalCoroutinesApi
@FlowPreview           
@InternalCoroutinesApi
fun getSearchTweets(token: String, query: String): Observable<List<Tweet>> {
    return networkDataSource
      .search(token, query)
          .map { either -> mapperTweets.map(either) }
          .onEach {
              tweetQueryJoinDBDataSource.deleteTweets(query)
              tweetsDBDataSource.insertTweets(it)

raulh82vlc

@Singleton
class TweetsRepositoryImpl @Inject constructor(
       private val networkDataSource: NetworkDataSource,
       private val tweetsDBDataSource: TweetDao,
       private val mapperTweets: TweetsNetworkToDBMapperList,
       private val tokenDBDataSource: TokenDao,
       private val queryDBDataSource: QueryDao,
       private val tweetQueryJoinDBDataSource: TweetQueryJoinDao,
       private val mapperToken: TokenNetworkToDBMapper,
       private val taskThreading: TaskThreading

raulh82vlc

// Tweet table, Database DAO's query
@Query("SELECT *
FROM ${Tweet.TABLE_NAME}
WHERE tweet_id IN(:tweetIds) ORDER BY created_at DESC")
suspend fun retrieveAllTweetsForTweetsIds(tweetIds: List<String>): List<Tweet>

raulh82vlc

fun getSearchTweets(token: String, query: String)
: Observable<List<Tweet>>
= networkDataSource.search(token, query)
      .map { either -> mapperTweets.map(either) }
      .doOnNext { tweetsToAdd ->
          tweetQueryJoinDBDataSource.deleteTweets(query)
          tweetsDBDataSource.insertTweets(tweetsToAdd)
          insertQuery(query)
          tweetQueryJoinDBDataSource.insertQueryAndTweetsId(convert(tweetsToAdd, query))
      }.flatMap {

raulh82vlc

// query from the Tweet DAO, or DB data source
@Query("SELECT *
FROM ${Tweet.TABLE_NAME}
WHERE tweet_id IN(:tweetIds) ORDER BY created_at DESC")
fun retrieveAllTweetsForTweetsIdsRx(tweetIds: List<String>): 
Observable<List<Tweet>>

raulh82vlc

private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())

@ExperimentalCoroutinesApi
override fun execute(query: String, callbackInput: SearchCallback?) {
   callback = callbackInput
   callback?.onShowLoader()
   scope.launch {
      repository.searchTweet(query)
        .catch { error ->
          // error handling

raulh82vlc

private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())

@ExperimentalCoroutinesApi
override fun execute(query: String, callbackInput: SearchCallback?) {
    callback = callbackInput
    callback?.onShowLoader()
    
    repository.searchTweet(query)
      .onEach { tweets ->
        // UI actions for each stream

raulh82vlc

private val scope = CoroutineScope(taskThreading.uiDispatcher() + SupervisorJob())

override fun cancel() {
   callback = null
   scope.cancel()
}