You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by je...@apache.org on 2021/10/28 15:22:07 UTC
[camel] branch main updated: CAMEL-17127: Support more salesforce
Bulk options
This is an automated email from the ASF dual-hosted git repository.
jeremyross pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0642209 CAMEL-17127: Support more salesforce Bulk options
0642209 is described below
commit 0642209c31f66756968236bd586aca32de3d1fc8
Author: Jeremy Ross <je...@gmail.com>
AuthorDate: Thu Oct 28 10:18:44 2021 -0500
CAMEL-17127: Support more salesforce Bulk options
Add support for salesforce bulk2GetQueryJobResults operation's `locator`
and `maxRecords`.
---
.../salesforce/SalesforceComponentConfigurer.java | 9 ++
.../salesforce/SalesforceEndpointConfigurer.java | 9 ++
.../salesforce/SalesforceEndpointUriFactory.java | 4 +-
.../camel/component/salesforce/salesforce.json | 4 +
.../src/main/docs/salesforce-component.adoc | 5 +-
.../salesforce/SalesforceEndpointConfig.java | 32 ++++
.../internal/client/BulkApiV2Client.java | 4 +-
.../internal/client/DefaultBulkApiV2Client.java | 17 +-
.../internal/processor/BulkApiV2Processor.java | 6 +-
.../BulkApiV2QueryJobIntegrationTest.java | 68 +++++++-
.../dsl/SalesforceComponentBuilderFactory.java | 39 +++++
.../dsl/SalesforceEndpointBuilderFactory.java | 174 +++++++++++++++++++++
12 files changed, 363 insertions(+), 8 deletions(-)
diff --git a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceComponentConfigurer.java b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceComponentConfigurer.java
index 22b0617..9144c49 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceComponentConfigurer.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceComponentConfigurer.java
@@ -112,6 +112,7 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
case "limit": getOrCreateConfig(target).setLimit(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "locator": getOrCreateConfig(target).setLocator(property(camelContext, java.lang.String.class, value)); return true;
case "loginconfig":
case "loginConfig": target.setLoginConfig(property(camelContext, org.apache.camel.component.salesforce.SalesforceLoginConfig.class, value)); return true;
case "loginurl":
@@ -120,6 +121,8 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "longPollingTransportProperties": target.setLongPollingTransportProperties(property(camelContext, java.util.Map.class, value)); return true;
case "maxbackoff":
case "maxBackoff": getOrCreateConfig(target).setMaxBackoff(property(camelContext, java.time.Duration.class, value).toMillis()); return true;
+ case "maxrecords":
+ case "maxRecords": getOrCreateConfig(target).setMaxRecords(property(camelContext, java.lang.Integer.class, value)); return true;
case "notfoundbehaviour":
case "notFoundBehaviour": getOrCreateConfig(target).setNotFoundBehaviour(property(camelContext, org.apache.camel.component.salesforce.NotFoundBehaviour.class, value)); return true;
case "notifyforfields":
@@ -287,6 +290,7 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
case "limit": return java.lang.Integer.class;
+ case "locator": return java.lang.String.class;
case "loginconfig":
case "loginConfig": return org.apache.camel.component.salesforce.SalesforceLoginConfig.class;
case "loginurl":
@@ -295,6 +299,8 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "longPollingTransportProperties": return java.util.Map.class;
case "maxbackoff":
case "maxBackoff": return long.class;
+ case "maxrecords":
+ case "maxRecords": return java.lang.Integer.class;
case "notfoundbehaviour":
case "notFoundBehaviour": return org.apache.camel.component.salesforce.NotFoundBehaviour.class;
case "notifyforfields":
@@ -463,6 +469,7 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
case "limit": return getOrCreateConfig(target).getLimit();
+ case "locator": return getOrCreateConfig(target).getLocator();
case "loginconfig":
case "loginConfig": return target.getLoginConfig();
case "loginurl":
@@ -471,6 +478,8 @@ public class SalesforceComponentConfigurer extends PropertyConfigurerSupport imp
case "longPollingTransportProperties": return target.getLongPollingTransportProperties();
case "maxbackoff":
case "maxBackoff": return getOrCreateConfig(target).getMaxBackoff();
+ case "maxrecords":
+ case "maxRecords": return getOrCreateConfig(target).getMaxRecords();
case "notfoundbehaviour":
case "notFoundBehaviour": return getOrCreateConfig(target).getNotFoundBehaviour();
case "notifyforfields":
diff --git a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointConfigurer.java b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointConfigurer.java
index 031c963..47c35d9 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointConfigurer.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointConfigurer.java
@@ -61,8 +61,11 @@ public class SalesforceEndpointConfigurer extends PropertyConfigurerSupport impl
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
case "limit": target.getConfiguration().setLimit(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "locator": target.getConfiguration().setLocator(property(camelContext, java.lang.String.class, value)); return true;
case "maxbackoff":
case "maxBackoff": target.getConfiguration().setMaxBackoff(property(camelContext, java.time.Duration.class, value).toMillis()); return true;
+ case "maxrecords":
+ case "maxRecords": target.getConfiguration().setMaxRecords(property(camelContext, java.lang.Integer.class, value)); return true;
case "notfoundbehaviour":
case "notFoundBehaviour": target.getConfiguration().setNotFoundBehaviour(property(camelContext, org.apache.camel.component.salesforce.NotFoundBehaviour.class, value)); return true;
case "notifyforfields":
@@ -174,8 +177,11 @@ public class SalesforceEndpointConfigurer extends PropertyConfigurerSupport impl
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
case "limit": return java.lang.Integer.class;
+ case "locator": return java.lang.String.class;
case "maxbackoff":
case "maxBackoff": return long.class;
+ case "maxrecords":
+ case "maxRecords": return java.lang.Integer.class;
case "notfoundbehaviour":
case "notFoundBehaviour": return org.apache.camel.component.salesforce.NotFoundBehaviour.class;
case "notifyforfields":
@@ -288,8 +294,11 @@ public class SalesforceEndpointConfigurer extends PropertyConfigurerSupport impl
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
case "limit": return target.getConfiguration().getLimit();
+ case "locator": return target.getConfiguration().getLocator();
case "maxbackoff":
case "maxBackoff": return target.getConfiguration().getMaxBackoff();
+ case "maxrecords":
+ case "maxRecords": return target.getConfiguration().getMaxRecords();
case "notfoundbehaviour":
case "notFoundBehaviour": return target.getConfiguration().getNotFoundBehaviour();
case "notifyforfields":
diff --git a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointUriFactory.java b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointUriFactory.java
index 40da452..1485f86 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointUriFactory.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/generated/java/org/apache/camel/component/salesforce/SalesforceEndpointUriFactory.java
@@ -20,7 +20,7 @@ public class SalesforceEndpointUriFactory extends org.apache.camel.support.compo
private static final Set<String> PROPERTY_NAMES;
private static final Set<String> SECRET_PROPERTY_NAMES;
static {
- Set<String> props = new HashSet<>(56);
+ Set<String> props = new HashSet<>(58);
props.add("initialReplayIdMap");
props.add("notifyForOperations");
props.add("sObjectQuery");
@@ -49,7 +49,9 @@ public class SalesforceEndpointUriFactory extends org.apache.camel.support.compo
props.add("notifyForOperationUpdate");
props.add("sObjectIdName");
props.add("replayId");
+ props.add("locator");
props.add("exceptionHandler");
+ props.add("maxRecords");
props.add("pkChunkingParent");
props.add("batchId");
props.add("notifyForOperationUndelete");
diff --git a/components/camel-salesforce/camel-salesforce-component/src/generated/resources/org/apache/camel/component/salesforce/salesforce.json b/components/camel-salesforce/camel-salesforce-component/src/generated/resources/org/apache/camel/component/salesforce/salesforce.json
index 6bfb0d7..97232a9 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/generated/resources/org/apache/camel/component/salesforce/salesforce.json
+++ b/components/camel-salesforce/camel-salesforce-component/src/generated/resources/org/apache/camel/component/salesforce/salesforce.json
@@ -41,7 +41,9 @@
"instanceId": { "kind": "property", "displayName": "Instance Id", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Salesforce1 Analytics report execution instance ID" },
"jobId": { "kind": "property", "displayName": "Job Id", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Bulk API Job ID" },
"limit": { "kind": "property", "displayName": "Limit", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Limit on number of returned records. Applicable to some of the API, check the Salesforce documentation." },
+ "locator": { "kind": "property", "displayName": "Locator", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Locator provided by salesforce Bulk 2.0 API for use in getting results for a Query job." },
"maxBackoff": { "kind": "property", "displayName": "Max Backoff", "group": "common", "label": "", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "30000", "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Maximum backoff interval for Streaming connection restart attempts for failures beyond CometD auto-reconnect." },
+ "maxRecords": { "kind": "property", "displayName": "Max Records", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "The maximum number of records to retrieve per set of results for a Bulk 2.0 Query. The request is still subject to the size l [...]
"notFoundBehaviour": { "kind": "property", "displayName": "Not Found Behaviour", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.salesforce.NotFoundBehaviour", "enum": [ "EXCEPTION", "NULL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "EXCEPTION", "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Sets the be [...]
"notifyForFields": { "kind": "property", "displayName": "Notify For Fields", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.salesforce.internal.dto.NotifyForFieldsEnum", "enum": [ "ALL", "REFERENCED", "SELECT", "WHERE" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Notify for [...]
"notifyForOperationCreate": { "kind": "property", "displayName": "Notify For Operation Create", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "config", "description": "Notify for create operation, defaults to false (API version = 29.0)" },
@@ -128,7 +130,9 @@
"instanceId": { "kind": "parameter", "displayName": "Instance Id", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Salesforce1 Analytics report execution instance ID" },
"jobId": { "kind": "parameter", "displayName": "Job Id", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Bulk API Job ID" },
"limit": { "kind": "parameter", "displayName": "Limit", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Limit on number of returned records. Applicable to some of the API, check the Salesforce documentation." },
+ "locator": { "kind": "parameter", "displayName": "Locator", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Locator provided by salesforce Bulk 2.0 API for use in getting results for a Query job." },
"maxBackoff": { "kind": "parameter", "displayName": "Max Backoff", "group": "common", "label": "", "required": false, "type": "duration", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "30000", "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Maximum backoff interval for Streaming connection restart attempts for failures beyond CometD auto-recon [...]
+ "maxRecords": { "kind": "parameter", "displayName": "Max Records", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "The maximum number of records to retrieve per set of results for a Bulk 2.0 Query. The request is still subject to th [...]
"notFoundBehaviour": { "kind": "parameter", "displayName": "Not Found Behaviour", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.salesforce.NotFoundBehaviour", "enum": [ "EXCEPTION", "NULL" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "EXCEPTION", "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Set [...]
"notifyForFields": { "kind": "parameter", "displayName": "Notify For Fields", "group": "common", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.salesforce.internal.dto.NotifyForFieldsEnum", "enum": [ "ALL", "REFERENCED", "SELECT", "WHERE" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "No [...]
"notifyForOperationCreate": { "kind": "parameter", "displayName": "Notify For Operation Create", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "java.lang.Boolean", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.salesforce.SalesforceEndpointConfig", "configurationField": "configuration", "description": "Notify for create operation, defaults to false (API version = 29.0)" },
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc b/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
index 0ea7267..fc310d0 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/docs/salesforce-component.adoc
@@ -224,7 +224,10 @@ to be set in the `queryLocator` parameter on subsequent calls.
* *bulk2CreateQueryJob* - Create a bulk query job. Supply an instance of `QueryJob` in the message
body.
* *bulk2GetQueryJob* - Get a bulk query job. `jobId` parameter is required.
-* *bulk2GetQueryJobResults* - Get bulk query job results. `jobId` parameter is required.
+* *bulk2GetQueryJobResults* - Get bulk query job results. `jobId` parameter is required. Accepts
+`maxRecords` and `locator` parameters. Response message headers include `Sforce-NumberOfRecords` and
+`Sforce-Locator` headers. The value of `Sforce-Locator` can be passed into subsequent calls via the
+`locator` parameter.
* *bulk2AbortQueryJob* - Abort a bulk query job. `jobId` parameter is required.
* *bulk2DeleteQueryJob* - Delete a bulk query job. `jobId` parameter is required.
* *bulk2GetAllQueryJobs* - Get all jobs. Response body is an instance of `QueryJobs`. If the `done`
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
index 4b00e41..3f12b78 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
@@ -75,6 +75,8 @@ public class SalesforceEndpointConfig implements Cloneable {
public static final String BATCH_ID = "batchId";
public static final String RESULT_ID = "resultId";
public static final String QUERY_LOCATOR = "queryLocator";
+ public static final String LOCATOR = "locator";
+ public static final String MAX_RECORDS = "maxRecords";
public static final String PK_CHUNKING = "pkChunking";
public static final String PK_CHUNKING_CHUNK_SIZE = "pkChunkingChunkSize";
public static final String PK_CHUNKING_PARENT = "pkChunkingParent";
@@ -161,6 +163,10 @@ public class SalesforceEndpointConfig implements Cloneable {
@UriParam
private String queryLocator;
@UriParam
+ private String locator;
+ @UriParam(javaType = "java.lang.Integer")
+ private Integer maxRecords;
+ @UriParam
private Boolean pkChunking;
@UriParam
private Integer pkChunkingChunkSize;
@@ -509,6 +515,32 @@ public class SalesforceEndpointConfig implements Cloneable {
this.queryLocator = queryLocator;
}
+ public String getLocator() {
+ return locator;
+ }
+
+ /**
+ * Locator provided by salesforce Bulk 2.0 API for use in getting results for a Query job.
+ */
+ public void setLocator(String locator) {
+ this.locator = locator;
+ }
+
+ public Integer getMaxRecords() {
+ return maxRecords;
+ }
+
+ /**
+ * The maximum number of records to retrieve per set of results for a Bulk 2.0 Query. The request is still subject
+ * to the size limits. If you are working with a very large number of query results, you may experience a timeout
+ * before receiving all the data from Salesforce. To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This splits the results into smaller sets with this
+ * value as the maximum size.
+ */
+ public void setMaxRecords(Integer maxRecords) {
+ this.maxRecords = maxRecords;
+ }
+
public Boolean getPkChunking() {
return pkChunking;
}
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/BulkApiV2Client.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/BulkApiV2Client.java
index 94832d4..397996f 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/BulkApiV2Client.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/BulkApiV2Client.java
@@ -77,7 +77,9 @@ public interface BulkApiV2Client {
void getQueryJob(String jobId, Map<String, List<String>> headers, QueryJobResponseCallback callback);
- void getQueryJobResults(String jobId, Map<String, List<String>> headers, StreamResponseCallback callback);
+ void getQueryJobResults(
+ String jobId, String locator, Integer maxRecords, Map<String, List<String>> headers,
+ StreamResponseCallback callback);
void changeQueryJobState(
String jobId, JobStateEnum state, Map<String, List<String>> headers, QueryJobResponseCallback callback);
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
index 2fbad6b..38f11b3 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
@@ -200,8 +200,21 @@ public class DefaultBulkApiV2Client extends AbstractClientBase implements BulkAp
}
@Override
- public void getQueryJobResults(String jobId, Map<String, List<String>> headers, StreamResponseCallback callback) {
- final Request request = getRequest(HttpMethod.GET, queryJobUrl(jobId) + "/results", headers);
+ public void getQueryJobResults(
+ String jobId, String locator, Integer maxRecords, Map<String, List<String>> headers,
+ StreamResponseCallback callback) {
+ String query = null;
+ if (locator != null) {
+ query = "locator=" + locator;
+ }
+ if (maxRecords != null) {
+ query = (query != null ? query + "&" : "") + "maxRecords=" + maxRecords;
+ }
+ String url = queryJobUrl(jobId) + "/results";
+ if (query != null) {
+ url = url + "?" + query;
+ }
+ final Request request = getRequest(HttpMethod.GET, url, headers);
doRequestWithCsvResponse(callback, request);
}
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiV2Processor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiV2Processor.java
index 6a6a001..d780c60 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiV2Processor.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiV2Processor.java
@@ -37,6 +37,8 @@ import org.apache.camel.component.salesforce.internal.client.DefaultBulkApiV2Cli
import org.apache.camel.support.service.ServiceHelper;
import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.JOB_ID;
+import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.LOCATOR;
+import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.MAX_RECORDS;
import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.QUERY_LOCATOR;
import static org.apache.camel.component.salesforce.internal.client.BulkApiV2Client.JobResponseCallback;
import static org.apache.camel.component.salesforce.internal.client.BulkApiV2Client.ResponseCallback;
@@ -312,7 +314,9 @@ public class BulkApiV2Processor extends AbstractSalesforceProcessor {
private void processGetQueryJobResults(Exchange exchange, AsyncCallback callback)
throws SalesforceException {
String jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
- bulkClient.getQueryJobResults(jobId, determineHeaders(exchange),
+ String locator = getParameter(LOCATOR, exchange, false, true);
+ Integer maxRecords = getParameter(MAX_RECORDS, exchange, false, true, Integer.class);
+ bulkClient.getQueryJobResults(jobId, locator, maxRecords, determineHeaders(exchange),
new StreamResponseCallback() {
@Override
public void onResponse(InputStream inputStream, Map<String, String> headers, SalesforceException ex) {
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiV2QueryJobIntegrationTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiV2QueryJobIntegrationTest.java
index e554651..ad969f4 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiV2QueryJobIntegrationTest.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiV2QueryJobIntegrationTest.java
@@ -18,16 +18,27 @@ package org.apache.camel.component.salesforce;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Exchange;
import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.api.dto.DeleteSObjectResult;
+import org.apache.camel.component.salesforce.api.dto.SaveSObjectResult;
import org.apache.camel.component.salesforce.api.dto.bulkv2.JobStateEnum;
import org.apache.camel.component.salesforce.api.dto.bulkv2.OperationEnum;
import org.apache.camel.component.salesforce.api.dto.bulkv2.QueryJob;
import org.apache.camel.component.salesforce.api.dto.bulkv2.QueryJobs;
+import org.apache.camel.component.salesforce.dto.generated.Contact;
+import org.apache.camel.component.salesforce.dto.generated.QueryRecordsContact;
+import org.apache.camel.support.DefaultExchange;
import org.apache.commons.io.IOUtils;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -37,6 +48,18 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@SuppressWarnings("BusyWait")
public class BulkApiV2QueryJobIntegrationTest extends AbstractSalesforceTestBase {
+ private static final String LAST_NAME = "CamelBulkTest";
+
+ @BeforeEach
+ private void setup() {
+ createContacts();
+ }
+
+ @AfterEach
+ private void teardown() {
+ deleteContacts();
+ }
+
@Test
public void testQueryLifecycle() throws Exception {
QueryJob job = new QueryJob();
@@ -56,11 +79,29 @@ public class BulkApiV2QueryJobIntegrationTest extends AbstractSalesforceTestBase
job.getId(), QueryJob.class);
}
- InputStream is = template().requestBodyAndHeader("salesforce:bulk2GetQueryJobResults",
- "", "jobId", job.getId(), InputStream.class);
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader("jobId", job.getId());
+ exchange.getIn().setHeader("maxRecords", 1);
+ template().send("salesforce:bulk2GetQueryJobResults?maxRecords=1",
+ exchange);
+ InputStream is = exchange.getIn().getBody(InputStream.class);
assertNotNull(is, "Query Job results");
List<String> results = IOUtils.readLines(is, StandardCharsets.UTF_8);
assertTrue(results.size() > 0, "Query Job results");
+ assertTrue(exchange.getIn().getHeaders().containsKey("Sforce-Locator"));
+ String locator = exchange.getIn().getHeader("Sforce-Locator", String.class);
+ assertNotNull(locator);
+
+ exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader("jobId", job.getId());
+ exchange.getIn().setHeader("locator", locator);
+ template().send("salesforce:bulk2GetQueryJobResults",
+ exchange);
+ is = exchange.getIn().getBody(InputStream.class);
+ assertNotNull(is, "Query Job results");
+ results = IOUtils.readLines(is, StandardCharsets.UTF_8);
+ assertTrue(results.size() > 0, "Query Job results");
+ assertNotNull(exchange.getIn().getHeader("Sforce-Locator"));
}
@Test
@@ -141,4 +182,27 @@ public class BulkApiV2QueryJobIntegrationTest extends AbstractSalesforceTestBase
QueryJobs.class);
assertNotNull(jobs);
}
+
+ private void createContacts() {
+ List<Contact> contacts = new ArrayList<>(2);
+ IntStream.range(0, 2).forEach(i -> {
+ Contact c = new Contact();
+ c.setLastName(LAST_NAME);
+ c.setFirstName("TestFirst");
+ contacts.add(c);
+ });
+ final List<SaveSObjectResult> results = (List<SaveSObjectResult>) template.requestBody(
+ "salesforce:compositeCreateSObjectCollections", contacts);
+ results.forEach(r -> assertTrue(r.getSuccess()));
+ }
+
+ private void deleteContacts() {
+ final List<Contact> records
+ = template.requestBody("salesforce:query?sObjectClass=" + QueryRecordsContact.class.getName(),
+ "SELECT Id FROM Contact WHERE LastName = '" + LAST_NAME + "'", QueryRecordsContact.class).getRecords();
+ final List<DeleteSObjectResult> results
+ = (List<DeleteSObjectResult>) template.requestBody("salesforce:compositeDeleteSObjectCollections",
+ records.stream().map(contact -> contact.getId()).collect(Collectors.toList()));
+ results.forEach(r -> assertTrue(r.getSuccess()));
+ }
}
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SalesforceComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SalesforceComponentBuilderFactory.java
index 109b257..74bb937 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SalesforceComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SalesforceComponentBuilderFactory.java
@@ -336,6 +336,21 @@ public interface SalesforceComponentBuilderFactory {
return this;
}
/**
+ * Locator provided by salesforce Bulk 2.0 API for use in getting
+ * results for a Query job.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ *
+ * @param locator the value to set
+ * @return the dsl builder
+ */
+ default SalesforceComponentBuilder locator(java.lang.String locator) {
+ doSetProperty("locator", locator);
+ return this;
+ }
+ /**
* Maximum backoff interval for Streaming connection restart attempts
* for failures beyond CometD auto-reconnect.
*
@@ -352,6 +367,28 @@ public interface SalesforceComponentBuilderFactory {
return this;
}
/**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceComponentBuilder maxRecords(
+ java.lang.Integer maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
* Sets the behaviour of 404 not found status received from Salesforce
* API. Should the body be set to NULL NotFoundBehaviour#NULL or should
* a exception be signaled on the exchange NotFoundBehaviour#EXCEPTION -
@@ -1517,7 +1554,9 @@ public interface SalesforceComponentBuilderFactory {
case "instanceId": getOrCreateConfiguration((SalesforceComponent) component).setInstanceId((java.lang.String) value); return true;
case "jobId": getOrCreateConfiguration((SalesforceComponent) component).setJobId((java.lang.String) value); return true;
case "limit": getOrCreateConfiguration((SalesforceComponent) component).setLimit((java.lang.Integer) value); return true;
+ case "locator": getOrCreateConfiguration((SalesforceComponent) component).setLocator((java.lang.String) value); return true;
case "maxBackoff": getOrCreateConfiguration((SalesforceComponent) component).setMaxBackoff((long) value); return true;
+ case "maxRecords": getOrCreateConfiguration((SalesforceComponent) component).setMaxRecords((java.lang.Integer) value); return true;
case "notFoundBehaviour": getOrCreateConfiguration((SalesforceComponent) component).setNotFoundBehaviour((org.apache.camel.component.salesforce.NotFoundBehaviour) value); return true;
case "notifyForFields": getOrCreateConfiguration((SalesforceComponent) component).setNotifyForFields((org.apache.camel.component.salesforce.internal.dto.NotifyForFieldsEnum) value); return true;
case "notifyForOperationCreate": getOrCreateConfiguration((SalesforceComponent) component).setNotifyForOperationCreate((java.lang.Boolean) value); return true;
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SalesforceEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SalesforceEndpointBuilderFactory.java
index 0a9ac5a..82f9f25 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SalesforceEndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SalesforceEndpointBuilderFactory.java
@@ -404,6 +404,21 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * Locator provided by salesforce Bulk 2.0 API for use in getting
+ * results for a Query job.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ *
+ * @param locator the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointConsumerBuilder locator(String locator) {
+ doSetProperty("locator", locator);
+ return this;
+ }
+ /**
* Maximum backoff interval for Streaming connection restart attempts
* for failures beyond CometD auto-reconnect.
*
@@ -436,6 +451,49 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointConsumerBuilder maxRecords(Integer maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option will be converted to a
+ * <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointConsumerBuilder maxRecords(String maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
* Sets the behaviour of 404 not found status received from Salesforce
* API. Should the body be set to NULL NotFoundBehaviour#NULL or should
* a exception be signaled on the exchange NotFoundBehaviour#EXCEPTION -
@@ -1610,6 +1668,21 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * Locator provided by salesforce Bulk 2.0 API for use in getting
+ * results for a Query job.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ *
+ * @param locator the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointProducerBuilder locator(String locator) {
+ doSetProperty("locator", locator);
+ return this;
+ }
+ /**
* Maximum backoff interval for Streaming connection restart attempts
* for failures beyond CometD auto-reconnect.
*
@@ -1642,6 +1715,49 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointProducerBuilder maxRecords(Integer maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option will be converted to a
+ * <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointProducerBuilder maxRecords(String maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
* Sets the behaviour of 404 not found status received from Salesforce
* API. Should the body be set to NULL NotFoundBehaviour#NULL or should
* a exception be signaled on the exchange NotFoundBehaviour#EXCEPTION -
@@ -2839,6 +2955,21 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * Locator provided by salesforce Bulk 2.0 API for use in getting
+ * results for a Query job.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ *
+ * @param locator the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointBuilder locator(String locator) {
+ doSetProperty("locator", locator);
+ return this;
+ }
+ /**
* Maximum backoff interval for Streaming connection restart attempts
* for failures beyond CometD auto-reconnect.
*
@@ -2871,6 +3002,49 @@ public interface SalesforceEndpointBuilderFactory {
return this;
}
/**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option is a: <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointBuilder maxRecords(Integer maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
+ * The maximum number of records to retrieve per set of results for a
+ * Bulk 2.0 Query. The request is still subject to the size limits. If
+ * you are working with a very large number of query results, you may
+ * experience a timeout before receiving all the data from Salesforce.
+ * To prevent a timeout, specify the maximum number of records your
+ * client is expecting to receive in the maxRecords parameter. This
+ * splits the results into smaller sets with this value as the maximum
+ * size.
+ *
+ * The option will be converted to a
+ * <code>java.lang.Integer</code> type.
+ *
+ * Group: common
+ *
+ * @param maxRecords the value to set
+ * @return the dsl builder
+ */
+ default SalesforceEndpointBuilder maxRecords(String maxRecords) {
+ doSetProperty("maxRecords", maxRecords);
+ return this;
+ }
+ /**
* Sets the behaviour of 404 not found status received from Salesforce
* API. Should the body be set to NULL NotFoundBehaviour#NULL or should
* a exception be signaled on the exchange NotFoundBehaviour#EXCEPTION -