Example1.kt

package com.github.luks91.streamnesting

import android.util.Log
import io.reactivex.Flowable
import io.reactivex.schedulers.Schedulers
import io.reactivex.subscribers.DefaultSubscriber

data class Contact(val firstName: String, 
                   val lastName: String, 
                   val phoneNumber: String)

class Repository {
    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)
                }
            })

    fun <T> openConnection(): Connection<T> {
        Log.i("TAG", "Opening database...")
        Thread.sleep(2000)
        return Connection()
    }

    class Connection<in T>: AutoCloseable {
        fun write(data: T) {
            Thread.sleep(10)
            Log.i("TAG", "Saving: $data")
        }

        override fun close() {
            Log.i("TAG", "Closing database...")
            Thread.sleep(2000)
        }
    }
}