vvsevolodovich

private fun attemptLoginRx() {
	showProgress(true)
	apiClient.login(auth)
			.flatMap { 
				user -> apiClient.getRepositories(user.repos_url, auth) 
			}
			.map { 
				list -> list.map { it.full_name } 
			}

vvsevolodovich

// new SingleFlatMap()
val flatMap = apiClient.login(auth)
		.flatMap { apiClient.getRepositories(it.repos_url, auth) }
// new SingleMap
val map = flatMap
		.map { list -> list.map { it.full_name } }
// new SingleSubscribeOn
val subscribeOn = map
		.subscribeOn(Schedulers.io())
// new SingleObserveOn

vvsevolodovich

apiClient.login(auth)
	// some code ommitted
	.subscribeOn(Schedulers.io())
	.observeOn(AndroidSchedulers.mainThread())

vvsevolodovich

    @SchedulerSupport(SchedulerSupport.NONE)
    @Override
    public final void subscribe(SingleObserver<? super T> subscriber) {
        ObjectHelper.requireNonNull(subscriber, "subscriber is null");

        subscriber = RxJavaPlugins.onSubscribe(this, subscriber);

        ObjectHelper.requireNonNull(subscriber, "subscriber returned by the RxJavaPlugins hook is null");

        try {

vvsevolodovich

override fun login(auth: Authorization): Single<GithubUser> = Single.fromCallable {
	val response = get("https://api.github.com/user", auth = auth)
	if (response.statusCode != 200) {
		throw RuntimeException("Incorrect login or password")
	}

	val jsonObject = response.jsonObject
	with(jsonObject) {
		return@with GithubUser(getString("login"), getInt("id"),
				getString("repos_url"), getString("name"))

vvsevolodovich

    @Override
    protected void subscribeActual(final SingleObserver<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s, source);
        s.onSubscribe(parent);

        Disposable f = scheduler.scheduleDirect(parent);

        parent.task.replace(f);

    }

vvsevolodovich

ObserveOnSingleObserver(SingleObserver<? super T> actual, Scheduler scheduler) {
  this.actual = actual;
  this.scheduler = scheduler;
}

@Override
public void onSuccess(T value) {
  this.value = value;
  Disposable d = scheduler.scheduleDirect(this);
  DisposableHelper.replace(this, d);

vvsevolodovich

    fun connectionObserveration(context: Context): Disposable? {
        if ((firstObserver == null || firstObserver!!.isDisposed) && (secondObserver == null || secondObserver!!.isDisposed())) {
            secondObserver = ReactiveNetwork.observeNetworkConnectivity(context)
                    .subscribeOn(AndroidSchedulers.mainThread())
                    .filter(ConnectivityPredicate.hasState(NetworkInfo.State.CONNECTED, NetworkInfo.State.DISCONNECTED))
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe({ connectivity ->
                        if (connectivity.state == NetworkInfo.State.CONNECTED) {
                            firstObserver = ReactiveNetwork.observeInternetConnectivity()
                                    .subscribeOn(AndroidSchedulers.mainThread())

vvsevolodovich

publishSubject
	.debounce(300, TimeUnit.MILLISECONDS)
	.distinctUntilChanged()
	.switchMap { 
		searchQuery -> apiClientRxImpl.searchRepositories(searchQuery) 
	}
	.subscribeOn(Schedulers.io())
	.observeOn(AndroidSchedulers.mainThread())
	.subscribe({
		repos.adapter = ReposAdapter(

vvsevolodovich

launch {
	broadcast.consumeEach {
		delay(300)
		val foundRepositories = 
				apiClient.searchRepositories(it).await()
		repos.adapter = ReposAdapter(
							foundRepositories.map { it.full_name },
							this@RepositoriesActivity
		)
	}