Websockets Next and streams with different message types

[starksm64]

I happened across this websocket streams service:
https://github.com/binance/binance-spot-api-docs/blob/master/web-socket-streams.md

and wanted to try hooking up the Quarkus websockets as their java library is a bit clunky. I'm able to do this, but I cannot deal with the initial response message to the subscribe attempt, or the response message if one unsubscribes as cleanly as I would like. I was hoping I could deal with this by having a generic text message callback along with the a callback with the expected unmarshalled type along the lines of:

@WebSocketClient(path = "/streams")
public class StreamsEndpoint {

    @OnPingMessage
    void ping(Buffer data) {
        Log.info("ping: " + data);
    }

    @OnPongMessage
    void pong(Buffer data) {
        Log.info("pong: " + data);
    }

    @OnTextMessage
    void onMessage(JsonObject data) {
        Log.info("onMessage(JsonObject): " + data);
    }

    @OnTextMessage
    void onMessage(MiniTicker[] miniTickers) {
        Log.info("onMessage(MiniTicker): " + miniTickers.length);
    }
}

This is not allowed. So instead I need to add an error callback and check the exception type:

    @OnError
    public void onError(Throwable t, WebSocketClientConnection connection) {
        if(t instanceof TextDecodeException tde) {
            Log.infof("onError, control message=%s", tde.getText());
        } else {
            Log.errorf(t, "onError, id=%s, closed=%s", connection.clientId(), connection.isClosed());
        }
    }

I would rather not have to be checking against an exception type that does not seem a reliable api type. I have seen these types of streaming services be very lax in what they put in even the expected object messages, so deserialization issues are a common occurrence.

What if instead of using the onError method or a general @OnTextMessage handler, the webservices framework supported a @OnSerializationError callback to distinguish from the expected text message type and a connection level error?

[cescoffier]

\CC @mkouba

[mkouba]

This is not allowed. So instead I need to add an error callback and check the exception type:

Yes, an endpoint may declare at most one @OnTextMessage callback.

What if instead of using the onError method or a general @OnTextMessage handler, the webservices framework supported a @OnSerializationError callback to distinguish from the expected text message type and a connection level error?

So if I understand it correctly you would like to keep the @OnTextMessage void onMessage(MiniTicker[] miniTickers) callback and declare @OnSerializationError callback that would be invoked when a text message cannot be deserialized as MiniTicker[]?

If you find the TextDecodeException an "unreliable api type" then a custom codec with a custom exception might be what you're looking for, i.e. something like:

@WebSocketClient(path = "/streams")
public class StreamsEndpoint {

    @OnTextMessage
    void onMessage(MiniTicker[] miniTickers) {
        Log.info("onMessage(MiniTicker): " + miniTickers.length);
    }

    @OnError
    void onDeserializationError(MiniTickerDeserializationError e) {
       Log.infof("onError, control message=%s", e.getText());
    }
}

public class MiniTickerDeserializationError extends Exception {
   // ...
}

@Singleton
@Priority(1) // use a higher priority then the built-int JsonTextMessageCodec 
public class MiniTickerCodec implements TextMessageCodec<MiniTicker[]> {

        @Inject
        ObjectMapper mapper;

        @Override
        public boolean supports(Type type) {
           if (type instanceof Class clazz) {
               return clazz.componentType().equals(MiniTicker.class);
           } 
           return false;
        }

        @Override
        public String encode(MiniTicker[] value) {
           try {
              return mapper.writeValueAsString(value);
           } catch (JsonProcessingException e) {
              throw new IllegalStateException(e);
           }
        }

        @Override
        public MiniTicker[] decode(Type type, String value) {
            try {
               return mapper.readValue(value, mapper.getTypeFactory().constructType(type));
            } catch (Exception e) {
               throw new MiniTickerDeserializationError(value, e);
            }
        }
}

Keep in mind that you can declare multiple @OnError callbacks provided they define different error params.