How to establish a connection to the server using RxJava?

Question

Java

I want to correctly connect to the server using RxJava. It seems to me that I need to change the implementation, but I don't know how. I use Single. I'm especially interested in where it is necessary to close ports and sockets. Tell me how to improve the architecture?) Any advice is welcome.

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

Something like this, only first release all the resources, and only then throw onSuccess.
Well, it’s completely incomprehensible what the hell is happening with some kind of mutable state. This is best avoided.
For example, you can do this - instead of connectToServer, which returns nothing, but mutates the state, you can make a function that will not mutate anything, but will return Observable states (Loading/Error/Success/etc). To this function there will be a subscription and already processing of states.