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

import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.lgo.domain.*;
import info.bitrich.xchangestream.lgo.dto.*;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import io.reactivex.Observable;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.dto.Order;
import org.knowm.xchange.dto.trade.*;
import org.slf4j.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.*;

class LgoUserBatchSubscription {

    private final Observable<LgoGroupedUserUpdate> downstream;
    private static final Logger LOGGER = LoggerFactory.getLogger(LgoUserBatchSubscription.class);
    private final LgoStreamingService streamingService;
    private final CurrencyPair currencyPair;


    static LgoUserBatchSubscription create(LgoStreamingService streamingService, CurrencyPair currencyPair) {
        return new LgoUserBatchSubscription(streamingService, currencyPair);
    }

    private LgoUserBatchSubscription(LgoStreamingService streamingService, CurrencyPair currencyPair) {
        this.streamingService = streamingService;
        this.currencyPair = currencyPair;
        downstream = createSubscription();
    }

    Observable<LgoGroupedUserUpdate> getPublisher() {
        return downstream;
    }

    private Observable<LgoGroupedUserUpdate> createSubscription() {
        final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
        return streamingService
                .subscribeChannel(LgoAdapter.channelName("user", currencyPair))
                .map(s -> mapper.readValue(s.toString(), LgoUserMessage.class))
                .scan(new LgoGroupedUserUpdate(), (acc, s) -> {
                    List<LgoBatchOrderEvent> events = new ArrayList<>();
                    if (s.getType().equals("update")) {
                        if (s.getBatchId() != acc.getBatchId() + 1) {
                            LOGGER.warn("Wrong batch id. Expected {} get {}", acc.getBatchId() + 1, s.getBatchId());
                            resubscribe();
                        }
                        LgoUserUpdate userUpdate = (LgoUserUpdate) s;
                        List<Order> updates = updateAllOrders(currencyPair, userUpdate.getOrderEvents(), acc.getAllOpenOrders());
                        events.addAll(LgoAdapter.adaptOrderEvent(userUpdate.getOrderEvents(), s.getBatchId(), updates));
                        return new LgoGroupedUserUpdate(acc.getAllOpenOrders(), updates, events, s.getBatchId(), s.getType());
                    } else {
                        Collection<LimitOrder> allOrders = handleUserSnapshot(currencyPair, (LgoUserSnapshot) s);
                        ConcurrentMap<String, Order> asMap = allOrders.stream().collect(toConcurrentMap(LimitOrder::getId, this::copyOrder));
                        return new LgoGroupedUserUpdate(asMap, new ArrayList<>(allOrders), events, s.getBatchId(), s.getType());
                    }

                })
                .skip(1) // skips the first element, for this is the empty accumulator
                .share();
    }

    private void resubscribe() {
        if (downstream == null) {
            return;
        }
        try {
            String channelName = LgoAdapter.channelName("user", currencyPair);
            streamingService.sendMessage(streamingService.getUnsubscribeMessage(channelName));
            streamingService.sendMessage(streamingService.getSubscribeMessage(channelName));
        } catch (IOException e) {
            LOGGER.warn("Error resubscribing", e);
        }
    }

    private List<Order> updateAllOrders(CurrencyPair currencyPair, List<LgoBatchOrderEvent> orderEvents, Map<String, Order> allOpenOrders) {
        return orderEvents.stream()
                .map(orderEvent -> orderEvent.applyOnOrders(currencyPair, allOpenOrders))
                .map(this::copyOrder)
                .collect(Collectors.toList());
    }

    private Collection<LimitOrder> handleUserSnapshot(CurrencyPair currencyPair, LgoUserSnapshot s) {
        return LgoAdapter.adaptOrdersSnapshot(s.getSnapshotData(), currencyPair);

    }

    private Order copyOrder(Order order) {
        Order copy = order instanceof LimitOrder ? LimitOrder.Builder.from(order).build() : MarketOrder.Builder.from(order).build();
        // because actual released version of xchange-core has buggy Builder.from methods
        copy.setFee(order.getFee());
        copy.setCumulativeAmount(order.getCumulativeAmount());
        // https://github.com/knowm/XChange/pull/3163
        return copy;
    }
}<commit_msg>removes batchId check on order user updates<commit_after>package info.bitrich.xchangestream.lgo;

import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.lgo.domain.*;
import info.bitrich.xchangestream.lgo.dto.*;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import io.reactivex.Observable;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.dto.Order;
import org.knowm.xchange.dto.trade.*;

import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.*;

class LgoUserBatchSubscription {

    private final Observable<LgoGroupedUserUpdate> downstream;
    private final LgoStreamingService streamingService;
    private final CurrencyPair currencyPair;


    static LgoUserBatchSubscription create(LgoStreamingService streamingService, CurrencyPair currencyPair) {
        return new LgoUserBatchSubscription(streamingService, currencyPair);
    }

    private LgoUserBatchSubscription(LgoStreamingService streamingService, CurrencyPair currencyPair) {
        this.streamingService = streamingService;
        this.currencyPair = currencyPair;
        downstream = createSubscription();
    }

    Observable<LgoGroupedUserUpdate> getPublisher() {
        return downstream;
    }

    private Observable<LgoGroupedUserUpdate> createSubscription() {
        final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
        return streamingService
                .subscribeChannel(LgoAdapter.channelName("user", currencyPair))
                .map(s -> mapper.readValue(s.toString(), LgoUserMessage.class))
                .scan(new LgoGroupedUserUpdate(), (acc, s) -> {
                    List<LgoBatchOrderEvent> events = new ArrayList<>();
                    if (s.getType().equals("update")) {
                        LgoUserUpdate userUpdate = (LgoUserUpdate) s;
                        List<Order> updates = updateAllOrders(currencyPair, userUpdate.getOrderEvents(), acc.getAllOpenOrders());
                        events.addAll(LgoAdapter.adaptOrderEvent(userUpdate.getOrderEvents(), s.getBatchId(), updates));
                        return new LgoGroupedUserUpdate(acc.getAllOpenOrders(), updates, events, s.getBatchId(), s.getType());
                    } else {
                        Collection<LimitOrder> allOrders = handleUserSnapshot(currencyPair, (LgoUserSnapshot) s);
                        ConcurrentMap<String, Order> asMap = allOrders.stream().collect(toConcurrentMap(LimitOrder::getId, this::copyOrder));
                        return new LgoGroupedUserUpdate(asMap, new ArrayList<>(allOrders), events, s.getBatchId(), s.getType());
                    }

                })
                .skip(1) // skips the first element, for this is the empty accumulator
                .share();
    }

    private List<Order> updateAllOrders(CurrencyPair currencyPair, List<LgoBatchOrderEvent> orderEvents, Map<String, Order> allOpenOrders) {
        return orderEvents.stream()
                .map(orderEvent -> orderEvent.applyOnOrders(currencyPair, allOpenOrders))
                .map(this::copyOrder)
                .collect(Collectors.toList());
    }

    private Collection<LimitOrder> handleUserSnapshot(CurrencyPair currencyPair, LgoUserSnapshot s) {
        return LgoAdapter.adaptOrdersSnapshot(s.getSnapshotData(), currencyPair);

    }

    private Order copyOrder(Order order) {
        Order copy = order instanceof LimitOrder ? LimitOrder.Builder.from(order).build() : MarketOrder.Builder.from(order).build();
        // because actual released version of xchange-core has buggy Builder.from methods
        copy.setFee(order.getFee());
        copy.setCumulativeAmount(order.getCumulativeAmount());
        // https://github.com/knowm/XChange/pull/3163
        return copy;
    }
}
{'fraction_non_alphanumeric': 0.09910013844023996, 'fraction_numerical': 0.0014997692662667283, 'mean_word_length': 2.9820854386770788, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 35, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '834493', 'n_tokens_mistral': 2348, 'n_tokens_neox': 2284, 'n_words': 431}