raulh82vlc

public class Pager<I, O> {

    private static final Observable FINISH_SEQUENCE = Observable.never();

    private PublishSubject<Observable<I>> pages;
    private Observable<I> nextPage = finish();
    private Subscription subscription = Subscriptions.empty();

    private final PagingFunction<I> pagingFunction;
    private final Func1<I, O> pageTransformer;

raulh82vlc

    coroutines_android_version = '1.3.2'
    // Coroutines
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutines_android_version"
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:$coroutines_android_version"

raulh82vlc

@Singleton
class NetworkDataSourceImpl @Inject constructor(
        private val twitterApi: TwitterApi,
        private val connectionHandler: ConnectionHandler,
        private val requestsIOHandler: RequestsIOHandler,
        private val taskThreading: TaskThreading
) : NetworkDataSource {}

raulh82vlc

class TaskThreadingImpl @Inject constructor() : TaskThreading {
  private val computationScheduler = Schedulers.computation()
  private val ioScheduler = Schedulers.io()
  private val mainScheduler = AndroidSchedulers.mainThread()

  override fun ui(): Scheduler = mainScheduler
  override fun io(): Scheduler = ioScheduler
  override fun computation(): Scheduler = computationScheduler
}

raulh82vlc

class TaskThreadingImpl @Inject constructor() : TaskThreading {
  // declaration of dispatchers
  private val computationCoroutineDispatcher: CoroutineDispatcher = Default 
  private val ioCoroutineDispatcher: CoroutineDispatcher = IO
  private val mainCoroutineDispatcher: CoroutineDispatcher = Main
			
  // other schedulers here
  // dispatchers:
  override fun uiDispatcher(): CoroutineDispatcher = mainCoroutineDispatcher
  override fun ioDispatcher(): CoroutineDispatcher = ioCoroutineDispatcher

raulh82vlc

override fun search(token: String, query: String)
: Single<Either<Throwable, List<TweetApiModel>>>
= rxSingle {
    val response = twitterApi.search(requestsIOHandler.getTokenFormatted(token), query)
    if (requestsIOHandler.searchIsSuccessful(response)) {
       val tweets = requestsIOHandler.getTweets(response)
       Either.right(tweets)
    } else {
       Either.left(EmptyResponseException())
    }

raulh82vlc

suspend fun search(query: String, token: String)
 :Either<Throwable, List<TweetApiModel>> {
    // a response coming from the Retrofit API
    val response = twitterApi.search(requestsIOHandler.getTokenFormatted(token), query)
    return if (requestsIOHandler.searchIsSuccessful(response)) {
        val tweets = requestsIOHandler.getTweets(response)
        // return a list of tweets wrapped into a Either.Right
        Either.right(tweets)
    } else {
        // return a bad response exception wrapped into an Either.Left

raulh82vlc

@NotNull
Single<List<TweetApiModel>> search(@NotNull String token, @NotNull String query) {
         return twitterApi.search(requestsIOHandler.getTokenFormatted(access.getToken()),query)
                  .subscribeOn(taskThreading.io())
                  .observeOn(taskThreading.computation())
                  .filter(requestsIOHandler::searchHasNoErrorResponse)
                  .map(requestsIOHandler::getSearchContent)
                  .map(requestsIOHandler::getTweets)
                  .flatMapSingle(Single::just));
}

raulh82vlc

@ExperimentalCoroutinesApi
override fun search(token: String, query: String)
: Observable<Either<Throwable, List<TweetApiModel>>>
= flow<Either<Throwable, List<TweetApiModel>>> {
    val response = twitterApi.search(
         requestsIOHandler.getTokenFormatted(token), query)
    if (requestsIOHandler.searchIsSuccessful(response)) {
       val tweets = requestsIOHandler.getTweets(response)
       emit(Either.right(tweets))
    } else {

raulh82vlc

    // Coroutines dependency for convenience backwards compatibility with rx2
    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-rx2:$coroutines_android_version"