import com.sforce.async.BulkConnection; //导入依赖的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.BulkConnection; //导入依赖的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.BulkConnection; //导入依赖的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.BulkConnection; //导入依赖的package包/类
public static BulkConnection getBulkConnection(
    ConnectorConfig partnerConfig,
    ForceConfigBean conf
) throws ConnectionException, AsyncApiException, StageException, URISyntaxException {
  // When PartnerConnection is instantiated, a login is implicitly
  // executed and, if successful,
  // a valid session is stored in the ConnectorConfig instance.
  // Use this key to initialize a BulkConnection:
  ConnectorConfig config = conf.mutualAuth.useMutualAuth
      ? new MutualAuthConnectorConfig(conf.mutualAuth.getUnderlyingConfig().getSslContext())
      : new ConnectorConfig();
  config.setSessionId(partnerConfig.getSessionId());

  // The endpoint for the Bulk API service is the same as for the normal
  // SOAP uri until the /Soap/ part. From here it's '/async/versionNumber'
  String soapEndpoint = partnerConfig.getServiceEndpoint();
  String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/"))
      + "async/" + conf.apiVersion;
  config.setRestEndpoint(restEndpoint);
  config.setCompression(conf.useCompression);
  config.setTraceMessage(conf.showTrace);
  config.setSessionRenewer(partnerConfig.getSessionRenewer());

  setProxyConfig(conf, config);

  BulkConnection bulkConnection = new BulkConnection(config);

  if (conf.mutualAuth.useMutualAuth) {
    setupMutualAuthBulk(config, conf.mutualAuth);
  }

  return bulkConnection;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public List<BatchInfo> createBatchesForApexClasses(BulkConnection bulkConnection, JobInfo jobInfo,
		String[] testClassesAsArray) {

	List<BatchInfo> batchInfos = new ArrayList<BatchInfo>();
	String stringToFeedIntoTheBatch = "ApexClassId\n";
	for (String testClass : testClassesAsArray) {
		stringToFeedIntoTheBatch += testClass;
		stringToFeedIntoTheBatch += "\n";
	}
	InputStream inputStream = new ByteArrayInputStream(stringToFeedIntoTheBatch.getBytes());

	batchInfos = createBatch(inputStream, batchInfos, jobInfo, bulkConnection);

	return batchInfos;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
/**
 * Wait for a job to complete by polling the Bulk API.
 * 
 * @param connection
 *            BulkConnection used to check results.
 * @param job
 *            The job awaiting completion.
 * @param batchInfoList
 *            List of batches for this job.
 * @throws AsyncApiException
 */
public void awaitCompletion(BulkConnection connection, JobInfo job, List<BatchInfo> batchInfoList)
		throws AsyncApiException {
	long sleepTime = 0L;
	Set<String> incompleteBatchInfos = new HashSet<String>();
	for (BatchInfo bi : batchInfoList) {
		incompleteBatchInfos.add(bi.getId());
	}
	while (!incompleteBatchInfos.isEmpty()) {
		try {
			Thread.sleep(sleepTime);
		} catch (InterruptedException e) {
			ApexUnitUtils
					.shutDownWithDebugLog(e, "InterruptedException encountered while the thread was attempting to sleep");
		}
		LOG.debug("Awaiting results... Batches remaining for processing: " + incompleteBatchInfos.size());
		sleepTime = 10000L;
		BatchInfo[] statusList = connection.getBatchInfoList(job.getId()).getBatchInfo();
		for (BatchInfo batchInfo : statusList) {
			// Retain the BatchInfo's which are in InProgress and Queued
			// status,
			// Remove the rest from the incompleteBatchInfos
			if (batchInfo.getState() == BatchStateEnum.Completed) {
				if (incompleteBatchInfos.remove(batchInfo.getId())) {
					LOG.debug("BATCH STATUS:" + batchInfo.getStateMessage());
				}
			} else if (batchInfo.getState() == BatchStateEnum.NotProcessed) {
				LOG.info("Batch " + batchInfo.getId() + " did not process, terminating it");
				incompleteBatchInfos.remove(batchInfo.getId());
			} else if (batchInfo.getState() == BatchStateEnum.Failed) {
				ApexUnitUtils.shutDownWithErrMsg("BATCH STATUS:" + batchInfo.getStateMessage());
			}
		}
	}
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
/**
 * Gets the results of the operation and checks for errors.
 */
public List<SaveResult> checkResults(BulkConnection bulkConnection, JobInfo job, List<BatchInfo> batchInfoList)
		throws AsyncApiException, IOException {
	LOG.debug("Checking Results.... ");
	List<SaveResult> saveResults = new ArrayList<SaveResult>();

	// batchInfoList was populated when batches were created and submitted
	for (BatchInfo batchInfo : batchInfoList) {
		CSVReader csvReaderForBatchResultStream = new CSVReader(
				bulkConnection.getBatchResultStream(job.getId(), batchInfo.getId()));
		List<String> resultHeader = csvReaderForBatchResultStream.nextRecord();
		int resultCols = resultHeader.size();

		List<String> batchResultStream = null;
		while ((batchResultStream = csvReaderForBatchResultStream.nextRecord()) != null) {

			Map<String, String> resultInfo = new HashMap<String, String>();
			for (int i = 0; i < resultCols; i++) {
				resultInfo.put(resultHeader.get(i), batchResultStream.get(i));
			}
			SaveResult sr = new SaveResult();
			sr.setId(resultInfo.get("Id"));
			boolean success = Boolean.valueOf(resultInfo.get("Success"));
			sr.setSuccess(success);

			if (!success) {
				if (resultInfo.get("Error") != null && StringUtils.isNotEmpty(resultInfo.get("Error"))) {
					ApexUnitUtils.shutDownWithErrMsg(
							"Error while fetching results for the batch job" + resultInfo.get("Error"));
				}
			}

			saveResults.add(sr);
		}
	}

	return saveResults;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
private BulkConnection createBulkConnection() {
	
	BulkConnectionConnectorConfig bcConnectorConfig = new BulkConnectionConnectorConfig();
	ConnectorConfig config = bcConnectorConfig.createConfig();
	try {
		bulkConnection = new BulkConnection(config);
		LOG.info("Bulk connection established.");
	} catch (AsyncApiException e) {
		ApexUnitUtils
				.shutDownWithDebugLog(e, "Caught AsyncApiException exception while trying to deal with bulk connection: "
						+ e.getMessage());
	}
	return bulkConnection;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public SalesforceBulkRuntime(BulkConnection bulkConnection) throws IOException {
    this.bulkConnection = bulkConnection;
    if (this.bulkConnection == null) {
        throw new RuntimeException(
                "Please check \"Bulk Connection\" checkbox in the setting of the referenced tSalesforceConnection.");
    }
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
protected BulkConnection connectBulk(ConnectorConfig config) throws ComponentException {
    /*
     * When PartnerConnection is instantiated, a login is implicitly executed and, if successful, a valid session is
     * stored in the ConnectorConfig instance. Use this key to initialize a BulkConnection:
     */
    ConnectorConfig bulkConfig = new ConnectorConfig();
    bulkConfig.setSessionId(config.getSessionId());
    // For session renew
    bulkConfig.setSessionRenewer(config.getSessionRenewer());
    bulkConfig.setUsername(config.getUsername());
    bulkConfig.setPassword(config.getPassword());
    /*
     * The endpoint for the Bulk API service is the same as for the normal SOAP uri until the /Soap/ part. From here
     * it's '/async/versionNumber'
     */
    String soapEndpoint = config.getServiceEndpoint();
    // set it by a default property file
    String api_version = "34.0";
    String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/")) + "async/" + api_version;
    bulkConfig.setRestEndpoint(restEndpoint);
    bulkConfig.setCompression(true);// This should only be false when doing debugging.
    bulkConfig.setTraceMessage(false);
    bulkConfig.setValidateSchema(false);
    try {
        return new BulkConnection(bulkConfig);
    } catch (AsyncApiException e) {
        throw new ComponentException(e);
    }
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
protected BulkConnection connectBulk(ConnectorConfig config) throws ComponentException {
    final SalesforceConnectionProperties connProps = getConnectionProperties();
    /*
     * When PartnerConnection is instantiated, a login is implicitly executed and, if successful, a valid session id is
     * stored in the ConnectorConfig instance. Use this key to initialize a BulkConnection:
     */
    ConnectorConfig bulkConfig = new ConnectorConfig();
    setProxy(bulkConfig);
    bulkConfig.setSessionId(config.getSessionId());
    // For session renew
    bulkConfig.setSessionRenewer(config.getSessionRenewer());
    bulkConfig.setUsername(config.getUsername());
    bulkConfig.setPassword(config.getPassword());
    /*
     * The endpoint for the Bulk API service is the same as for the normal SOAP uri until the /Soap/ part. From here
     * it's '/async/versionNumber'
     */
    String soapEndpoint = config.getServiceEndpoint();
    // Service endpoint should be like this:
    // https://ap1.salesforce.com/services/Soap/u/37.0/00D90000000eSq3
    String apiVersion = soapEndpoint.substring(soapEndpoint.lastIndexOf("/services/Soap/u/") + 17);
    apiVersion = apiVersion.substring(0, apiVersion.indexOf("/"));
    String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/")) + "async/" + apiVersion;
    bulkConfig.setRestEndpoint(restEndpoint);
    // This should only be false when doing debugging.
    bulkConfig.setCompression(connProps.needCompression.getValue());
    bulkConfig.setTraceMessage(connProps.httpTraceMessage.getValue());
    bulkConfig.setValidateSchema(false);
    try {
        return new BulkConnection(bulkConfig);
    } catch (AsyncApiException e) {
        throw new ComponentException(e);
    }
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    inputProperties = new TSalesforceInputProperties("input");

    conn = mock(BulkConnection.class);

    runtime = new SalesforceBulkRuntime(conn);
    assertTrue(conn == runtime.getBulkConnection());
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
private BulkConnection getBulkConnection()
{
    return _bConn;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
private void setBulkConnection(BulkConnection bConn)
{
    this._bConn = bConn;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
/**
 * Login to salesforce
   * @return login status
 */
public boolean bulkApiLogin() throws Exception {
  this.log.info("Authenticating salesforce bulk api");
  boolean success = false;
  String hostName = this.workUnit.getProp(ConfigurationKeys.SOURCE_CONN_HOST_NAME);
  String apiVersion = this.workUnit.getProp(ConfigurationKeys.SOURCE_CONN_VERSION);
  if (Strings.isNullOrEmpty(apiVersion)) {
    apiVersion = "29.0";
  }

  String soapAuthEndPoint = hostName + SALESFORCE_SOAP_AUTH_SERVICE + "/" + apiVersion;
  try {
    ConnectorConfig partnerConfig = new ConnectorConfig();
    if (super.workUnitState.contains(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL)
        && !super.workUnitState.getProp(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL).isEmpty()) {
      partnerConfig.setProxy(super.workUnitState.getProp(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL),
          super.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_CONN_USE_PROXY_PORT));
    }

    partnerConfig.setUsername(this.workUnit.getProp(ConfigurationKeys.SOURCE_CONN_USERNAME));
    partnerConfig.setPassword(PasswordManager.getInstance(this.workUnit)
        .readPassword(this.workUnit.getProp(ConfigurationKeys.SOURCE_CONN_PASSWORD)));
    partnerConfig.setAuthEndpoint(soapAuthEndPoint);
    PartnerConnection connection = new PartnerConnection(partnerConfig);
    String soapEndpoint = partnerConfig.getServiceEndpoint();
    String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/")) + "async/" + apiVersion;

    ConnectorConfig config = new ConnectorConfig();
    config.setSessionId(partnerConfig.getSessionId());
    config.setRestEndpoint(restEndpoint);
    config.setCompression(true);
    config.setTraceFile("traceLogs.txt");
    config.setTraceMessage(false);
    config.setPrettyPrintXml(true);

    if (super.workUnitState.contains(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL)
        && !super.workUnitState.getProp(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL).isEmpty()) {
      config.setProxy(super.workUnitState.getProp(ConfigurationKeys.SOURCE_CONN_USE_PROXY_URL),
          super.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_CONN_USE_PROXY_PORT));
    }

    this.bulkConnection = new BulkConnection(config);
    success = true;
  } catch (Exception e) {
    throw new Exception("Failed to connect to salesforce bulk api; error - " + e.getMessage(), e);
  }
  return success;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public ForceBulkWriter(Map<String, String> fieldMappings, BulkConnection bulkConnection, Target.Context context) {
  super(fieldMappings);
  this.bulkConnection = bulkConnection;
  this.context = context;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public void closeJob(BulkConnection bulkConnection, String jobId) throws AsyncApiException {
	JobInfo job = new JobInfo();
	job.setId(jobId);
	job.setState(JobStateEnum.Closed);
	bulkConnection.updateJob(job);
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public BulkConnection getBulkConnection() {
	if (bulkConnection == null) {
		createBulkConnection();
	}
	return bulkConnection;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public void setBulkConnection(BulkConnection bulkConnection) {
	this.bulkConnection = bulkConnection;
}
 

import com.sforce.async.BulkConnection; //导入依赖的package包/类
public BulkConnection getBulkConnection() {
    return bulkConnection;
}