import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Test
public void checkpointAlgorithm() throws Exception {
	logger.info("starting checkpoint test");
	CountDownLatch latch = new CountDownLatch(1);
	JobOperator jobOperator = getJobOperator();
	Properties props = new Properties();
	long executionId = jobOperator.start(JOB_NAME, props);
	latch.await(10, SECONDS);

	JobInstance jobInstance = jobInstance(jobOperator.getJobInstance(executionId), JOB_NAME);
	JobExecution jobExecution = jobOperator.getJobExecutions(jobInstance).get(0);
	Properties jobProperties = jobExecution.getJobParameters();
	assertEquals("properties are: ", 0, jobProperties.size());
	assertEquals("batch failed because a NoRollbackException is throwed", FAILED, jobExecution.getBatchStatus());
	StepExecution stepExecution = jobOperator.getStepExecutions(executionId).get(0);
	PersistentCheckpointUserData persistentCheckpointUserData = (PersistentCheckpointUserData) stepExecution
			.getPersistentUserData();
	assertNull("persistent user data after step", persistentCheckpointUserData.getStartedAfterStep());
	Metric[] stepMetrics = stepExecution.getMetrics();
	stepMetrics(stepMetrics, 3, 10, 12, 2, 1);
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
/**
 * Start the job.
 *
 * @response the JobInstance + JobExecution record of the newly started job.
 */
protected void start(HttpServletRequest request, HttpServletResponse response) throws IOException {

    String jobXMLName = getRequiredParm(request, "jobXMLName");
    Properties jobParameters = getJobParameters(request, "jobParameters");

    JobOperator jobOperator = getJobOperator();
    long execId = jobOperator.start(jobXMLName, jobParameters);

    JobInstance jobInstance = jobOperator.getJobInstance(execId);
    JobExecution jobExecution = jobOperator.getJobExecution(execId);

    getResponseWriter().setHttpServletResponse( response )
                       .beginResponse(HttpServletResponse.SC_OK)
                       .println( "start(jobXMLName=" + jobXMLName + ", jobParameters=" + jobParameters + "): Job started!" )
                       .printJobInstance( jobInstance )
                       .printJobExecution( jobExecution )
                       .endResponse();
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
/**
 * Get status for the job identified by the 'executionId' query parm.
 * 
 * @response instance + execs
 */
protected void status(HttpServletRequest request, HttpServletResponse response) throws IOException {

    long execId = getLongParm(request, "executionId");

    JobOperator jobOperator = getJobOperator();
    JobInstance jobInstance = jobOperator.getJobInstance(execId);
    List<JobExecution> jobExecutions = jobOperator.getJobExecutions(jobInstance);

    getResponseWriter().setHttpServletResponse( response )
                       .beginResponse(HttpServletResponse.SC_OK)
                       .println( "status(executionId=" + execId + "): ")
                       .printJobInstance( jobInstance )
                       .printJobExecutions( jobExecutions )
                       .endResponse();
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
/**
 * @response the jobexecution records
 */
protected void getJobExecutions(HttpServletRequest request, HttpServletResponse response) throws IOException {

    final long instanceId = getLongParm(request, "instanceId") ;

    List<JobExecution> jobExecutions = getJobOperator().getJobExecutions( new JobInstance() {
        public long getInstanceId() { return instanceId; }
        public String getJobName() { return "dummy-for-call-to-getJobExecutions"; }
    });

    getResponseWriter().setHttpServletResponse( response )
                       .beginResponse(HttpServletResponse.SC_OK)
                       .println( "getJobExecutions(instanceId=" + instanceId + "): ")
                       .printJobExecutions(jobExecutions)
                       .endResponse();
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public List<JobExecution> getJobExecutions(JobInstance instance)
		throws NoSuchJobInstanceException, JobSecurityException {
	List<JobExecution> executions = new ArrayList<JobExecution>();

	if (isAuthorized(instance.getInstanceId())) {
		// Mediate between one 
		List<IJobExecution> executionImpls = persistenceService.jobOperatorGetJobExecutions(instance.getInstanceId());
		if (executionImpls.size() == 0 ){
			logger.warning("The current user is not authorized to perform this operation");
			throw new NoSuchJobInstanceException( "Job: " + instance.getJobName() + " does not exist");
		}
		for (IJobExecution e : executionImpls) {
			executions.add(e);
		}
	} else {
		logger.warning("The current user is not authorized to perform this operation");
		throw new JobSecurityException("The current user is not authorized to perform this operation");
	}

	return executions;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public Properties getParameters(long executionId)
		throws NoSuchJobExecutionException, JobSecurityException{

	Properties props = null;
	JobInstance requestedJobInstance = batchKernel.getJobInstance(executionId);

	if (isAuthorized(requestedJobInstance.getInstanceId())) {
		props = persistenceService.getParameters(executionId);
	} else {
		logger.warning("getParameters: The current user is not authorized to perform this operation");
		throw new JobSecurityException("The current user is not authorized to perform this operation");
	}

	return props;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public JobInstance createSubJobInstance(String name, String apptag) {
	Connection conn = null;
	PreparedStatement statement = null;
	ResultSet rs = null;
	JobInstanceImpl jobInstance = null;

	try {
		conn = getConnection();
		statement = conn.prepareStatement("INSERT INTO jobinstancedata (name, apptag) VALUES(?, ?)", new String[] { "JOBINSTANCEID" } );
		statement.setString(1, name);
		statement.setString(2, apptag); 
		statement.executeUpdate();
		rs = statement.getGeneratedKeys();
		if(rs.next()) {
			long jobInstanceID = rs.getLong(1);
			jobInstance = new JobInstanceImpl(jobInstanceID);
			jobInstance.setJobName(name);
		}
	} catch (SQLException e) {
		throw new PersistenceException(e);
	} finally {
		cleanupConnection(conn, rs, statement);
	}
	return jobInstance;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public JobInstance createJobInstance(String name, String apptag, String jobXml) {
	Connection conn = null;
	PreparedStatement statement = null;
	ResultSet rs = null;
	JobInstanceImpl jobInstance = null;

	try {
		conn = getConnection();
		statement = conn.prepareStatement("INSERT INTO jobinstancedata (name, apptag) VALUES(?, ?)", new String[] { "JOBINSTANCEID" } );
		statement.setString(1, name);
		statement.setString(2, apptag);
		statement.executeUpdate();
		rs = statement.getGeneratedKeys();
		if(rs.next()) {
			long jobInstanceID = rs.getLong(1);
			jobInstance = new JobInstanceImpl(jobInstanceID, jobXml);
			jobInstance.setJobName(name);
		}
	} catch (SQLException e) {
		throw new PersistenceException(e);
	} finally {
		cleanupConnection(conn, rs, statement);
	}
	return jobInstance;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Test
public void test19928() throws Exception {
	for (String jn : jobOp.getJobNames()) {
		System.out.println("JN: " + jn);
		List<JobInstance> exe = jobOp.getJobInstances(jn, 0, 200);
		if (exe != null) {
			for (JobInstance ji : exe) {
				System.out.println("JI: " + ji.getInstanceId());
				for (JobExecution je : jobOp.getJobExecutions(ji)) { 
					System.out.println("JE: " + je.getExecutionId());
					if (je.getJobName() != null) {
						System.out.println("SKSK: ");
					} else {
						throw new RuntimeException("saw null for je: " + je.getExecutionId() 
								+ ", ji: " + ji.getInstanceId() + ", jn= " + jn);
					}
				}
			}
		}
	}
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Test
public void testJobParametersFromJobExecutionsFromInstance() throws Exception {
	Properties origParams = new Properties();

	String sleepPropName = "sleepTime";
	String sleepPropVal = "100";
	origParams.setProperty(sleepPropName, sleepPropVal);

	long execId = jobOp.start("simpleJob", origParams);
	Thread.sleep(sleepTime);

	JobInstance ji = jobOp.getJobInstance(execId);
	JobExecution je = jobOp.getJobExecutions(ji).get(0);
	assertEquals("batch status", BatchStatus.COMPLETED, je.getBatchStatus());
	
	Properties obtainedParams = je.getJobParameters();
	assertEquals("Unexpected number of job parameters", 1, obtainedParams.size());
	assertEquals("Unexpected sleepTime parameter value", sleepPropVal, obtainedParams.getProperty(sleepPropName));

}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public JobInstance createJobInstance(final String name, final String jobXml) {
    final EntityManager em = emProvider.newEntityManager();
    try {
        final JobInstanceEntity instance = new JobInstanceEntity();
        instance.setName(name);
        instance.setJobXml(jobXml);

        final Object tx = txProvider.start(em);
        try {
            em.persist(instance);
            txProvider.commit(tx);

            final JobInstanceImpl jobInstance = new JobInstanceImpl(instance.getJobInstanceId(), jobXml);
            jobInstance.setJobName(name);
            return jobInstance;
        } catch (final Exception e) {
            throw new BatchContainerRuntimeException(performRollback(tx, e));
        }
    } finally {
        emProvider.release(em);
    }
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public JobInstance createJobInstance(final String name, final String jobXml) {
    final JobInstanceImpl jobInstance = new JobInstanceImpl(data.jobInstanceIdGenerator.getAndIncrement(), jobXml);
    jobInstance.setJobName(name);

    final Structures.JobInstanceData jobInstanceData = new Structures.JobInstanceData();
    jobInstanceData.instance = jobInstance;
    if (maxSize > 0 && data.jobInstanceData.size() >= maxSize) {
        synchronized (this) {
            while (data.jobInstanceData.size() >= maxSize) {
                cleanUp(data.lastCleanedJobInstanceId.getAndIncrement());
            }
        }
    }
    data.jobInstanceData.put(jobInstance.getInstanceId(), jobInstanceData);

    return jobInstance;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
private Structures.ExecutionInstanceData createRuntimeJobExecutionEntry(final JobInstance jobInstance, final Properties jobParameters,
                                                                        final BatchStatus batchStatus, final Timestamp now) {
    final Structures.ExecutionInstanceData executionInstanceData = new Structures.ExecutionInstanceData();
    final long id = data.executionInstanceIdGenerator.getAndIncrement();
    executionInstanceData.execution = new JobExecutionImpl(id, jobInstance.getInstanceId(), this);
    executionInstanceData.execution.setExecutionId(id);
    executionInstanceData.execution.setInstanceId(jobInstance.getInstanceId());
    executionInstanceData.execution.setBatchStatus(batchStatus.name());
    executionInstanceData.execution.setCreateTime(now);
    executionInstanceData.execution.setLastUpdateTime(now);
    executionInstanceData.execution.setJobParameters(jobParameters);

    data.executionInstanceData.put(id, executionInstanceData);
    final Structures.JobInstanceData jobInstanceData = data.jobInstanceData.get(jobInstance.getInstanceId());
    synchronized (jobInstanceData.executions) {
        jobInstanceData.executions.add(executionInstanceData);
    }
    return executionInstanceData;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
public static RuntimeJobExecution startJob(final ServicesManager servicesManager, final String jobXML, final Properties jobParameters) throws JobStartException {
    final JSLJob jobModel = new JobModelResolver().resolveModel(jobXML);
    final ModelNavigator<JSLJob> jobNavigator = getResolvedJobNavigator(jobModel, jobParameters, false);
    final JobContextImpl jobContext = getJobContext(jobNavigator);
    final JobInstance jobInstance = getNewJobInstance(servicesManager, jobNavigator.getRootModelElement().getId(), jobXML);
    final RuntimeJobExecution executionHelper =
            servicesManager.service(PersistenceManagerService.class).createJobExecution(jobInstance, jobParameters, jobContext.getBatchStatus());

    executionHelper.prepareForExecution(jobContext);

    final JobStatusManagerService statusManagerService = servicesManager.service(JobStatusManagerService.class);
    final JobStatus jobStatus = createNewJobStatus(statusManagerService, jobInstance);
    statusManagerService.updateJobStatus(jobStatus);

    return executionHelper;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
public static RuntimeJobExecution startPartition(final ServicesManager servicesManager, final JSLJob jobModel, final Properties jobParameters) throws JobStartException {
    final ModelNavigator<JSLJob> jobNavigator = getResolvedJobNavigator(jobModel, jobParameters, true);
    final JobContextImpl jobContext = getJobContext(jobNavigator);

    final JobInstance jobInstance = getNewSubJobInstance(servicesManager, jobNavigator.getRootModelElement().getId());

    final RuntimeJobExecution executionHelper =
            servicesManager.service(PersistenceManagerService.class).createJobExecution(jobInstance, jobParameters, jobContext.getBatchStatus());

    executionHelper.prepareForExecution(jobContext);

    final JobStatusManagerService statusManagerService = servicesManager.service(JobStatusManagerService.class);
    final JobStatus jobStatus = createNewJobStatus(statusManagerService, jobInstance);
    statusManagerService.updateJobStatus(jobStatus);

    return executionHelper;
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public void doRun() {
    final JobOperator operator = operator();
    final long total = operator.getJobInstanceCount(name);
    info(name + " has " + total + " job instances");

    info("");
    info("instance id");
    info("-----------");
    final List<JobInstance> id = operator.getJobInstances(name, start, count);
    if (id != null) {
        for (final JobInstance instance : id) {
            info(Long.toString(instance.getInstanceId()));
        }
        info("-----------");
        info("Current/Total: " + (start + id.size()) + "/" + operator.getJobInstanceCount(name));
    } else {
        info("No instance found.");
    }
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
public void doRun() {
    final JobOperator operator = operator();
    final Set<String> names = operator.getJobNames();
    if (names == null || names.isEmpty()) {
        info("No job");
    } else {
        info("     Name   \t|\texecution id\t|\tbatch status\t|\texit status\t|\tstart time\t|\tend time");
        for (final String name : names) {
            try {
                final JobExecution exec = new LinkedList<JobExecution>(
                        operator.getJobExecutions(
                            new LinkedList<JobInstance>(
                                operator.getJobInstances(name, operator.getJobInstanceCount(name) - 1, 2)).getLast())).getLast();
                info(String.format("%s\t|\t%12d\t|\t%s\t|\t%s\t|\t%tc\t|\t%tc",
                        StringUtils.leftPad(name, 12),
                        exec.getExecutionId(),
                        StringUtils.leftPad(exec.getBatchStatus() != null ? exec.getBatchStatus().toString() : "null", 12),
                        StringUtils.leftPad(exec.getExitStatus(), 11), exec.getStartTime(), exec.getEndTime()));
            } catch (final NoSuchJobException nsje) {
                // no-op
            }
        }
    }
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Override
protected Object doInvoke(final Method jaxrsMethod, final Method method, final Object[] args) throws Throwable {
    Object[] usedArgs = args;

    final Class<?>[] parameterTypes = method.getParameterTypes();
    if (parameterTypes.length == 1 && JobInstance.class.equals(parameterTypes[0])) {
        if (args[0] == null) {
            usedArgs = new Object[2];
        } else {
            final JobInstance ji = JobInstance.class.cast(args[0]);
            usedArgs = new Object[] { ji.getInstanceId(), ji.getJobName() };
        }
    }
    if (args != null && args.length == 2 && Properties.class.isInstance(args[1])) {
        usedArgs[1] = RestProperties.wrap(Properties.class.cast(args[1]));
    }

    try {
        return jaxrsMethod.invoke(client, usedArgs);
    } catch (final InvocationTargetException ite) {
        throw ite.getCause();
    } catch (final Exception e) {
        throw new IllegalStateException(e);
    }
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
protected static Object convertParam(final Object arg, final Type type) {
    if (arg == null) {
        return null;
    }

    // simple types are mapped 1-1 between JobOperator and JBatchResource
    if (String.class.equals(type) || int.class.equals(type) || long.class.equals(type)) {
        return arg;
    }

    if (Properties.class.equals(type)) {
        final Properties props = Properties.class.cast(arg);
        return RestProperties.wrap(props);
    }
    if (JobInstance.class.equals(type)) {
        final JobInstance instance = JobInstance.class.cast(arg);
        return new RestJobInstance(instance.getJobName(), instance.getInstanceId());
    }

    throw new IllegalArgumentException("Unexpected type " + type + " for a parameter");
}
 

import javax.batch.runtime.JobInstance; //导入依赖的package包/类
@Test
public void startProcess() throws Exception {
	logger.info("starting process test");
	CountDownLatch latch = new CountDownLatch(1);
	JobOperator jobOperator = getJobOperator();
	Properties props = new Properties();
	props.setProperty(INPUT_DATA_FILE_NAME, INPUT_PROPERTIES);
	long executionId = jobOperator.start(JOB_NAME, props);
	latch.await(10, SECONDS);
	List<Long> runningExecutions = jobOperator.getRunningExecutions(JOB_NAME);
	assertEquals("running executions. The process is end", 0, runningExecutions.size());
	Set<String> jobNames = jobOperator.getJobNames();
	assertTrue("one or three jobs", jobNames.size() >= 1 && jobNames.size() <= 3);
	String strJobNames = "";
	for (String jobName : jobNames)
		strJobNames += jobName;
	assertTrue("one only job called", strJobNames.contains(JOB_NAME));
	List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
	assertEquals("no step executions", 1, stepExecutions.size());
	stepExecutions(stepExecutions.get(0));
	Properties properties = jobOperator.getParameters(executionId);
	assertEquals("one property found", 1, properties.size());
	assertEquals("MY_NEW_PROPERTY found", INPUT_PROPERTIES, properties.get(INPUT_DATA_FILE_NAME));
	JobInstance jobInstance = jobInstance(jobOperator.getJobInstance(executionId), JOB_NAME);
	jobExecutions(jobOperator.getJobExecutions(jobInstance));
	assertNotNull("executionId not empty", executionId);
	assertTrue("Created file from writer 1", new File(PAYROLL_TEMP_FILE + "1.tmp").exists());
	assertTrue("Created file from writer 2", new File(PAYROLL_TEMP_FILE + "2.tmp").exists());
	assertTrue("Created file from writer 3", new File(PAYROLL_TEMP_FILE + "3.tmp").exists());
}