protected void onJson()

in java/connectors/bybit/src/main/java/com/epam/deltix/data/connectors/bybit/BybitFeed.java [136:185]
44 lines of code
10 McCabe index (conditional complexity)

    protected void onJson(final CharSequence data, final boolean last, final JsonWriter jsonWriter) {
        jsonParser.parse(data);

        if (!last) {
            return;
        }

        JsonValue jsonValue = jsonParser.eoj();
        JsonObject object = jsonValue.asObject();

        String topic = object.getString("topic");
        if (topic == null) {
            return;
        }

        if (topic.startsWith("orderbook")) {
            String type = object.getString("type");
            long timestamp = object.getLong("ts");
            JsonObject jsonData = object.getObject("data");
            String symbol = jsonData.getString("s");
            if (type.equalsIgnoreCase("snapshot")) {
                QuoteSequenceProcessor quotesListener = processor().onBookSnapshot(symbol, timestamp);
                processSnapshotSide(quotesListener, jsonData.getArray("b"), false);
                processSnapshotSide(quotesListener, jsonData.getArray("a"), true);
                quotesListener.onFinish();
            } else if (type.equalsIgnoreCase("delta")) {
                QuoteSequenceProcessor quotesListener = processor().onBookUpdate(symbol, timestamp);
                processChanges(quotesListener, jsonData.getArray("b"), false);
                processChanges(quotesListener, jsonData.getArray("a"), true);
                quotesListener.onFinish();
            }
        } else if (topic.startsWith("publicTrade")) {
            String type = object.getString("type");
            if (type.equalsIgnoreCase("snapshot")) {
                long timestamp = object.getLong("ts");
                JsonArray jsonDataArray = object.getArrayRequired("data");
                for (int i = 0; i < jsonDataArray.size(); ++i) {
                    JsonObject trade = jsonDataArray.getObjectRequired(i);
                    String symbol = trade.getString("s");
                    long price = trade.getDecimal64Required("p");
                    long size = trade.getDecimal64Required("v");
                    String sideStr = trade.getString("S");

                    AggressorSide side = sideStr.equalsIgnoreCase("buy") ? AggressorSide.BUY : AggressorSide.SELL;

                    processor().onTrade(symbol, timestamp, price, size, side);
                }
            }
        }
    }