<commit_before>package com.tqmall.search.canal;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.tqmall.search.commons.lang.Function;
import com.tqmall.search.commons.utils.CommonsUtils;

import java.io.Serializable;
import java.util.*;

/**
 * Created by xing on 16/2/22.
 * {@link CanalEntry}中一行数据改动封装
 * 对于RowChangedData对象, 可以通过{@link #getEventType(RowChangedData)}判断对象类型
 *
 * @see #getEventType(RowChangedData)
 */
public abstract class RowChangedData<V> implements Function<String, V>, Serializable {

    private static final long serialVersionUID = -8712239138384357603L;

    final Map<String, V> fieldValueMap = new HashMap<>();

    RowChangedData() {
    }

    RowChangedData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
        initByRowData(rowData, interestedColumns);
    }

    @Override
    public V apply(String s) {
        return fieldValueMap.get(s);
    }

    @Override
    public String toString() {
        return fieldValueMap.toString();
    }

    /**
     * 通过{@link CanalEntry.RowData}初始化数据, 仅限于内部调用
     *
     * @param rowData           canal变化的列值
     * @param interestedColumns 感兴趣的列, 如果{@link CommonsUtils#isEmpty(Collection)}为true则全部包含
     */
    protected abstract void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns);

    public static final class Insert extends RowChangedData<String> {

        private static final long serialVersionUID = -2687037454927572799L;

        Insert() {
        }

        public Insert(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            super(rowData, interestedColumns);
        }

        @Override
        public String toString() {
            return CanalEntry.EventType.INSERT.toString() + ':' + super.toString();
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            if (CommonsUtils.isEmpty(interestedColumns)) {
                for (CanalEntry.Column c : rowData.getAfterColumnsList()) {
                    fieldValueMap.put(c.getName(), c.getValue());
                }
            } else {
                for (CanalEntry.Column c : rowData.getAfterColumnsList()) {
                    if (interestedColumns.contains(c.getName())) {
                        fieldValueMap.put(c.getName(), c.getValue());
                    }
                }
            }
        }
    }

    public static final class Delete extends RowChangedData<String> {

        private static final long serialVersionUID = 6254540878604970123L;

        Delete() {
        }

        public Delete(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            super(rowData, interestedColumns);
        }

        @Override
        public String toString() {
            return CanalEntry.EventType.DELETE.toString() + ':' + super.toString();
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            if (CommonsUtils.isEmpty(interestedColumns)) {
                for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                    fieldValueMap.put(c.getName(), c.getValue());
                }
            } else {
                for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                    if (interestedColumns.contains(c.getName())) {
                        fieldValueMap.put(c.getName(), c.getValue());
                    }
                }
            }
        }
    }

    public static final class Update extends RowChangedData<Pair> {

        private static final long serialVersionUID = 585376007150297603L;

        public Update(Map<String, Pair> dataMap) {
            super();
            fieldValueMap.putAll(dataMap);
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            //do nothing
        }

        public String getBefore(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) == null ? null : pair.before;
        }

        public String getAfter(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) == null ? null : pair.after;
        }

        public boolean isChanged(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) != null && pair.changed;
        }

        /**
         * 将{@link Update}实例转化成{@link Delete}, 当然,能不能转换自己在外面判断
         *
         * @return delete实例
         */
        public Delete transferToDelete() {
            Delete delete = new Delete();
            for (Map.Entry<String, Pair> e : fieldValueMap.entrySet()) {
                delete.fieldValueMap.put(e.getKey(), e.getValue().before);
            }
            return delete;<commit_msg>RowChangedData add some convert function<commit_after>package com.tqmall.search.canal;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.tqmall.search.commons.lang.Function;
import com.tqmall.search.commons.lang.StrValueConvert;
import com.tqmall.search.commons.utils.CommonsUtils;
import com.tqmall.search.commons.utils.StrValueConverts;

import java.io.Serializable;
import java.util.*;

/**
 * Created by xing on 16/2/22.
 * {@link CanalEntry}中一行数据改动封装
 * 对于RowChangedData对象, 可以通过{@link #getEventType(RowChangedData)}判断对象类型
 *
 * @see #getEventType(RowChangedData)
 */
public abstract class RowChangedData<V> implements Function<String, V>, Serializable {

    private static final long serialVersionUID = -8712239138384357603L;

    final Map<String, V> fieldValueMap = new HashMap<>();

    RowChangedData() {
    }

    RowChangedData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
        initByRowData(rowData, interestedColumns);
    }

    @Override
    public V apply(String s) {
        return fieldValueMap.get(s);
    }

    @Override
    public String toString() {
        return fieldValueMap.toString();
    }

    /**
     * 通过{@link CanalEntry.RowData}初始化数据, 仅限于内部调用
     *
     * @param rowData           canal变化的列值
     * @param interestedColumns 感兴趣的列, 如果{@link CommonsUtils#isEmpty(Collection)}为true则全部包含
     */
    protected abstract void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns);

    public static abstract class StrRowChangedData extends RowChangedData<String> {

        public StrRowChangedData() {
        }

        public StrRowChangedData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            super(rowData, interestedColumns);
        }

        private static final long serialVersionUID = -6170440278516894897L;

        public final <T> T getValue(String column, Class<T> cls) {
            StrValueConvert<T> convert = StrValueConverts.getConvert(cls);
            return convert != null ? getValue(column, convert) : null;
        }

        public final <T> T getValue(String column, StrValueConvert<T> convert) {
            String str = fieldValueMap.get(column);
            return str != null ? convert.convert(str) : null;
        }
    }

    public static final class Insert extends StrRowChangedData {

        private static final long serialVersionUID = -2687037454927572799L;

        Insert() {
        }

        public Insert(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            super(rowData, interestedColumns);
        }

        @Override
        public String toString() {
            return CanalEntry.EventType.INSERT.toString() + ':' + super.toString();
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            if (CommonsUtils.isEmpty(interestedColumns)) {
                for (CanalEntry.Column c : rowData.getAfterColumnsList()) {
                    fieldValueMap.put(c.getName(), c.getValue());
                }
            } else {
                for (CanalEntry.Column c : rowData.getAfterColumnsList()) {
                    if (interestedColumns.contains(c.getName())) {
                        fieldValueMap.put(c.getName(), c.getValue());
                    }
                }
            }
        }

    }

    public static final class Delete extends StrRowChangedData {

        private static final long serialVersionUID = 6254540878604970123L;

        Delete() {
        }

        public Delete(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            super(rowData, interestedColumns);
        }

        @Override
        public String toString() {
            return CanalEntry.EventType.DELETE.toString() + ':' + super.toString();
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            if (CommonsUtils.isEmpty(interestedColumns)) {
                for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                    fieldValueMap.put(c.getName(), c.getValue());
                }
            } else {
                for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                    if (interestedColumns.contains(c.getName())) {
                        fieldValueMap.put(c.getName(), c.getValue());
                    }
                }
            }
        }

    }

    public static final class Update extends RowChangedData<Pair> {

        private static final long serialVersionUID = 585376007150297603L;

        public Update(Map<String, Pair> dataMap) {
            super();
            fieldValueMap.putAll(dataMap);
        }

        @Override
        protected final void initByRowData(CanalEntry.RowData rowData, Set<String> interestedColumns) {
            //do nothing
        }

        public String getBefore(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) == null ? null : pair.before;
        }

        public <T> T getBefore(String column, Class<T> cls) {
            StrValueConvert<T> convert = StrValueConverts.getConvert(cls);
            return convert != null ? getAfter(column, convert) : null;
        }

        public <T> T getBefore(String column, StrValueConvert<T> convert) {
            String str = fieldValueMap.get(column) != null ? fieldValueMap.get(column).getBefore() : null;
            return str != null ? convert.convert(str) : null;
        }

        public String getAfter(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) == null ? null : pair.after;
        }

        public <T> T getAfter(String column, Class<T> cls) {
            StrValueConvert<T> convert = StrValueConverts.getConvert(cls);
            return convert != null ? getAfter(column, convert) : null;
        }

        public <T> T getAfter(String column, StrValueConvert<T> convert) {
            String str = fieldValueMap.get(column) != null ? fieldValueMap.get(column).getAfter() : null;
            return str != null ? convert.convert(str) : null;
        }

        public boolean isChanged(String column) {
            Pair pair;
            return (pair = fieldValueMap.get(column)) != null && pair.changed;
        }

        /**
         * 将{@link Update}实例转化成{@link Delete}, 当然,能不能转换自己在外面判断
         *
         * @return delete实例
         */
        public Delete transferToDelete() {
            Delete delete = new Delete();
            for (Map.Entry<String, Pair> e : fieldValueMap.entrySet()) {
                delete.fieldValueMap.put(e.getKey(), e.getValue().before);
            }
            return delete;
{'fraction_non_alphanumeric': 0.10085984522785899, 'fraction_numerical': 0.01539122957867584, 'mean_word_length': 2.2006053935057786, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 50, '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': '2654555', 'n_tokens_mistral': 3403, 'n_tokens_neox': 3208, 'n_words': 678}