<commit_before>package info.bitrich.xchangestream.kraken;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionConfig;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionMessage;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionStatusMessage;
import info.bitrich.xchangestream.kraken.dto.KrakenSystemStatus;
import info.bitrich.xchangestream.kraken.dto.enums.KrakenEventType;
import info.bitrich.xchangestream.kraken.dto.enums.KrakenSubscriptionName;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import org.apache.commons.lang3.StringUtils;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.kraken.dto.account.KrakenWebsocketToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static info.bitrich.xchangestream.kraken.dto.enums.KrakenEventType.subscribe;

/**
 * @author makarid, pchertalev
 */
public class KrakenStreamingService extends JsonNettyStreamingService {
    private static final Logger LOG = LoggerFactory.getLogger(KrakenStreamingService.class);
    private static final String EVENT = "event";
    private final Map<Integer, String> channels = new ConcurrentHashMap<>();
    private ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
    private final boolean isPrivate;

    private final Map<Integer, String> subscriptionRequestMap = new ConcurrentHashMap<>();

    public KrakenStreamingService(boolean isPrivate, String uri) {
        super(uri, Integer.MAX_VALUE);
        this.isPrivate = isPrivate;
    }


    @Override
    public boolean processArrayMassageSeparately() {
        return false;
    }

    @Override
    protected void handleMessage(JsonNode message) {
        String channelName = getChannel(message);

        try {
            JsonNode event = message.get(EVENT);
            KrakenEventType krakenEvent;
            if (event != null && (krakenEvent = KrakenEventType.getEvent(event.textValue())) != null) {
                switch (krakenEvent) {
                    case heartbeat:
                        LOG.debug("Heartbeat received");
                        break;
                    case systemStatus:
                        KrakenSystemStatus systemStatus = mapper.treeToValue(message, KrakenSystemStatus.class);
                        LOG.info("System status: {}", systemStatus);
                        break;
                    case subscriptionStatus:
                        KrakenSubscriptionStatusMessage statusMessage = mapper.treeToValue(message, KrakenSubscriptionStatusMessage.class);
                        Integer reqid = statusMessage.getReqid();
                        if (!isPrivate && reqid != null)
                            channelName = subscriptionRequestMap.remove(reqid);

                        switch (statusMessage.getStatus()) {
                            case subscribed:
                                LOG.info("Channel {} has been subscribed", channelName);

                                if (statusMessage.getChannelID() != null)
                                    channels.put(statusMessage.getChannelID(), channelName);

                                break;
                            case unsubscribed:
                                LOG.info("Channel {} has been unsubscribed", channelName);
                                channels.remove(statusMessage.getChannelID());
                                break;
                            case error:
                                LOG.error("Channel {} has been failed: {}", channelName, statusMessage.getErrorMessage());
                        }
                        break;
                    default:
                        LOG.warn("Unexpected event type has been received: {}", krakenEvent);
                }
                return;

            }
        } catch (JsonProcessingException e) {
            LOG.error("Error reading message: {}", e.getMessage(), e);
        }

        if (!message.isArray() || channelName == null) {
            LOG.error("Unknown message: {}", message.asText());
            return;
        }

        super.handleMessage(message);
    }

    @Override
    protected String getChannelNameFromMessage(JsonNode message) throws IOException {
        String channelName = null;
        if (message.has("channelID")) {
            channelName = channels.get(message.get("channelID").asInt());
        }
        if (message.has("channelName")) {
            channelName = message.get("channelName").asText();
        }

        if (message.isArray()) {
            if (message.get(0).isInt()) {
                channelName = channels.get(message.get(0).asInt());
            }
            if (message.get(1).isTextual()) {
                channelName = message.get(1).asText();
            }
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("ChannelName {}", StringUtils.isBlank(channelName) ? "not defined" : channelName);
        }
        return channelName;
    }

    @Override
    public String getSubscribeMessage(String channelName, Object... args) throws IOException {
        int reqID = Math.abs(UUID.randomUUID().hashCode());
        String [] channelData = channelName.split(KrakenStreamingMarketDataService.KRAKEN_CHANNEL_DELIMITER);
        KrakenSubscriptionName subscriptionName = KrakenSubscriptionName.valueOf(channelData[0]);

        if (isPrivate) {
            String token = (String) args[0];

            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, subscribe,
                    null, new KrakenSubscriptionConfig(subscriptionName, null, token));

            return objectMapper.writeValueAsString(subscriptionMessage);
        }

        String pair = channelData[1];

        Integer depth = null;
        if (args.length > 0 && args[0] != null) {
            depth = (Integer) args[0];
        }
        subscriptionRequestMap.put(reqID, channelName);

        KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, subscribe,
                Collections.singletonList(pair), new KrakenSubscriptionConfig(subscriptionName, depth, null));
        return objectMapper.writeValueAsString(subscriptionMessage);
    }

    @Override
    public String getUnsubscribeMessage(String channelName) throws IOException {
        int reqID = Math.abs(UUID.randomUUID().hashCode());
        String [] channelData = channelName.split(KrakenStreamingMarketDataService.KRAKEN_CHANNEL_DELIMITER);
        KrakenSubscriptionName subscriptionName = KrakenSubscriptionName.valueOf(channelData[0]);

        if (isPrivate) {
            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, KrakenEventType.unsubscribe,
                    null, new KrakenSubscriptionConfig(subscriptionName));
            return objectMapper.writeValueAsString(subscriptionMessage);
        }

        String pair = channelData[1];

        subscriptionRequestMap.put(reqID, channelName);
        KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, KrakenEventType.unsubscribe,
                Collections.singletonList(pair), new KrakenSubscriptionConfig(subscriptionName));
        return objectMapper.writeValueAsString(subscriptionMessage);
    }
}
<commit_msg>notes in Kraken getUnsubscribeMessage<commit_after>package info.bitrich.xchangestream.kraken;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionConfig;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionMessage;
import info.bitrich.xchangestream.kraken.dto.KrakenSubscriptionStatusMessage;
import info.bitrich.xchangestream.kraken.dto.KrakenSystemStatus;
import info.bitrich.xchangestream.kraken.dto.enums.KrakenEventType;
import info.bitrich.xchangestream.kraken.dto.enums.KrakenSubscriptionName;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import org.apache.commons.lang3.StringUtils;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.kraken.dto.account.KrakenWebsocketToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static info.bitrich.xchangestream.kraken.dto.enums.KrakenEventType.subscribe;

/**
 * @author makarid, pchertalev
 */
public class KrakenStreamingService extends JsonNettyStreamingService {
    private static final Logger LOG = LoggerFactory.getLogger(KrakenStreamingService.class);
    private static final String EVENT = "event";
    private final Map<Integer, String> channels = new ConcurrentHashMap<>();
    private ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
    private final boolean isPrivate;

    private final Map<Integer, String> subscriptionRequestMap = new ConcurrentHashMap<>();

    public KrakenStreamingService(boolean isPrivate, String uri) {
        super(uri, Integer.MAX_VALUE);
        this.isPrivate = isPrivate;
    }


    @Override
    public boolean processArrayMassageSeparately() {
        return false;
    }

    @Override
    protected void handleMessage(JsonNode message) {
        String channelName = getChannel(message);

        try {
            JsonNode event = message.get(EVENT);
            KrakenEventType krakenEvent;
            if (event != null && (krakenEvent = KrakenEventType.getEvent(event.textValue())) != null) {
                switch (krakenEvent) {
                    case heartbeat:
                        LOG.debug("Heartbeat received");
                        break;
                    case systemStatus:
                        KrakenSystemStatus systemStatus = mapper.treeToValue(message, KrakenSystemStatus.class);
                        LOG.info("System status: {}", systemStatus);
                        break;
                    case subscriptionStatus:
                        KrakenSubscriptionStatusMessage statusMessage = mapper.treeToValue(message, KrakenSubscriptionStatusMessage.class);
                        Integer reqid = statusMessage.getReqid();
                        if (!isPrivate && reqid != null)
                            channelName = subscriptionRequestMap.remove(reqid);

                        switch (statusMessage.getStatus()) {
                            case subscribed:
                                LOG.info("Channel {} has been subscribed", channelName);

                                if (statusMessage.getChannelID() != null)
                                    channels.put(statusMessage.getChannelID(), channelName);

                                break;
                            case unsubscribed:
                                LOG.info("Channel {} has been unsubscribed", channelName);
                                channels.remove(statusMessage.getChannelID());
                                break;
                            case error:
                                LOG.error("Channel {} has been failed: {}", channelName, statusMessage.getErrorMessage());
                        }
                        break;
                    default:
                        LOG.warn("Unexpected event type has been received: {}", krakenEvent);
                }
                return;

            }
        } catch (JsonProcessingException e) {
            LOG.error("Error reading message: {}", e.getMessage(), e);
        }

        if (!message.isArray() || channelName == null) {
            LOG.error("Unknown message: {}", message.asText());
            return;
        }

        super.handleMessage(message);
    }

    @Override
    protected String getChannelNameFromMessage(JsonNode message) throws IOException {
        String channelName = null;
        if (message.has("channelID")) {
            channelName = channels.get(message.get("channelID").asInt());
        }
        if (message.has("channelName")) {
            channelName = message.get("channelName").asText();
        }

        if (message.isArray()) {
            if (message.get(0).isInt()) {
                channelName = channels.get(message.get(0).asInt());
            }
            if (message.get(1).isTextual()) {
                channelName = message.get(1).asText();
            }
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("ChannelName {}", StringUtils.isBlank(channelName) ? "not defined" : channelName);
        }
        return channelName;
    }

    @Override
    public String getSubscribeMessage(String channelName, Object... args) throws IOException {
        int reqID = Math.abs(UUID.randomUUID().hashCode());
        String [] channelData = channelName.split(KrakenStreamingMarketDataService.KRAKEN_CHANNEL_DELIMITER);
        KrakenSubscriptionName subscriptionName = KrakenSubscriptionName.valueOf(channelData[0]);

        if (isPrivate) {
            String token = (String) args[0];

            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, subscribe,
                    null, new KrakenSubscriptionConfig(subscriptionName, null, token));

            return objectMapper.writeValueAsString(subscriptionMessage);
        } else {
            String pair = channelData[1];

            Integer depth = null;
            if (args.length > 0 && args[0] != null) {
                depth = (Integer) args[0];
            }
            subscriptionRequestMap.put(reqID, channelName);

            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, subscribe,
                    Collections.singletonList(pair), new KrakenSubscriptionConfig(subscriptionName, depth, null));
            return objectMapper.writeValueAsString(subscriptionMessage);
        }
    }

    @Override
    public String getUnsubscribeMessage(String channelName) throws IOException {
        int reqID = Math.abs(UUID.randomUUID().hashCode());
        String [] channelData = channelName.split(KrakenStreamingMarketDataService.KRAKEN_CHANNEL_DELIMITER);
        KrakenSubscriptionName subscriptionName = KrakenSubscriptionName.valueOf(channelData[0]);

        if (isPrivate) {
            String token = null; // TODO: We probably need to pass the token to unsubscribe (but we dont have args[0])
            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, KrakenEventType.unsubscribe,
                    null, new KrakenSubscriptionConfig(subscriptionName, null, null));
            return objectMapper.writeValueAsString(subscriptionMessage);
        } else {
            String pair = channelData[1];

            subscriptionRequestMap.put(reqID, channelName);
            KrakenSubscriptionMessage subscriptionMessage = new KrakenSubscriptionMessage(reqID, KrakenEventType.unsubscribe,
                    Collections.singletonList(pair), new KrakenSubscriptionConfig(subscriptionName));
            return objectMapper.writeValueAsString(subscriptionMessage);
        }
    }
}

{'fraction_non_alphanumeric': 0.09058226872953713, 'fraction_numerical': 0.0019901136290684986, 'mean_word_length': 2.4297666226332013, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2409706', 'n_tokens_mistral': 3926, 'n_tokens_neox': 3909, 'n_words': 735}