I tested a little bit and ended up with this versi...

# rx

[bodo]

I tested a little bit and ended up with this version:

Observable.using({ Socket(InetAddress.getByName(ipAddress), port) },
        { socket ->
          Observable.using({ InputStreamReader(socket.inputStream) }, { isr ->
            Observable.using({ BufferedReader(isr) }, { br ->
              Observable.create(ObservableOnSubscribe<String> {
                var read: String = br.readLine()
                while (read.isNotBlank()) {
                  it.onNext(read)
                  read = br.readLine()
                }
              }).filter { it.startsWith(indicator) }
                  .map { it.replace(indicator, "").trim().toInt() }
            }, { br -> br.close() })
          }, { isr -> isr.close() })
        },
        { socket -> socket?.close() })
        .subscribeOn(<http://Schedulers.io|Schedulers.io>())
        .observeOn(AndroidSchedulers.mainThread())