A working example of reactive implementation of a socket connection using okhttp3 socket api

RxSocketConnection.java

import java.util.concurrent.TimeUnit;

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;
import rx.Observable;

public class RxSocketConnection extends WebSocketListener {

    private final BehaviorSubject<String> broadcaster = BehaviorSubject.create();

    public Observable<String> subscribe() {
        OkHttpClient client = new OkHttpClient.Builder()
                .readTimeout(0, TimeUnit.MILLISECONDS)
                .build();

        Request request = new Request.Builder()
                .url("ws://echo.websocket.org")
                .build();
        client.newWebSocket(request, this);

        client.dispatcher().executorService().shutdown();

        return broadcaster;
    }

    @Override public void onOpen(WebSocket webSocket, Response response) {
        super.onOpen(webSocket, response);
        broadcaster.onNext("Connection open");
    }

    @Override public void onClosed(WebSocket webSocket, int code, String reason) {
        super.onClosed(webSocket, code, reason);
        broadcaster.onNext("Connection closed");
    }

    @Override public void onMessage(WebSocket webSocket, String text) {
        super.onMessage(webSocket, text);
        broadcaster.onNext(text);
    }

    @Override public void onFailure(WebSocket webSocket, Throwable throwable, Response response) {
        super.onFailure(webSocket, throwable, response);
        broadcaster.onError(throwable);
    }
}