<commit_before>package com.lts.jobclient;

import com.lts.core.Application;
import com.lts.core.cluster.AbstractClientNode;
import com.lts.core.commons.utils.BatchUtils;
import com.lts.core.commons.utils.CollectionUtils;
import com.lts.core.commons.utils.CommonUtils;
import com.lts.core.constant.Constants;
import com.lts.core.domain.Job;
import com.lts.core.exception.JobSubmitException;
import com.lts.core.exception.JobTrackerNotFoundException;
import com.lts.core.logger.Logger;
import com.lts.core.logger.LoggerFactory;
import com.lts.core.protocol.JobProtos;
import com.lts.core.protocol.command.CommandBodyWrapper;
import com.lts.core.protocol.command.JobSubmitRequest;
import com.lts.core.protocol.command.JobSubmitResponse;
import com.lts.core.support.LoggerName;
import com.lts.jobclient.domain.JobClientApplication;
import com.lts.jobclient.domain.JobClientNode;
import com.lts.jobclient.domain.Response;
import com.lts.jobclient.domain.ResponseCode;
import com.lts.jobclient.processor.RemotingDispatcher;
import com.lts.jobclient.support.JobFinishedHandler;
import com.lts.jobclient.support.JobSubmitExecutor;
import com.lts.jobclient.support.JobSubmitProtector;
import com.lts.jobclient.support.SubmitCallback;
import com.lts.remoting.InvokeCallback;
import com.lts.remoting.netty.NettyRequestProcessor;
import com.lts.remoting.netty.ResponseFuture;
import com.lts.remoting.protocol.RemotingCommand;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author <NAME> (<EMAIL>) on 7/25/14.
 *         任务客户端
 */
public class JobClient<T extends JobClientNode, App extends Application> extends
        AbstractClientNode<JobClientNode, JobClientApplication> {

    protected static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.JobClient);

    private static final int BATCH_SIZE = 50;

    // 过载保护的提交者
    private JobSubmitProtector protector;

    private JobFinishedHandler jobFinishedHandler;

    public JobClient() {
        int concurrentSize = config.getParameter(Constants.JOB_SUBMIT_CONCURRENCY_SIZE,
                Constants.DEFAULT_JOB_SUBMIT_CONCURRENCY_SIZE);
        protector = new JobSubmitProtector(concurrentSize);
    }

    /**
     * @param job
     * @return
     */
    public Response submitJob(Job job) throws JobSubmitException {
        return protectSubmit(Arrays.asList(job));
    }

    private Response protectSubmit(List<Job> jobs) throws JobSubmitException {
        return protector.execute(jobs, new JobSubmitExecutor<Response>() {
            @Override
            public Response execute(List<Job> jobs) throws JobSubmitException {
                return submitJob(jobs, SubmitType.ASYNC);
            }
        });
    }

    private void checkFields(List<Job> jobs) {
        // 参数验证
        if (CollectionUtils.isEmpty(jobs)) {
            throw new JobSubmitException("Job can not be null!");
        }
        for (Job job : jobs) {
            if (job == null) {<commit_msg>Update JobClient.java<commit_after>package com.lts.jobclient;

import com.lts.core.Application;
import com.lts.core.cluster.AbstractClientNode;
import com.lts.core.commons.utils.BatchUtils;
import com.lts.core.commons.utils.CollectionUtils;
import com.lts.core.commons.utils.CommonUtils;
import com.lts.core.constant.Constants;
import com.lts.core.domain.Job;
import com.lts.core.exception.JobSubmitException;
import com.lts.core.exception.JobTrackerNotFoundException;
import com.lts.core.logger.Logger;
import com.lts.core.logger.LoggerFactory;
import com.lts.core.protocol.JobProtos;
import com.lts.core.protocol.command.CommandBodyWrapper;
import com.lts.core.protocol.command.JobSubmitRequest;
import com.lts.core.protocol.command.JobSubmitResponse;
import com.lts.core.support.LoggerName;
import com.lts.jobclient.domain.JobClientApplication;
import com.lts.jobclient.domain.JobClientNode;
import com.lts.jobclient.domain.Response;
import com.lts.jobclient.domain.ResponseCode;
import com.lts.jobclient.processor.RemotingDispatcher;
import com.lts.jobclient.support.JobFinishedHandler;
import com.lts.jobclient.support.JobSubmitExecutor;
import com.lts.jobclient.support.JobSubmitProtector;
import com.lts.jobclient.support.SubmitCallback;
import com.lts.remoting.InvokeCallback;
import com.lts.remoting.netty.NettyRequestProcessor;
import com.lts.remoting.netty.ResponseFuture;
import com.lts.remoting.protocol.RemotingCommand;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author <NAME> (<EMAIL>) on 7/25/14.
 *         任务客户端
 */
public class JobClient<T extends JobClientNode, App extends Application> extends
        AbstractClientNode<JobClientNode, JobClientApplication> {

    protected static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.JobClient);

    private static final int BATCH_SIZE = 50;

    // 过载保护的提交者
    private JobSubmitProtector protector;

    private JobFinishedHandler jobFinishedHandler;

    @Override
    protected void innerStart() {
        super.innerStart();
        int concurrentSize = config.getParameter(Constants.JOB_SUBMIT_CONCURRENCY_SIZE,
                Constants.DEFAULT_JOB_SUBMIT_CONCURRENCY_SIZE);
        protector = new JobSubmitProtector(concurrentSize);
    }

    /**
     * @param job
     * @return
     */
    public Response submitJob(Job job) throws JobSubmitException {
        return protectSubmit(Arrays.asList(job));
    }

    private Response protectSubmit(List<Job> jobs) throws JobSubmitException {
        return protector.execute(jobs, new JobSubmitExecutor<Response>() {
            @Override
            public Response execute(List<Job> jobs) throws JobSubmitException {
                return submitJob(jobs, SubmitType.ASYNC);
            }
        });
    }

    private void checkFields(List<Job> jobs) {
        // 参数验证
        if (CollectionUtils.isEmpty(jobs)) {
            throw new JobSubmitException("Job can not be null!");
        }
        for (Job job : jobs) {
            if (job == null) {
{'fraction_non_alphanumeric': 0.09841425535393167, 'fraction_numerical': 0.0022887036128821318, 'mean_word_length': 3.9458367016976554, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, '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': '7485733', 'n_tokens_mistral': 1873, 'n_tokens_neox': 1842, 'n_words': 334}