import com.sforce.async.AsyncApiException; //导入依赖的package包/类
public void createRemoteSite(RemoteSite rs)
  throws ConnectionException, DeployException, AsyncApiException, Exception {
    createMetadataConnection();

    RemoteSiteSetting rss = new RemoteSiteSetting();
    rss.setFullName(rs.fullName);
    rss.setUrl(rs.url);
    rss.setDescription(rs.description);
    rss.setIsActive(true);
    rss.setDisableProtocolSecurity(false);

    com.sforce.soap.metadata.SaveResult[] results = getMetadataConnection().createMetadata(new Metadata[] { rss });
    
    for (com.sforce.soap.metadata.SaveResult r : results) {
        if (r.isSuccess()) {
            System.out.println("Created component: " + r.getFullName());
        } else {
            throw new Exception(r.getErrors()[0].getMessage());
        }
    }
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
public DeployResponse upsertData(String IdField, List<SObject> sobjects)
    throws ConnectionException, DeployException, AsyncApiException
{
    DeployResponse dr = new DeployResponse();

    // ensure the partner connection has been initialized
    createPartnerConnection();

    // check if the SObject queue is empty
    if (sobjects == null || sobjects.isEmpty())
    {
        Utils.log("An empty list of SObject was passed to upsertData()!");
    }
    else
    {
        Utils.log(String.format("Upserting %d records with Id field [%s].", +sobjects.size(), IdField));

        // upsert the records
        com.sforce.soap.partner.UpsertResult[] URs = _pConn.upsert(IdField,
            sobjects.toArray(new SObject[] {}));

        // process the results and log errors
        dr = SalesforceService.handleUpsertResponse(dr, URs);
    }
    return dr;
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
@Override
public BatchResult[] getBulkResults(JobInfo job, List<String> ids) throws ResourceException {
	try {
		JobInfo info = this.bulkConnection.getJobStatus(job.getId());
		if (info.getNumberBatchesTotal() != info.getNumberBatchesFailed() + info.getNumberBatchesCompleted()) {
			//TODO: this should be configurable
			throw new DataNotAvailableException(500);
		}
		BatchResult[] results = new BatchResult[ids.size()];
		for (int i = 0; i < ids.size(); i++) {
			results[i] = this.bulkConnection.getBatchResult(job.getId(), ids.get(i));	
		}
		return results;
	} catch (AsyncApiException e) {
		throw new ResourceException(e);
	}
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
/**
 * Constructor
 */
public SalesforceBulkWrapper(
        String userName,
        String password,
        String authEndpointUrl,
        boolean isCompression,
        int pollingIntervalMillisecond,
        boolean queryAll)
        throws AsyncApiException, ConnectionException {

    partnerConnection = createPartnerConnection(
            authEndpointUrl,
            userName,
            password);
    bulkConnection = createBulkConnection(partnerConnection.getConfig());

    this.pollingIntervalMillisecond = pollingIntervalMillisecond;
    this.queryAll = queryAll;
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
private BulkConnection createBulkConnection(ConnectorConfig partnerConfig)
        throws AsyncApiException {

    ConnectorConfig config = new ConnectorConfig();
    config.setSessionId(partnerConfig.getSessionId());

    String soapEndpoint = partnerConfig.getServiceEndpoint();
    String restEndpoint = soapEndpoint.substring(
            0, soapEndpoint.indexOf("Soap/")) + "async/" + API_VERSION;
    config.setRestEndpoint(restEndpoint);
    config.setCompression(isCompression);

    config.setTraceMessage(false);

    return new BulkConnection(config);
}
 

import com.sforce.async.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的package包/类
private void awaitCompletion(JobBatches jb)
    throws AsyncApiException {
  long sleepTime = 0L;
  Set<String> incomplete = new HashSet<>();
  for (BatchInfo bi : jb.batchInfoList) {
    incomplete.add(bi.getId());
  }
  while (!incomplete.isEmpty()) {
    try {
      Thread.sleep(sleepTime);
    } catch (InterruptedException e) {}
    LOG.info("Awaiting Bulk API results... {}", incomplete.size());
    sleepTime = 1000L;
    BatchInfo[] statusList =
        bulkConnection.getBatchInfoList(jb.job.getId()).getBatchInfo();
    for (BatchInfo b : statusList) {
      if (b.getState() == BatchStateEnum.Completed
          || b.getState() == BatchStateEnum.Failed) {
        if (incomplete.remove(b.getId())) {
          LOG.info("Batch status: {}", b);
        }
      }
    }
  }
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
/**
 * Wait for a job to complete by polling the Bulk API.
 *
 * @throws AsyncApiException
 * @throws ConnectionException
 */
private void awaitCompletion() throws AsyncApiException, ConnectionException {
    long sleepTime = 0L;
    Set<String> incomplete = new HashSet<String>();
    for (BatchInfo bi : batchInfoList) {
        incomplete.add(bi.getId());
    }
    while (!incomplete.isEmpty()) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
        }
        sleepTime = awaitTime;
        BatchInfo[] statusList = getBatchInfoList(job.getId()).getBatchInfo();
        for (BatchInfo b : statusList) {
            if (b.getState() == BatchStateEnum.Completed || b.getState() == BatchStateEnum.Failed) {
                incomplete.remove(b.getId());
            }
        }
    }
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
protected JobInfo createJob(JobInfo job) throws AsyncApiException, ConnectionException {
    try {
        if (0 != chunkSize) {
            // Enabling PK chunking by setting header and chunk size.
            bulkConnection.addHeader(PK_CHUNKING_HEADER_NAME, CHUNK_SIZE_PROPERTY_NAME + chunkSize);
        }
        return bulkConnection.createJob(job);
    } catch (AsyncApiException sfException) {
        if (AsyncExceptionCode.InvalidSessionId.equals(sfException.getExceptionCode())) {
            SalesforceRuntimeCommon.renewSession(bulkConnection.getConfig());
            return createJob(job);
        }
        throw sfException;
    } finally {
        if (0 != chunkSize) {
            // Need to disable PK chunking after job was created.
            bulkConnection.addHeader(PK_CHUNKING_HEADER_NAME, Boolean.FALSE.toString());
        }
    }
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
public void bulkExecute(RuntimeContainer container) {
    TSalesforceBulkExecProperties sprops = (TSalesforceBulkExecProperties) properties;
    try {
        SalesforceBulkRuntime bulkRuntime = new SalesforceBulkRuntime(connect(container).bulkConnection);
        bulkRuntime.setConcurrencyMode(sprops.bulkProperties.concurrencyMode.getValue());
        bulkRuntime.setAwaitTime(sprops.bulkProperties.waitTimeCheckBatchState.getValue());
        // We only support CSV file for bulk output
        bulkRuntime.executeBulk(sprops.module.moduleName.getStringValue(), sprops.outputAction.getValue(),
                sprops.upsertKeyColumn.getStringValue(), "csv", sprops.bulkFilePath.getStringValue(),
                sprops.bulkProperties.bytesToCommit.getValue(), sprops.bulkProperties.rowsToCommit.getValue());
        // count results
        for (int i = 0; i < bulkRuntime.getBatchCount(); i++) {
            for (BulkResult result : bulkRuntime.getBatchLog(i)) {
                dataCount++;
                if ("true".equalsIgnoreCase(String.valueOf(result.getValue("Success")))) {
                    successCount++;
                } else {
                    rejectCount++;
                }
            }
        }
        bulkRuntime.close();
    } catch (IOException | AsyncApiException | ConnectionException e) {
        throw new ComponentException(e);
    }
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
private boolean retrieveNextResultSet() throws IOException {
    while (bulkRuntime.hasNextResultId()) {
        String resultId = bulkRuntime.nextResultId();
        if (null != resultId) {
            try {
                // Get a new result set
                bulkResultSet = bulkRuntime.getQueryResultSet(resultId);
            } catch (AsyncApiException | ConnectionException e) {
                throw new IOException(e);
            }

            currentRecord = bulkResultSet.next();
            // If currentRecord is null, we need to check if resultId set has more entries.
            if (null != currentRecord) {
                // New result set available to retrieve
                dataCount++;
                return true;
            }
        }
    }
    return false;
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
@Override
public boolean advance() throws IOException {
    if (++resultIndex >= currentBatchResult.size()) {
        if (++batchIndex >= bulkRuntime.getBatchCount()) {
            return false;
        } else {
            try {
                currentBatchResult = bulkRuntime.getBatchLog(batchIndex);
                resultIndex = 0;
                boolean isAdvanced = currentBatchResult.size() > 0;
                if (isAdvanced) {
                    countData();
                }
                return isAdvanced;
            } catch (AsyncApiException | ConnectionException e) {
                throw new IOException(e);
            }
        }
    }
    countData();
    return true;
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
private void createMetadataConnection()
    throws ConnectionException, AsyncApiException
{
    Utils.log("SalesforceService::createMetadataConnection() entered");
    // check if connection has already been created
    if (getMetadataConnection() != null)
    {
        // connection already created
        return;
    }

    ConnectorConfig config = getConnectorConfig(getServerUrl(), getSessionId());
    config.setServiceEndpoint(getMetadataUrl());

    // check if tracing is enabled
    if (getenv(SALESFORCE_TRACE_METADATA) != null && getenv(SALESFORCE_TRACE_METADATA).equalsIgnoreCase("1"))
    {
        // set this to true to see HTTP requests and responses on stdout
        config.setTraceMessage(true);
        config.setPrettyPrintXml(true);

        // this should only be false when doing debugging.
        config.setCompression(false);
    }

    setMetadataConnection(new MetadataConnection(config));

    // allow partial success
    getMetadataConnection().setAllOrNoneHeader(false);

    // print the endpoint
    Utils.log(
        "\n\tSession ID:            " + getSessionId() +
        "\n\tEndpoint:              " + getServerUrl() +
        "\n\tConnection Session ID: " + _mConn.getConfig().getSessionId() +
        "\n\tAuth Endpoint:         " + _mConn.getConfig().getAuthEndpoint());
}
 

import com.sforce.async.AsyncApiException; //导入依赖的package包/类
private void createBulkConnection() throws AsyncApiException
{
    // check if connection has already been created
    if (getBulkConnection() != null)
    {
        // connection already created
        return;
    }

    // print the info we will use to build the connection
    Utils.log("SalesforceService::createBulkConnection() entered" + "\n\tSession ID:    " + getSessionId()
        + "\n\tBulk Endpoint: " + getBulkEndpoint());

    // create partner connector configuration
    ConnectorConfig bulkConfig = getConnectorConfig(getServerUrl(), getSessionId());
    bulkConfig.setSessionId(getSessionId());
    bulkConfig.setRestEndpoint(getBulkEndpoint());
    bulkConfig.setCompression(true);

    // check if tracing is enabled
    if (getenv(SALESFORCE_TRACE_BULK) != null && getenv(SALESFORCE_TRACE_BULK).equalsIgnoreCase("1"))
    {
        // set this to true to see HTTP requests and responses on stdout
        bulkConfig.setTraceMessage(true);
        bulkConfig.setPrettyPrintXml(true);

        // this should only be false when doing debugging.
        bulkConfig.setCompression(false);
    }

    setBulkConnection(new BulkConnection(bulkConfig));
}
 

import com.sforce.async.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的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.AsyncApiException; //导入依赖的package包/类
@Override
public String addBatch(List<com.sforce.async.SObject> payload, JobInfo job) throws ResourceException {
	try {
		BatchRequest request = this.bulkConnection.createBatch(job);
		request.addSObjects(payload.toArray(new com.sforce.async.SObject[payload.size()]));
		return request.completeRequest().getId();
	} catch (AsyncApiException e) {
		throw new ResourceException(e);
	}
}