Shared observable test

Test.kt

import io.reactivex.Observable
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnit.SECONDS
import java.util.concurrent.atomic.AtomicInteger

val counter = AtomicInteger()
val source = Observable
	.fromCallable {
			println("\nCalled")
			val number = counter.incrementAndGet()
			if (number == 2) throw RuntimeException("test exception")
			number
	}
		.delaySubscription(5L, SECONDS)
		.subscribeOn(Schedulers.single())
		.observeOn(Schedulers.single())
		.doOnSubscribe { println("\nSubscribe source") }
		.doOnDispose { println("\nDispose source") }
		.doOnTerminate { println("\nTerminate source") }
		.share()
		.doOnSubscribe { println("\nSubscribe shared") }
		.doOnDispose { println("\nDispose shared") }
		.doOnTerminate { println("\nTerminate shared") }

source.subscribe({ println("\n#1: $it") }, { println("\n#1: Error")})
source.subscribe({ println("\n#2: $it") }, { println("\n#2: Error")})

Thread {
	Thread.sleep(3000L)
	source.subscribe({ println("\n#3: $it") }, { println("\n#3: Error")})
	Thread.sleep(1000L)
	source.subscribe({ println("\n#4: $it") }, { println("\n#4: Error")})
}.start()