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

Question

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

MainRepo class:

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 class 中的一个函数正在观察我的 MainRepo 中的 Observable class:

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.");
            }
        });
    }

Answer 1

尝试使用 PublishSubject，它似乎是为像您这样的情况创建的。

Answer 2

创建一个 PublishSubject 并将您的 createListener 方法更改为 return 它：

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

public Observable<String> createListener(){
    return publishSubject;
}

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

然后在您的 websocket 侦听器中，您可以在 onFailure 方法中使用 onNext method. You should also call onComplete in the onClosed method and onError 将消息发送到 PublishSubject：

 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);
        }
    };

Answer 3

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)));
}

您对subscribeToSymbolsreturns的内容不感兴趣。如果你订阅了这个主题，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 是一个起点.

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

RxJava + Websocket - How to add Observable to Websocket listener?

android

rx-java

rx-java2

okhttp

android-mvvm