JsonReaderObservable.kt

class JsonReaderObservable<T>(elementType: Type, gson: Gson, private val reader: JsonReader) : Observable<T>() {

    private val subject = ReplaySubject.create<T>()

    init {
        reader.beginArray()
        while (reader.hasNext()) {
            subject.onNext(gson.fromJson(reader, elementType))
        }
        reader.endArray()
        subject.onComplete()
    }

    override fun subscribeActual(observer: Observer<in T>) {
        subject.subscribe(observer)
    }

}