class AuthCompletable implements CompletableOnSubscribe {
    private CompletableEmitter completableEmitter;

    @Override
    public void subscribe(CompletableEmitter e) throws Exception {
      this.completableEmitter = e;
    }

    public void SignalAuthComplete() {
      completableEmitter.onComplete();
    }

    public void SignalError(String error) {
      completableEmitter.onError(new IllegalStateException(error));
    }
  }

  @Override
  public Completable connect() {
    synchronized (authCompletable) {
      Completable parentCompletable = super.connect();
      parentCompletable.blockingAwait();
      return Completable.create(authCompletable);
    }
  }

  @Override
  protected void handleMessage(JsonNode message) {
    LOG.debug("Receiving message: {}", message);
    JsonNode cexioMessage = message.get("e");

    try {
      if (cexioMessage != null) {
        switch (cexioMessage.textValue()) {
          case CONNECTED:
            auth();
            break;
          case AUTH:
            CexioWebSocketAuthResponse response =
                deserialize(message, CexioWebSocketAuthResponse.class);
            if (response != null) {
              if (response.isSuccess()) {
                synchronized (authCompletable) {
                  authCompletable.SignalAuthComplete();
                }
              } else {
                String authErrorString =
                    new String("Authentication error: " + response.getData().getError());
                LOG.error(authErrorString);
                synchronized (authCompletable) {
                  authCompletable.SignalError(authErrorString);
                }
              }
            }
            break;
          case PING:
            pong();
            break;
          case ORDER:
            try {
              CexioWebSocketOrderMessage cexioOrder =
                  deserialize(message, CexioWebSocketOrderMessage.class);
              Order order = CexioAdapters.adaptOrder(cexioOrder.getData());
              LOG.debug(String.format("Order is updated: %s", order));
              subjectOrder.onNext(order);
            } catch (Exception e) {
              LOG.error("Order parsing error: {}", e.getMessage(), e);
              subjectOrder.onError(e);
            }
            break;
          case TRANSACTION:
            try {
              CexioWebSocketTransactionMessage transaction =
                  deserialize(message, CexioWebSocketTransactionMessage.class);
              LOG.debug(String.format("New transaction: %s", transaction.getData()));
              subjectTransaction.onNext(transaction.getData());
            } catch (Exception e) {
              LOG.error("Transaction parsing error: {}", e.getMessage(), e);
              subjectTransaction.onError(e);
            }
            break;
          case ORDERBOOK:
            JsonNode okNode = message.get("ok");
            if (okNode.textValue().compareTo("ok") != 0) {
              String errorString =
                  "Error response for order book subscription: %s" + message.toString();
              LOG.error(errorString);
              subjectOrder.onError(new IllegalArgumentException(errorString));
            } else {
              super.handleMessage(message);
            }
            break;
          case ORDERBOOK_UPDATE:
            super.handleMessage(message);
            break;
        }
      }
    } catch (JsonProcessingException e) {
      LOG.error("Json parsing error: {}", e.getMessage());
    }
  }

  private void auth() {<FILL_FUNCTION_BODY>

    if (apiSecret == null || apiKey == null) {
      throw new IllegalStateException("API keys must be provided to use cexio streaming exchange");
    }
    long timestamp = System.currentTimeMillis() / 1000;
    CexioDigest cexioDigest = CexioDigest.createInstance(apiSecret);
    String signature = cexioDigest.createSignature(timestamp, apiKey);
    CexioWebSocketAuthMessage message =
        new CexioWebSocketAuthMessage(new CexioWebSocketAuth(apiKey, signature, timestamp));
    sendMessage(message);
  
<methods>public void <init>(java.lang.String) ,public void <init>(java.lang.String, int) ,public void <init>(java.lang.String, int, java.time.Duration, java.time.Duration, int) ,public void messageHandler(java.lang.String) ,public boolean processArrayMessageSeparately() <variables>private static final Logger LOG,protected final ObjectMapper objectMapper