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

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

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

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

解决方法

创建一个 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,@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);
        }
    };

,

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 是一个起点.

,

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