import com.sforce.async.ContentType; //导入依赖的package包/类
private JobInfo createJob(String sobjectType, BulkConnection connection)
        throws AsyncApiException {
  JobInfo job = new JobInfo();
  job.setObject(sobjectType);
  job.setOperation(conf.queryAll ? OperationEnum.queryAll : OperationEnum.query);
  job.setContentType(ContentType.CSV);
  if (conf.usePKChunking) {
    String headerValue = CHUNK_SIZE + "=" + conf.chunkSize;
    if (!StringUtils.isEmpty(conf.startId)) {
      headerValue += "; " + START_ROW + "=" + conf.startId;
    }
    connection.addHeader(SFORCE_ENABLE_PKCHUNKING, headerValue);
  }
  job = connection.createJob(job);
  return job;
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
public String createBulkJob(String sobjectType, String upsertField,
    OperationEnum op) throws AsyncApiException
{
    Utils.log("[BULK] Creating Bulk Job:" + "\n\tObject:       [" + sobjectType + "]" + "\n\tUnique Field: ["
        + upsertField + "]" + "\n\tOperation:    [" + op + "]");

    // create a connection
    createBulkConnection();

    // create batch job
    JobInfo job = new JobInfo();
    job.setObject(sobjectType);
    job.setOperation(op);
    job.setConcurrencyMode(ConcurrencyMode.Serial);
    // JSON available in Spring '16
    job.setContentType(ContentType.JSON);
    if (upsertField != null)
    {
        job.setExternalIdFieldName(upsertField);
    }

    // create the job
    job = _bConn.createJob(job);

    Utils.log("Job created: " + job.getId());
    return job.getId();
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
public void addBatchToJob(String jobId, List<Map<String, Object>> records)
    throws UnsupportedEncodingException, AsyncApiException
{
    Utils.log("[BULK] Adding [" + records.size() + "] records to job [" + jobId + "].");

    // convert the records into a byte stream
    ByteArrayInputStream jsonStream = new ByteArrayInputStream(JsonUtil.toJson(records).getBytes("UTF-8"));

    JobInfo job = new JobInfo();
    job.setId(jobId);
    job.setContentType(ContentType.JSON);

    // submit a batch to the job
    _bConn.createBatchFromStream(job, jsonStream);
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
public void closeBulkJob(String jobId) throws AsyncApiException
{
    Utils.log("[BULK] Closing Bulk Job: [" + jobId + "]");

    JobInfo job = new JobInfo();
    job.setId(jobId);
    job.setState(JobStateEnum.Closed);
    job.setContentType(ContentType.JSON);
    // _bConn.updateJob(job, ContentType.JSON);

    // unclear if I can use this
    _bConn.closeJob(jobId);
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
public boolean createNewJob(String jobId) throws AsyncApiException
{
    JobInfo job = _bConn.getJobStatus(jobId, ContentType.JSON);
    Utils.log("[BULK] Getting Bulk Job Status: [" + jobId + "]");
    Calendar cal = job.getCreatedDate();

    // If current time is more than the time since the job was created, true
    return ((System.currentTimeMillis() - cal.getTime().getTime()) > SalesforceConstants.JOB_LIFE);
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
@Override
public JobInfo createBulkJob(String objectName) throws ResourceException {
       try {
		JobInfo job = new JobInfo();
		job.setObject(objectName);
		job.setOperation(OperationEnum.insert);
		job.setContentType(ContentType.XML);
		return this.bulkConnection.createJob(job);
	} catch (AsyncApiException e) {
		throw new ResourceException(e);
	}
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
public List<Map<String, String>> syncQuery(String objectType, String query)
        throws InterruptedException, AsyncApiException, IOException {

    // ジョブ作成
    JobInfo jobInfo = new JobInfo();
    jobInfo.setObject(objectType);
    if (queryAll) {
        jobInfo.setOperation(OperationEnum.queryAll);
    } else {
        jobInfo.setOperation(OperationEnum.query);
    }
    jobInfo.setContentType(ContentType.CSV);
    jobInfo = bulkConnection.createJob(jobInfo);

    // バッチ作成
    InputStream is = new ByteArrayInputStream(query.getBytes());
    BatchInfo batchInfo = bulkConnection.createBatchFromStream(jobInfo, is);

    // ジョブクローズ
    JobInfo closeJob = new JobInfo();
    closeJob.setId(jobInfo.getId());
    closeJob.setState(JobStateEnum.Closed);
    bulkConnection.updateJob(closeJob);

    // 実行状況取得
    batchInfo = waitBatch(batchInfo);
    BatchStateEnum state = batchInfo.getState();

    // 実行結果取得
    if (state == BatchStateEnum.Completed) {
        QueryResultList queryResultList =
                bulkConnection.getQueryResultList(
                        batchInfo.getJobId(),
                        batchInfo.getId());
        return getQueryResultMapList(batchInfo, queryResultList);
    } else {
        throw new AsyncApiException(batchInfo.getStateMessage(), AsyncExceptionCode.InvalidBatch);
    }
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
private JobInfo createJob(String sobjectType, OperationEnum operation, String externalIdField)
    throws AsyncApiException {
  JobInfo job = new JobInfo();
  job.setObject(sobjectType);
  job.setOperation(operation);
  if (externalIdField != null) {
    job.setExternalIdFieldName(externalIdField);
  }
  job.setContentType(ContentType.CSV);
  job = bulkConnection.createJob(job);
  LOG.info("Created Bulk API job {}", job.getId());
  return job;
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
private void setBulkOperation(String sObjectType, OutputAction userOperation, String externalIdFieldName,
        String contentTypeStr, String bulkFileName, int maxBytes, int maxRows) {
    this.sObjectType = sObjectType;
    switch (userOperation) {
    case INSERT:
        operation = OperationEnum.insert;
        break;
    case UPDATE:
        operation = OperationEnum.update;
        break;
    case UPSERT:
        operation = OperationEnum.upsert;
        break;
    case DELETE:
        operation = OperationEnum.delete;
        break;

    default:
        operation = OperationEnum.insert;
        break;
    }
    this.externalIdFieldName = externalIdFieldName;

    if ("csv".equals(contentTypeStr)) {
        contentType = ContentType.CSV;
    } else if ("xml".equals(contentTypeStr)) {
        contentType = ContentType.XML;
    }
    this.bulkFileName = bulkFileName;

    int sforceMaxBytes = 10 * 1024 * 1024;
    int sforceMaxRows = 10000;
    maxBytesPerBatch = (maxBytes > sforceMaxBytes) ? sforceMaxBytes : maxBytes;
    maxRowsPerBatch = (maxRows > sforceMaxRows) ? sforceMaxRows : maxRows;
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
/**
 * Creates and executes job for bulk query. Job must be finished in 2 minutes on Salesforce side.<br/>
 * From Salesforce documentation two scenarios are possible here:
 * <ul>
 * <li>simple bulk query. It should have status - {@link BatchStateEnum#Completed}.</li>
 * <li>primary key chunking bulk query. It should return first batch info with status - {@link BatchStateEnum#NotProcessed}.<br/>
 * Other batch info's should have status - {@link BatchStateEnum#Completed}</li>
 * </ul>
 *
 * @param moduleName - input module name.
 * @param queryStatement - to be executed.
 * @throws AsyncApiException
 * @throws InterruptedException
 * @throws ConnectionException
 */
public void doBulkQuery(String moduleName, String queryStatement)
        throws AsyncApiException, InterruptedException, ConnectionException {
    job = new JobInfo();
    job.setObject(moduleName);
    job.setOperation(OperationEnum.query);
    if (concurrencyMode != null) {
        job.setConcurrencyMode(concurrencyMode);
    }
    job.setContentType(ContentType.CSV);
    job = createJob(job);
    if (job.getId() == null) { // job creation failed
        throw new ComponentException(new DefaultErrorCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "failedBatch"),
                ExceptionContext.build().put("failedBatch", job));
    }

    ByteArrayInputStream bout = new ByteArrayInputStream(queryStatement.getBytes());
    BatchInfo info = createBatchFromStream(job, bout);
    int secToWait = 1;
    int tryCount = 0;
    while (true) {
        LOGGER.debug("Awaiting " + secToWait + " seconds for results ...\n" + info);
        Thread.sleep(secToWait * 1000);
        info = getBatchInfo(job.getId(), info.getId());

        if (info.getState() == BatchStateEnum.Completed
                || (BatchStateEnum.NotProcessed == info.getState() && 0 < chunkSize)) {
            break;
        } else if (info.getState() == BatchStateEnum.Failed) {
            throw new ComponentException(new DefaultErrorCode(HttpServletResponse.SC_BAD_REQUEST, "failedBatch"),
                    ExceptionContext.build().put("failedBatch", info));
        }
        tryCount++;
        if (tryCount % 3 == 0) {// after 3 attempt to get the result we multiply the time to wait by 2
            secToWait = secToWait * 2;
        }
        // There is also a 2-minute limit on the time to process the query.
        // If the query takes more than 2 minutes to process, a QUERY_TIMEOUT error is returned.
        // https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/asynch_api_concepts_limits.htm
        int processingTime = (int) ((System.currentTimeMillis() - job.getCreatedDate().getTimeInMillis()) / 1000);
        if (processingTime > 120) {
            throw new ComponentException(new DefaultErrorCode(HttpServletResponse.SC_REQUEST_TIMEOUT, "failedBatch"),
                    ExceptionContext.build().put("failedBatch", info));
        }
    }

    retrieveResultsOfQuery(info);
}
 

import com.sforce.async.ContentType; //导入依赖的package包/类
/**
 * Create a new job using the Bulk API.
 * 
 * @param sobjectType
 *            The object type being loaded, such as "Account"
 * @param connection
 *            BulkConnection used to create the new job.
 * @param operation
 *            operation to be performed - insert/update/query/upsert
 * @return The JobInfo for the new job.
 * @throws AsyncApiException
 */
public JobInfo createJob(String sobjectType, BulkConnection bulkConnection, OperationEnum operation)
		throws AsyncApiException {
	JobInfo job = new JobInfo();
	job.setObject(sobjectType);
	job.setOperation(operation);
	job.setContentType(ContentType.CSV);
	job = bulkConnection.createJob(job);

	return job;
}