MessagingClientWithCache.java

        Observable<WebSocketConnection> source = HttpClient.newClient()
                                                           .createGet("/ws")
                                                           .requestWebSocketUpgrade()
                                                           .doOnNext(resp -> logger.info(resp.toString()))
                                                           .map(WebSocketResponse::getWebSocketConnection)
                                                           .nest()
                                                           .lift(new OperatorCacheSingleWebsocketConnection());

        Observable.interval(1, TimeUnit.SECONDS)
                  .map(aTick -> Observable.<WebSocketFrame>just(new MessageFrame(MessageType.Message,
                                                                                 msgIdGenerator.incrementAndGet())))
                  .flatMap(data -> source.flatMap(conn -> conn.write(data)))
                  .cast(MessageFrame.class)
                  .mergeWith(source.flatMap((webSocketConnection) -> webSocketConnection.getInput(false))
                                   .filter(AcceptOnlyBinaryFramesFilter.INSTANCE)
                                   .cast(BinaryWebSocketFrame.class)
                                   .map(frm -> new MessageFrame(frm.content())))
                  .retryWhen(errStream -> errStream.flatMap(err -> {
                      if (err instanceof IOException) {
                          return Observable.timer(1, TimeUnit.SECONDS);
                      }
                      return Observable.error(err);
                  }))
                  .take(10)
                  .toBlocking()
                  .forEach(msgId -> logger.info("Received acknowledgment for message id => " + msgId.toString()));