Example1_part1.kt

fun save(contacts: Flowable<Contact>) =
    contacts
        .observeOn(Schedulers.io())
        .subscribe(object : DefaultSubscriber<Contact>() {
            override fun onStart() = request(1)

            override fun onNext(contact: Contact) =
                    openConnection<Contact>().use {
                        it.write(contact)
                        request(1)
                    }

            override fun onComplete() {}
            override fun onError(t: Throwable) {
                Log.e("TAG", "Oops!", t)
            }
        })