Как организовать подключение к серверу используя RxJava?

Question

[Tsudzukeru]

Хочу правильно сделать подключение к серверу используя RxJava. Мне кажется, что нужно изменить реализацию, но не знаю как. Использую Single. Особенно меня интересует - где необходимо закрывать порты и соккеты. Подскажите как улучшить архитектуру?)Любые советы приветствуются.

fun connectToServer() {
        val TAG = "Server answer"
        connectToServerWrapper()
            .subscribeOn(Schedulers.io())
            .subscribeBy (onSuccess = {
                if(it=="200 OK"){
                    _controllerState.postValue(ControllerState.ON)

                    Log.e(TAG, it)
                }
            }, onError = {
                if (it is ConnectException){
                    _controllerState.postValue(ControllerState.NO_INTERNET)
                }
                else {
                    _controllerState.postValue(ControllerState.ERROR)
                }
                Log.e(TAG, it.toString())
            })
            .addTo(compositeDisposable)
    }

    fun connectToServerWrapper(): Single<String> {
        _controllerState.postValue(ControllerState.LOADING)
        return Single.create { emitter ->
            val pw = PrintWriter(Utils.SOCKET.getOutputStream(), true)
            val br = BufferedReader(InputStreamReader(Utils.SOCKET.getInputStream()))
            pw.println("Hello from client 1")
            emitter.onSuccess(br.readLine())
//            pw.close()
//            br.close()
//            socket.close()
        }
    }

Answer 1

[Денис Загаевский]

Примерно так, только сначала все ресурсы освободи, а уже потом кидай onSuccess.
Ну и совсем непонятно, что за хрень происходит с каким-то мутабельным стейтом. Такого лучше избегать.
Например, можно сделать так - вместо connectToServer, которая ничего не возвращает, но зато мутирует стейт, можно сделать функцию, которая ничего не будет мутировать, но зато будет возвращать Observable состояний(Loading/Error/Success/etc). К этой функции будет происходить подписка и уже обработка состояний.