RxJava + Websocket - 如何将 Observable 添加到 Websocket 监听器？

Posted 2023-03-22

【中文标题】RxJava + Websocket - 如何将 Observable 添加到 Websocket 监听器？

【英文标题】：RxJava + Websocket - How to add Observable to Websocket listener?

【发布时间】：2021-08-19 07:10:30

【问题描述】：

我有一个 ViewModel 在我的 MainRepo 类中观察 RxJava Observable。我正在尝试让 MainRepo 类中的 WebsocketListener 发出事件，但我不确定该怎么做。

MainRepo 类：

private WebSocket ws;

public void createWsConnection() 
        Request request = new Request.Builder()
                .url(Constants.WEBSOCKET_ENDPOINT)
                .addHeader(Constants.WEBSOCKET_HEADERS_KEY, Constants.USER_ID)
                .build();

        OkHttpClient client = new OkHttpClient
                .Builder()
                .pingInterval(30, TimeUnit.SECONDS)
                .build();

        this.ws = client.newWebSocket(request, webSocketListener);
    

这就是我感到困惑的地方。我不知道如何将 websocket 与 RxJava observable 一起使用。

public Observable<String> createListener()
        return Observable.create(new ObservableOnSubscribe<String>() 
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) 
                 //I don't know what to put here in order to emit messages
                 //back to my ViewModel class using the websocket listener
            
        );
    

websocket 监听器：

 private WebSocketListener webSocketListener = new WebSocketListener() 

        @Override
        public void onOpen(@NotNull WebSocket webSocket, Response response) 
            Timber.d("Ws connection opened...", response.toString());
        

        @Override
        public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) 
            Timber.d("Ws connection closing...");
        

        @Override
        public void onClosed(@NotNull WebSocket webSocket, int code, @NotNull String reason) 
            Timber.d("Ws connection closed...");
        

        @Override
        public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) 
            Timber.d("Ws incoming message.");

        

        @Override
        public void onFailure(@NotNull WebSocket webSocket, @NotNull Throwable t, Response response) 
            Timber.e(t, "Ws connection failure.", response.toString());

        
    ;

ViewModel 类中的一个函数正在观察我的 MainRepo 类中的 Observable：

public void connectToWs()
        mainRepo.createListener()
                .observeOn(androidSchedulers.mainThread())
                .subscribe(new Observer<String>() 
            @Override
            public void onSubscribe(@NonNull Disposable d) 
                Timber.d("Subscribed");
            

            @Override
            public void onNext(@NonNull String s) 
                Timber.d("Message: " + s);
            

            @Override
            public void onError(@NonNull Throwable e) 
                Timber.e(e, "Something went wrong.");
            

            @Override
            public void onComplete() 
                Timber.d("On complete.");
            
        );
    

【参考方案1】：

创建一个PublishSubject 并更改您的createListener 方法以返回它：

private PublishSubject<String> publishSubject = PublishSubject.create<String>();

public Observable<String> createListener()
    return publishSubject;

PublishSubject 是一个 Observable，因此请注意您不需要更改方法签名，但我建议您将方法名称重命名为 observeMessages 之类的名称。

然后在您的 websocket 监听器中，您可以使用 onNext 方法将消息发送到 PublishSubject。您还应该在 onClosed 方法中调用 onComplete 并在 onFailure 方法中调用 onError：

 private WebSocketListener webSocketListener = new WebSocketListener() 

        @Override
        public void onOpen(@NotNull WebSocket webSocket, Response response) 
            Timber.d("Ws connection opened...", response.toString());
        

        @Override
        public void onClosing(@NotNull WebSocket webSocket, int code, @NotNull String reason) 
            Timber.d("Ws connection closing...");
        

        @Override
        public void onClosed(@NotNull WebSocket webSocket, int code, @NotNull String reason) 
            Timber.d("Ws connection closed...");

            publishSubject.onComplete();
        

        @Override
        public void onMessage(@NotNull WebSocket webSocket, @NotNull String text) 
            Timber.d("Ws incoming message.");

            publishSubject.onNext(text);
        

        @Override
        public void onFailure(@NotNull WebSocket webSocket, @NotNull Throwable t, Response response) 
            Timber.e(t, "Ws connection failure.", response.toString());

            publishSubject.onError(t);
        
    ;

【参考方案2】：

PublishSubject 是 Gustavo 发布的解决方案。但是createListener() 看起来很奇怪，我只是想展示一下我是如何做到的。

我没有使用 OkHttp，但我使用 nv-websocket-client 做完全相同的事情。它只是另一个 websocket 客户端。

在我的场景中，我之前做了很多反应性的事情，但套接字流程是：

连接到套接字 注册 OnTextMessage 监听器 - 它应该向我提供消息 发送消息以从套接字获取永久答案

OnCreate / OnCreateView 某处：

// Creating socket with credentials
WebSocketFactory factory = new WebSocketFactory();

try 
    socket = factory.createSocket("wss://ws.example.com", 3000);
 catch (IOException e) 
    e.printStackTrace();


// Create a subject
PublishSubject<String> subject = PublishSubject.create();

我之前做过一些HTTP/GET 的工作，并将结果保存在List<String> symbols - 它也是一种反应方式。在此之后，我调用 subscribeToSymbols 方法来完成整个套接字的工作：

public static Single<WebSocket> subscribeToSymbols(WebSocket socket,
                                                   PublishSubject<String> subject,
                                                   List<StockHttpGetData> symbols) 
    // connect -> register onTextMessage callback -> sendMessages to Socket
    return Single.fromCallable(socket::connect)
            .subscribeOn(Schedulers.io())
            .map(s -> s.addListener(new WebSocketAdapter() 
                @Override
                public void onTextMessage(WebSocket websocket, String text) 
                    subject.onNext(text);
                
            ))
            .doAfterSuccess(s -> symbols.forEach(httpData -> sendMessageToSubscribe(s, httpData)));

您对subscribeToSymbols 返回的内容不感兴趣。如果你订阅了这个主题，subject.onNext(text) 提供从套接字接收到的消息给你的关键点。

最后，订阅你之前创建的主题，然后用消息做你想做的事：

subject.flatMap(t -> Observable.just(new GsonBuilder()
            .registerTypeAdapter(TreeSet.class, new SocketMessageDeserializer())
            .create()
            .fromJson(t, TreeSet.class)))
       .filter(treeSet -> !treeSet.isEmpty())
       .observeOn(AndroidSchedulers.mainThread())
       .doOnNext(StocksRepository.getInstance()::printSocketMessage)
       .subscribe(adapter::setStocksChanged, Throwable::printStackTrace);

这是一个复杂的问题，因为您还应该处理连接错误、数据压力、配置更改并具有生命周期意识，但它也与非反应式方式（例如回调地狱）相关，因此 PublishSubject 是一个起点.

【参考方案3】：

尝试使用PublishSubject，好像它是为像你这样的情况而创建的。

【讨论】：

我的问题是我不知道如何在 websocket 侦听器中获取消息以使用subscribe(@NonNull ObservableEmitter<String> emitter 发回消息。如果您可以发布示例，那就太好了。