<commit_before>package cn.cerc.mis.sync;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.cerc.core.ISession;
import cn.cerc.core.Record;
import cn.cerc.mis.core.Application;

public class SyncDatabase implements IPopProcesser {
    private static final Logger log = LoggerFactory.getLogger(SyncDatabase.class);
    private ISyncServer queue;

    public SyncDatabase(ISyncServer queue) {
        super();
        this.queue = queue;
    }

    public void push(ISession session, String tableCode, Record record, SyncOpera opera) {
        Record rs = new Record();
        rs.setField("__table", tableCode);
        rs.setField("__opera", opera.ordinal());
        rs.copyValues(record);
        queue.push(session, rs);
    }

    public int pop(ISession session, int maxRecords) {
       return queue.pop(session, this, maxRecords);
    }

    @Override
    public boolean popRecord(ISession session, Record record) {
        String tableCode = record.getString("__table");
        int opera = record.getInt("__opera");
        int error = record.getInt("__error");
        record.delete("__table");
        record.delete("__opera");
        record.delete("__error");

        IPushProcesser processer = Application.getBean(IPushProcesser.class, "sync_" + tableCode);
        if (processer == null) {
            processer = new PushTableDefault().setTableCode(tableCode);
        }
        processer.setSession(session);

        boolean result;
        switch (SyncOpera.values()[opera]) {
            case Append:
                result = processer.appendRecord(record);
                break;
            case Delete:
                result = processer.deleteRecord(record);
                break;
            case Update:
                result = processer.updateRecord(record);
                break;
            case Reset:
                result = processer.resetRecord(record);
                break;
            default:
                throw new RuntimeException("not support opera.");
        }

        if (!result) {
            record.setField("__table", tableCode);
            record.setField("__opera", opera);
            record.setField("__error", error + 1);
            if (error < 5) {
                queue.repush(session, record);
                log.warn("sync {}.{} fail, times {}", tableCode, opera, error);
            } else {
                processer.abortRecord(record, SyncOpera.values()[opera]);
            }
        }
        return result;
    }

    public ISyncServer getQueue() {
        return queue;
    }

}<commit_msg>Update SyncDatabase.java<commit_after>package cn.cerc.mis.sync;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.cerc.core.ISession;
import cn.cerc.core.Record;
import cn.cerc.mis.core.Application;

public class SyncDatabase implements IPopProcesser {
    private static final Logger log = LoggerFactory.getLogger(SyncDatabase.class);
    private ISyncServer queue;

    public SyncDatabase(ISyncServer queue) {
        super();
        this.queue = queue;
    }

    public void push(ISession session, String tableCode, Record record, SyncOpera opera) {
        Record rs = new Record();
        rs.setField("__table", tableCode);
        rs.setField("__opera", opera.ordinal());
        rs.copyValues(record);
        queue.push(session, rs);
    }

    public int pop(ISession session, int maxRecords) {
        return queue.pop(session, this, maxRecords);
    }

    @Override
    public boolean popRecord(ISession session, Record record) {
        String tableCode = record.getString("__table");
        int opera = record.getInt("__opera");
        int error = record.getInt("__error");
        record.delete("__table");
        record.delete("__opera");
        record.delete("__error");

        IPushProcesser processer = Application.getBean(IPushProcesser.class, "sync_" + tableCode);
        if (processer == null) {
            processer = new PushTableDefault().setTableCode(tableCode);
        }
        processer.setSession(session);

        boolean result;
        switch (SyncOpera.values()[opera]) {
        case Append:
            result = processer.appendRecord(record);
            break;
        case Delete:
            result = processer.deleteRecord(record);
            break;
        case Update:
            result = processer.updateRecord(record);
            break;
        case Reset:
            result = processer.resetRecord(record);
            break;
        default:
            throw new RuntimeException("not support opera.");
        }

        if (!result) {
            record.setField("__table", tableCode);
            record.setField("__opera", opera);
            record.setField("__error", error + 1);
            if (error < 5) {
                queue.repush(session, record);
                log.warn("sync {}.{} fail, times {}, record {}", tableCode, opera, error, record);
            } else {
                processer.abortRecord(record, SyncOpera.values()[opera]);
            }
        }
        return result;
    }

    public ISyncServer getQueue() {
        return queue;
    }

}
{'fraction_non_alphanumeric': 0.11054983485525549, 'fraction_numerical': 0.0015543034777540316, 'mean_word_length': 2.148623853211009, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '4821675', 'n_tokens_mistral': 1429, 'n_tokens_neox': 1394, 'n_words': 311}