Android 中带有 Socket.io 的 RxKotlin

Question

我想要做的是监听套接字数据并将其转换为我的 UI 可以订阅此事件并在 UI 上进行更改的可观察字符串

到目前为止，我创建了一个类 SocketConnection，在 dagger 连接中维护正确并接收到数据并能够正确处理接口，但想与 rxkotlin 一起应用。

使用 Socket.io,kotlin

SocketConnection 类

 class SocketConnection : SocketStreamListener {

    private var socket: Socket? = null

   var responseSocket :ResponseHandler?= null
    companion object {
        var instance = SocketConnection()

    }

    override fun createSocket(socketQuery: SocketQuery): Socket? {
        try {
            val okHttpClient = UnsafeOkHttpClient.getUnsafeOkHttpClient()
            IO.setDefaultOkHttpWebSocketFactory(okHttpClient)
            IO.setDefaultOkHttpCallFactory(okHttpClient)
            val opts = IO.Options()
            opts.reconnection = false
            opts.callFactory = okHttpClient
            opts.webSocketFactory = okHttpClient
            opts.query = "userID=" + socketQuery.userID + "&token=" + socketQuery.token
            socket = IO.socket(CommonContents.BASE_API_LAYER, opts)
            L.d("Socket object created")
        } catch (e: URISyntaxException) {
            L.e("Error creating socket", e)
        }
        return socket
    }

    override fun createSocketListener(socket: Socket) {
        L.d("inside the socket Listner")
        socket.connect()?.on(Socket.EVENT_CONNECT, {
            L.d("connected")
            listenSocketEvents()
            //socketDataListener()
            createMessageListener()

        })?.on(Socket.EVENT_DISCONNECT,
                {
                    L.d("disconnected")
                    return@on
                })

    }


    /**
     * function used to listen a socket chanel data
     */
    private fun listenSocketEvents() {


       /* socket?.on("1502", { args ->
      // This Will Work 
             L.d("Socket market depth event successfully")
            val socketData = args[0] as String
            L.d(socketData)
         //   instance.data = Observable.just(socketData)
            //data!!.doOnNext({ socketData })

            *//*
            data = args[0] as String
             for (i in 0 until arr.size) {
                 arr[i].socketStreamingData(data)
             }*//*

        })*/

    }

// This Will Not Work
    fun socketDataListener(): Observable<String>{
      return Observable.create({
          subscibe ->
         // L.d("Socket market depth event successfully")
            socket?.on("1502", { args ->
                L.d("Socket market depth event successfully")
                val socketData = args[0] as String
                subscibe.onNext(socketData)
            })

        })
    }

  }

存储库

fun getSocketData(): Observable<String> {
   // L.e("" + SocketConnection.instance.socketDataListener())
    return SocketConnection.instance.createMessageListener()
}

视图模型

fun getSocketData(): Observable<String>{
    return groupRepository.getSocketData()
}

OnFragment (UI)

private fun getSocketUpdate(){
    subscribe(watchlistViewModel.getSocketData()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({
                L.d("SocketData :  " + it.count())

            }, {
                L.e("Error")
            }))
}

在这个 UI 中使用一次性订阅方法进入基类。

请提前告诉我我做错了什么

Answer 1

我建议不要在每次发送消息时都创建Observable，而是使用Subject，因为它具有与套接字连接相似的“性质”。

   val subject = PublishSubject.create<String>()
   ...

    fun listenSocketEvents() {
      socket?.on("1502") { args ->
        val socketData = args[0] as String
        subject.onNext(socketData)
      }
    }

    fun observable(): Observable<String>{
        return subject
    }

然后您可以通过（不包括存储库层等，您必须自己做）来收听主题的更改

private fun getSocketUpdate() {
    disposable = socketConnection.observable()
            .subscribeOn(Schedulers.io())
            .observeOn(...)
            .subscribe({...}, {...})
}

附带说明，您的单例实例不是您在 kotlin 中的做法。 而不是在companion object 中包含instance 字段，您应该将类​​声明为object SocketConnection。
这将自动为您提供所有单例功能。 （我不知道将单例与 socket.io 一起使用是否明智，但我假设您知道自己在做什么 :-)）