camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhira...@apache.org
Subject [03/23] camel git commit: Added Salesforce component
Date Mon, 06 Apr 2015 20:37:55 GMT
http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/NotifyForOperationsEnum.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/NotifyForOperationsEnum.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/NotifyForOperationsEnum.java
new file mode 100644
index 0000000..807fef5
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/NotifyForOperationsEnum.java
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.dto;
+
+import org.codehaus.jackson.annotate.JsonCreator;
+import org.codehaus.jackson.annotate.JsonValue;
+
+/**
+ * Salesforce Enumeration DTO for picklist NotifyForOperations
+ */
+public enum NotifyForOperationsEnum {
+
+    CREATE("Create"),
+    UPDATE("Update"),
+    ALL("All");
+
+    final String value;
+
+    private NotifyForOperationsEnum(String value) {
+        this.value = value;
+    }
+
+    @JsonValue
+    public String value() {
+        return this.value;
+    }
+
+    @JsonCreator
+    public static NotifyForOperationsEnum forValue(String value) {
+        for (NotifyForOperationsEnum e : NotifyForOperationsEnum.values()) {
+            if (e.value.equals(value)) {
+                return e;
+            }
+        }
+        throw new IllegalArgumentException(value);
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/PushTopic.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/PushTopic.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/PushTopic.java
new file mode 100644
index 0000000..2135a16
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/PushTopic.java
@@ -0,0 +1,100 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.dto;
+
+import com.thoughtworks.xstream.annotations.XStreamAlias;
+import com.thoughtworks.xstream.annotations.XStreamConverter;
+import org.codehaus.jackson.annotate.JsonProperty;
+import org.apache.camel.component.salesforce.api.PicklistEnumConverter;
+import org.apache.camel.component.salesforce.api.dto.AbstractSObjectBase;
+
+/**
+ * Salesforce DTO for SObject PushTopic
+ */
+@XStreamAlias("PushTopic")
+public class PushTopic extends AbstractSObjectBase {
+
+    private String Query;
+    private Double ApiVersion;
+    private Boolean IsActive;
+    @XStreamConverter(PicklistEnumConverter.class)
+    private NotifyForFieldsEnum NotifyForFields;
+    @XStreamConverter(PicklistEnumConverter.class)
+    private NotifyForOperationsEnum NotifyForOperations;
+    private String Description;
+
+    @JsonProperty("Query")
+    public String getQuery() {
+        return this.Query;
+    }
+
+    @JsonProperty("Query")
+    public void setQuery(String Query) {
+        this.Query = Query;
+    }
+
+    @JsonProperty("ApiVersion")
+    public Double getApiVersion() {
+        return this.ApiVersion;
+    }
+
+    @JsonProperty("ApiVersion")
+    public void setApiVersion(Double ApiVersion) {
+        this.ApiVersion = ApiVersion;
+    }
+
+    @JsonProperty("IsActive")
+    public Boolean getIsActive() {
+        return this.IsActive;
+    }
+
+    @JsonProperty("IsActive")
+    public void setIsActive(Boolean IsActive) {
+        this.IsActive = IsActive;
+    }
+
+    @JsonProperty("NotifyForFields")
+    public NotifyForFieldsEnum getNotifyForFields() {
+        return this.NotifyForFields;
+    }
+
+    @JsonProperty("NotifyForFields")
+    public void setNotifyForFields(NotifyForFieldsEnum NotifyForFields) {
+        this.NotifyForFields = NotifyForFields;
+    }
+
+    @JsonProperty("NotifyForOperations")
+    public NotifyForOperationsEnum getNotifyForOperations() {
+        return this.NotifyForOperations;
+    }
+
+    @JsonProperty("NotifyForOperations")
+    public void setNotifyForOperations(NotifyForOperationsEnum NotifyForOperations) {
+        this.NotifyForOperations = NotifyForOperations;
+    }
+
+    @JsonProperty("Description")
+    public String getDescription() {
+        return this.Description;
+    }
+
+    @JsonProperty("Description")
+    public void setDescription(String Description) {
+        this.Description = Description;
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/QueryRecordsPushTopic.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/QueryRecordsPushTopic.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/QueryRecordsPushTopic.java
new file mode 100644
index 0000000..4adc13c
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/QueryRecordsPushTopic.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.dto;
+
+import com.thoughtworks.xstream.annotations.XStreamImplicit;
+import org.apache.camel.component.salesforce.api.dto.AbstractQueryRecordsBase;
+
+import java.util.List;
+
+/**
+ * Salesforce Query Records DTO for PushTopic
+ */
+public class QueryRecordsPushTopic extends AbstractQueryRecordsBase {
+    @XStreamImplicit
+    private List<PushTopic> records;
+
+    public List<PushTopic> getRecords() {
+        return records;
+    }
+
+    public void setRecords(List<PushTopic> records) {
+        this.records = records;
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/RestErrors.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/RestErrors.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/RestErrors.java
new file mode 100644
index 0000000..caf59fe
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/dto/RestErrors.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.dto;
+
+import com.thoughtworks.xstream.annotations.XStreamAlias;
+import com.thoughtworks.xstream.annotations.XStreamImplicit;
+import org.apache.camel.component.salesforce.api.dto.RestError;
+
+import java.util.List;
+
+/**
+ * DTO for Salesforce REST errors
+ */
+@XStreamAlias("Errors")
+public class RestErrors {
+
+    @XStreamImplicit(itemFieldName = "Error")
+    private List<RestError> errors;
+
+    public List<RestError> getErrors() {
+        return errors;
+    }
+
+    public void setErrors(List<RestError> errors) {
+        this.errors = errors;
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java
new file mode 100644
index 0000000..1e40d18
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java
@@ -0,0 +1,538 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.camel.component.salesforce.SalesforceEndpoint;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.api.dto.AbstractSObjectBase;
+import org.apache.camel.component.salesforce.internal.PayloadFormat;
+import org.apache.camel.component.salesforce.internal.client.DefaultRestClient;
+import org.apache.camel.component.salesforce.internal.client.RestClient;
+
+import java.io.InputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Map;
+
+import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.*;
+
+public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor {
+
+    protected static final String RESPONSE_CLASS = AbstractRestProcessor.class.getName() + ".responseClass";
+
+    private RestClient restClient;
+    private Map<String, Class<?>> classMap;
+
+    public AbstractRestProcessor(SalesforceEndpoint endpoint) throws SalesforceException {
+        super(endpoint);
+
+        final PayloadFormat payloadFormat = endpoint.getConfiguration().getPayloadFormat();
+
+        this.restClient = new DefaultRestClient(httpClient, endpointConfigMap.get(API_VERSION),
+            payloadFormat.toString().toLowerCase() , session);
+
+        this.classMap = endpoint.getComponent().getClassMap();
+    }
+
+    @Override
+    public void start() throws Exception {
+        ServiceHelper.startService(restClient);
+    }
+
+    @Override
+    public void stop() throws Exception {
+        ServiceHelper.stopService(restClient);
+    }
+
+    @Override
+    public final boolean process(final Exchange exchange, final AsyncCallback callback) {
+
+        // pre-process request message
+        try {
+            processRequest(exchange);
+        } catch (SalesforceException e) {
+            exchange.setException(e);
+            callback.done(true);
+            return true;
+        } catch (RuntimeException e) {
+            exchange.setException(new SalesforceException(e.getMessage(), e));
+            callback.done(true);
+            return true;
+        }
+
+        // call Salesforce asynchronously
+        try {
+
+            // call Operation using REST client
+            switch (operationName) {
+                case GET_VERSIONS:
+                    restClient.getVersions(new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            // process response entity and create out message
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case GET_RESOURCES:
+                    restClient.getResources(new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case GET_GLOBAL_OBJECTS:
+                    restClient.getGlobalObjects(new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case GET_BASIC_INFO:
+                    String sObjectName = getParameter(SOBJECT_NAME, exchange, USE_BODY, NOT_OPTIONAL);
+                    restClient.getBasicInfo(sObjectName, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case GET_DESCRIPTION:
+                    sObjectName = getParameter(SOBJECT_NAME, exchange, USE_BODY, NOT_OPTIONAL);
+                    restClient.getDescription(sObjectName, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case GET_SOBJECT:
+                {
+                    String sObjectIdValue;
+                    // determine parameters from input AbstractSObject
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        sObjectIdValue = sObjectBase.getId();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectIdValue = getParameter(SOBJECT_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    final String sObjectId = sObjectIdValue;
+
+                    // use sObject name to load class
+                    setResponseClass(exchange, sObjectName);
+
+                    // get optional field list
+                    String fieldsValue = getParameter(SOBJECT_FIELDS, exchange, IGNORE_BODY, IS_OPTIONAL);
+                    String[] fields = null;
+                    if (fieldsValue != null) {
+                        fields = fieldsValue.split(",");
+                    }
+
+                    restClient.getSObject(sObjectName, sObjectId, fields, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                            restoreFields(exchange, sObjectBase, sObjectId, null, null);
+                        }
+                    });
+
+                    break;
+                }
+
+                case CREATE_SOBJECT:
+                {
+                    // determine parameters from input AbstractSObject
+                    AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                    }
+
+                    restClient.createSObject(sObjectName, getRequestStream(exchange),
+                        new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+
+                    break;
+                }
+
+                case UPDATE_SOBJECT:
+                {
+                    // determine parameters from input AbstractSObject
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    String sObjectId;
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        // remember the sObject Id
+                        sObjectId = sObjectBase.getId();
+                        // clear base object fields, which cannot be updated
+                        sObjectBase.clearBaseFields();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectId = getParameter(SOBJECT_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                    }
+
+                    final String finalsObjectId = sObjectId;
+                    restClient.updateSObject(sObjectName, sObjectId, getRequestStream(exchange),
+                        new RestClient.ResponseCallback() {
+                            @Override
+                            public void onResponse(InputStream response, SalesforceException exception) {
+                                processResponse(exchange, response, exception, callback);
+                                restoreFields(exchange, sObjectBase, finalsObjectId, null, null);
+                            }
+                        });
+
+                    break;
+                }
+
+                case DELETE_SOBJECT:
+                {
+                    // determine parameters from input AbstractSObject
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    String sObjectIdValue;
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        sObjectIdValue = sObjectBase.getId();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectIdValue = getParameter(SOBJECT_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    final String sObjectId = sObjectIdValue;
+
+                    restClient.deleteSObject(sObjectName, sObjectId, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                            restoreFields(exchange, sObjectBase, sObjectId, null, null);
+                        }
+                    });
+                    break;
+                }
+
+                case GET_SOBJECT_WITH_ID:
+                {
+                    Object oldValue = null;
+                    String sObjectExtIdValue;
+                    final String sObjectExtIdName = getParameter(SOBJECT_EXT_ID_NAME,
+                        exchange, IGNORE_BODY, NOT_OPTIONAL);
+
+                    // determine parameters from input AbstractSObject
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName);
+                        sObjectExtIdValue = oldValue.toString();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+
+                    // use sObject name to load class
+                    setResponseClass(exchange, sObjectName);
+
+                    final Object finalOldValue = oldValue;
+                    restClient.getSObjectWithId(sObjectName, sObjectExtIdName, sObjectExtIdValue,
+                        new RestClient.ResponseCallback() {
+                            @Override
+                            public void onResponse(InputStream response, SalesforceException exception) {
+                                processResponse(exchange, response, exception, callback);
+                                restoreFields(exchange, sObjectBase, null, sObjectExtIdName, finalOldValue);
+                            }
+                        });
+
+                    break;
+                }
+
+                case UPSERT_SOBJECT:
+                {
+                    String sObjectExtIdValue;
+                    final String sObjectExtIdName = getParameter(SOBJECT_EXT_ID_NAME, exchange,
+                        IGNORE_BODY, NOT_OPTIONAL);
+
+                    // determine parameters from input AbstractSObject
+                    Object oldValue = null;
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName);
+                        sObjectExtIdValue = oldValue.toString();
+                        // clear base object fields, which cannot be updated
+                        sObjectBase.clearBaseFields();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                    }
+
+                    final Object finalOldValue = oldValue;
+                    restClient.upsertSObject(sObjectName, sObjectExtIdName, sObjectExtIdValue,
+                        getRequestStream(exchange), new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                            restoreFields(exchange, sObjectBase, null, sObjectExtIdName, finalOldValue);
+                        }
+                    });
+
+                    break;
+                }
+
+                case DELETE_SOBJECT_WITH_ID:
+                {
+                    final String sObjectExtIdName = getParameter(SOBJECT_EXT_ID_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+
+                    // determine parameters from input AbstractSObject
+                    Object oldValue = null;
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    String sObjectExtIdValue;
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName);
+                        sObjectExtIdValue = oldValue.toString();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+
+                    final Object finalOldValue = oldValue;
+                    restClient.deleteSObjectWithId(sObjectName, sObjectExtIdName, sObjectExtIdValue,
+                        new RestClient.ResponseCallback() {
+                            @Override
+                            public void onResponse(InputStream response, SalesforceException exception) {
+                                processResponse(exchange, response, exception, callback);
+                                restoreFields(exchange, sObjectBase, null, sObjectExtIdName, finalOldValue);
+                            }
+                        });
+
+                    break;
+                }
+
+                case GET_BLOB_FIELD:
+                {
+                    // get blob field name
+                    final String sObjectBlobFieldName = getParameter(SOBJECT_BLOB_FIELD_NAME,
+                        exchange, IGNORE_BODY, NOT_OPTIONAL);
+
+                    // determine parameters from input AbstractSObject
+                    final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class);
+                    String sObjectIdValue;
+                    if (sObjectBase != null) {
+                        sObjectName = sObjectBase.getClass().getSimpleName();
+                        sObjectIdValue = sObjectBase.getId();
+                    } else {
+                        sObjectName = getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        sObjectIdValue = getParameter(SOBJECT_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    final String sObjectId = sObjectIdValue;
+
+                    restClient.getBlobField(sObjectName, sObjectId, sObjectBlobFieldName,
+                        new RestClient.ResponseCallback() {
+                            @Override
+                            public void onResponse(InputStream response, SalesforceException exception) {
+                                processResponse(exchange, response, exception, callback);
+                                restoreFields(exchange, sObjectBase, sObjectId, null, null);
+                            }
+                        });
+                    break;
+                }
+
+                case QUERY:
+                    final String sObjectQuery = getParameter(SOBJECT_QUERY, exchange, USE_BODY, NOT_OPTIONAL);
+
+                    // use sObject name to load class
+                    setResponseClass(exchange, null);
+
+                    restClient.query(sObjectQuery, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case QUERY_MORE:
+                    // reuse SOBJECT_QUERY parameter name for nextRecordsUrl
+                    final String nextRecordsUrl = getParameter(SOBJECT_QUERY, exchange, USE_BODY, NOT_OPTIONAL);
+
+                    // use custom response class property
+                    setResponseClass(exchange, null);
+
+                    restClient.queryMore(nextRecordsUrl, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+                case SEARCH:
+                    final String sObjectSearch = getParameter(SOBJECT_SEARCH, exchange, USE_BODY, NOT_OPTIONAL);
+
+                    restClient.search(sObjectSearch, new RestClient.ResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream response, SalesforceException exception) {
+                            processResponse(exchange, response, exception, callback);
+                        }
+                    });
+                    break;
+
+            }
+
+        } catch (SalesforceException e) {
+            exchange.setException(new SalesforceException(
+                String.format("Error processing %s: [%s] \"%s\"",
+                    operationName, e.getStatusCode(), e.getMessage()),
+                e));
+            callback.done(true);
+            return true;
+        } catch (RuntimeException e) {
+            exchange.setException(new SalesforceException(
+                String.format("Unexpected Error processing %s: \"%s\"",
+                    operationName, e.getMessage()),
+                e));
+            callback.done(true);
+            return true;
+        }
+
+        // continue routing asynchronously
+        return false;
+    }
+
+    private void restoreFields(Exchange exchange, AbstractSObjectBase sObjectBase,
+                               String sObjectId, String sObjectExtIdName, Object oldValue) {
+        // restore fields
+        if (sObjectBase != null) {
+            // restore the Id if it was cleared
+            if (sObjectId != null) {
+                sObjectBase.setId(sObjectId);
+            }
+            // restore the external id if it was cleared
+            if (sObjectExtIdName != null && oldValue != null) {
+                try {
+                    setPropertyValue(sObjectBase, sObjectExtIdName, oldValue);
+                } catch (SalesforceException e) {
+                    // YES, the exchange may fail if the property cannot be reset!!!
+                    exchange.setException(e);
+                }
+            }
+        }
+    }
+
+    private void setPropertyValue(AbstractSObjectBase sObjectBase, String name, Object value) throws SalesforceException {
+        try {
+            // set the value with the set method
+            Method setMethod = sObjectBase.getClass().getMethod("set" + name, value.getClass());
+            setMethod.invoke(sObjectBase, value);
+        } catch (NoSuchMethodException e) {
+            throw new SalesforceException(
+                String.format("SObject %s does not have a field %s",
+                    sObjectBase.getClass().getName(), name),
+                e);
+        } catch (InvocationTargetException e) {
+            throw new SalesforceException(
+                String.format("Error setting value %s.%s",
+                    sObjectBase.getClass().getSimpleName(), name),
+                e);
+        } catch (IllegalAccessException e) {
+            throw new SalesforceException(
+                String.format("Error accessing value %s.%s",
+                    sObjectBase.getClass().getSimpleName(), name),
+                e);
+        }
+    }
+
+    private Object getAndClearPropertyValue(AbstractSObjectBase sObjectBase, String propertyName) throws SalesforceException {
+        try {
+            // obtain the value using the get method
+            Method getMethod = sObjectBase.getClass().getMethod("get" + propertyName);
+            Object value = getMethod.invoke(sObjectBase);
+
+            // clear the value with the set method
+            Method setMethod = sObjectBase.getClass().getMethod("set" + propertyName, getMethod.getReturnType());
+            setMethod.invoke(sObjectBase, new Object[] { null });
+
+            return value;
+        } catch (NoSuchMethodException e) {
+            throw new SalesforceException(
+                String.format("SObject %s does not have a field %s",
+                    sObjectBase.getClass().getSimpleName(), propertyName),
+                e);
+        } catch (InvocationTargetException e) {
+            throw new SalesforceException(
+                String.format("Error getting/setting value %s.%s",
+                    sObjectBase.getClass().getSimpleName(), propertyName),
+                e);
+        } catch (IllegalAccessException e) {
+            throw new SalesforceException(
+                String.format("Error accessing value %s.%s",
+                    sObjectBase.getClass().getSimpleName(), propertyName),
+                e);
+        }
+    }
+
+    // pre-process request message
+    protected abstract void processRequest(Exchange exchange) throws SalesforceException;
+
+    // get request stream from In message
+    protected abstract InputStream getRequestStream(Exchange exchange) throws SalesforceException;
+
+    private void setResponseClass(Exchange exchange, String sObjectName) throws SalesforceException {
+        Class<?> sObjectClass;
+
+        if (sObjectName != null) {
+            // lookup class from class map
+            sObjectClass = classMap.get(sObjectName);
+            if (null == sObjectClass) {
+                throw new SalesforceException(String.format("No class found for SObject %s", sObjectName), null);
+            }
+
+        } else {
+
+            // use custom response class property
+            final String className = getParameter(SOBJECT_CLASS, exchange, IGNORE_BODY, NOT_OPTIONAL);
+            try {
+                sObjectClass = endpoint.getComponent().getCamelContext()
+                    .getClassResolver().resolveMandatoryClass(className);
+            } catch (ClassNotFoundException e) {
+                throw new SalesforceException(
+                    String.format("SObject class not found %s, %s",
+                        className, e.getMessage()),
+                    e);
+            }
+        }
+        exchange.setProperty(RESPONSE_CLASS, sObjectClass);
+    }
+
+    // process response entity and set out message in exchange
+    protected abstract void processResponse(Exchange exchange, InputStream responseEntity, SalesforceException ex, AsyncCallback callback);
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractSalesforceProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractSalesforceProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractSalesforceProcessor.java
new file mode 100644
index 0000000..e784458
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractSalesforceProcessor.java
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.eclipse.jetty.client.HttpClient;
+import org.apache.camel.component.salesforce.SalesforceComponent;
+import org.apache.camel.component.salesforce.SalesforceEndpoint;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.internal.OperationName;
+import org.apache.camel.component.salesforce.internal.SalesforceSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+public abstract class AbstractSalesforceProcessor implements SalesforceProcessor {
+
+    protected static final boolean NOT_OPTIONAL = false;
+    protected static final boolean IS_OPTIONAL = true;
+    protected static final boolean USE_BODY = true;
+    protected static final boolean IGNORE_BODY = false;
+    protected final Logger LOG = LoggerFactory.getLogger(this.getClass());
+
+    protected final SalesforceEndpoint endpoint;
+    protected final Map<String, String> endpointConfigMap;
+
+    protected final OperationName operationName;
+    protected final SalesforceSession session;
+    protected final HttpClient httpClient;
+
+    public AbstractSalesforceProcessor(SalesforceEndpoint endpoint) {
+        this.endpoint = endpoint;
+        this.operationName = endpoint.getOperationName();
+        this.endpointConfigMap = endpoint.getConfiguration().toValueMap();
+
+        final SalesforceComponent component = endpoint.getComponent();
+        this.session = component.getSession();
+        this.httpClient = endpoint.getConfiguration().getHttpClient();
+    }
+
+    @Override
+    public abstract boolean process(Exchange exchange, AsyncCallback callback);
+
+    /**
+     * Gets value for a parameter from header, endpoint config, or exchange body (optional).
+     *
+     * @param exchange exchange to inspect
+     * @param convertInBody converts In body to String value if true
+     * @param propName name of property
+     * @param optional if {@code true} returns null, otherwise throws RestException
+     * @return value of property, or {@code null} for optional parameters if not found.
+     * @throws org.apache.camel.component.salesforce.api.SalesforceException if the property can't be found.
+     */
+    protected final String getParameter(String propName, Exchange exchange, boolean convertInBody, boolean optional) throws SalesforceException {
+        String propValue = exchange.getIn().getHeader(propName, String.class);
+        propValue = propValue == null ? endpointConfigMap.get(propName) : propValue;
+        propValue = (propValue == null && convertInBody) ? exchange.getIn().getBody(String.class) : propValue;
+
+        // error if property was not set
+        if (propValue == null && !optional) {
+            String msg = "Missing property " + propName;
+            throw new SalesforceException(msg, null);
+        }
+
+        return propValue;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiProcessor.java
new file mode 100644
index 0000000..6e070f7
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/BulkApiProcessor.java
@@ -0,0 +1,377 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import org.apache.camel.*;
+import org.apache.camel.converter.stream.StreamCacheConverter;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.camel.component.salesforce.SalesforceEndpoint;
+import org.apache.camel.component.salesforce.SalesforceEndpointConfig;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.api.dto.bulk.*;
+import org.apache.camel.component.salesforce.internal.client.BulkApiClient;
+import org.apache.camel.component.salesforce.internal.client.DefaultBulkApiClient;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.*;
+
+public class BulkApiProcessor extends AbstractSalesforceProcessor {
+
+    private BulkApiClient bulkClient;
+
+    public BulkApiProcessor(SalesforceEndpoint endpoint) throws SalesforceException {
+        super(endpoint);
+
+        this.bulkClient = new DefaultBulkApiClient(
+            endpointConfigMap.get(SalesforceEndpointConfig.API_VERSION), session, httpClient);
+    }
+
+    @Override
+    public boolean process(final Exchange exchange, final AsyncCallback callback) {
+
+        boolean done = false;
+        try {
+            switch (operationName) {
+                case CREATE_JOB:
+                    JobInfo jobBody = exchange.getIn().getMandatoryBody(JobInfo.class);
+                    bulkClient.createJob(jobBody, new BulkApiClient.JobInfoResponseCallback() {
+                        @Override
+                        public void onResponse(JobInfo jobInfo, SalesforceException ex) {
+                            processResponse(exchange, jobInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_JOB:
+                    jobBody = exchange.getIn().getBody(JobInfo.class);
+                    String jobId;
+                    if (jobBody != null) {
+                        jobId = jobBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getJob(jobId, new BulkApiClient.JobInfoResponseCallback() {
+                        @Override
+                        public void onResponse(JobInfo jobInfo, SalesforceException ex) {
+                            processResponse(exchange, jobInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case CLOSE_JOB:
+                    jobBody = exchange.getIn().getBody(JobInfo.class);
+                    if (jobBody != null) {
+                        jobId = jobBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.closeJob(jobId, new BulkApiClient.JobInfoResponseCallback() {
+                        @Override
+                        public void onResponse(JobInfo jobInfo, SalesforceException ex) {
+                            processResponse(exchange, jobInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case ABORT_JOB:
+                    jobBody = exchange.getIn().getBody(JobInfo.class);
+                    if (jobBody != null) {
+                        jobId = jobBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.abortJob(jobId, new BulkApiClient.JobInfoResponseCallback() {
+                        @Override
+                        public void onResponse(JobInfo jobInfo, SalesforceException ex) {
+                            processResponse(exchange, jobInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case CREATE_BATCH:
+                    // since request is in the body, use headers or endpoint params
+                    ContentType contentType = ContentType.fromValue(
+                        getParameter(CONTENT_TYPE, exchange, IGNORE_BODY, NOT_OPTIONAL));
+                    jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+
+                    InputStream request;
+                    try {
+                        request = exchange.getIn().getMandatoryBody(InputStream.class);
+                    } catch (CamelException e) {
+                        String msg = "Error preparing batch request: " + e.getMessage();
+                        throw new SalesforceException(msg, e);
+                    }
+
+                    bulkClient.createBatch(request, jobId, contentType, new BulkApiClient.BatchInfoResponseCallback() {
+                        @Override
+                        public void onResponse(BatchInfo batchInfo, SalesforceException ex) {
+                            processResponse(exchange, batchInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_BATCH:
+                    BatchInfo batchBody = exchange.getIn().getBody(BatchInfo.class);
+                    String batchId;
+                    if (batchBody != null) {
+                        jobId = batchBody.getJobId();
+                        batchId = batchBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        batchId = getParameter(BATCH_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getBatch(jobId, batchId, new BulkApiClient.BatchInfoResponseCallback() {
+                        @Override
+                        public void onResponse(BatchInfo batchInfo, SalesforceException ex) {
+                            processResponse(exchange, batchInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_ALL_BATCHES:
+                    jobBody = exchange.getIn().getBody(JobInfo.class);
+                    if (jobBody != null) {
+                        jobId = jobBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getAllBatches(jobId, new BulkApiClient.BatchInfoListResponseCallback() {
+                        @Override
+                        public void onResponse(List<BatchInfo> batchInfoList, SalesforceException ex) {
+                            processResponse(exchange, batchInfoList, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_REQUEST:
+                    batchBody = exchange.getIn().getBody(BatchInfo.class);
+                    if (batchBody != null) {
+                        jobId = batchBody.getJobId();
+                        batchId = batchBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        batchId = getParameter(BATCH_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+
+                    bulkClient.getRequest(jobId, batchId, new BulkApiClient.StreamResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream inputStream, SalesforceException ex) {
+                            // read the request stream into a StreamCache temp file
+                            // ensures the connection is read
+                            StreamCache body = null;
+                            if (inputStream != null) {
+                                try {
+                                    body = StreamCacheConverter.convertToStreamCache(inputStream, exchange);
+                                } catch (IOException e) {
+                                    String msg = "Error retrieving batch request: " + e.getMessage();
+                                    ex = new SalesforceException(msg, e);
+                                } finally {
+                                    // close the input stream to release the Http connection
+                                    try {
+                                        inputStream.close();
+                                    } catch (IOException ignore) {
+                                    }
+                                }
+                            }
+                            processResponse(exchange, body, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_RESULTS:
+                    batchBody = exchange.getIn().getBody(BatchInfo.class);
+                    if (batchBody != null) {
+                        jobId = batchBody.getJobId();
+                        batchId = batchBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        batchId = getParameter(BATCH_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getResults(jobId, batchId, new BulkApiClient.StreamResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream inputStream, SalesforceException ex) {
+                            // read the result stream into a StreamCache temp file
+                            // ensures the connection is read
+                            StreamCache body = null;
+                            if (inputStream != null) {
+                                try {
+                                    body = StreamCacheConverter.convertToStreamCache(inputStream, exchange);
+                                } catch (IOException e) {
+                                    String msg = "Error retrieving batch results: " + e.getMessage();
+                                    ex = new SalesforceException(msg, e);
+                                } finally {
+                                    // close the input stream to release the Http connection
+                                    try {
+                                        inputStream.close();
+                                    } catch (IOException ignore) {
+                                    }
+                                }
+                            }
+                            processResponse(exchange, body, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case CREATE_BATCH_QUERY:
+                    jobBody = exchange.getIn().getBody(JobInfo.class);
+                    String soqlQuery;
+                    if (jobBody != null) {
+                        jobId = jobBody.getId();
+                        contentType = jobBody.getContentType();
+                        // use SOQL query from header or endpoint config
+                        soqlQuery = getParameter(SOBJECT_QUERY, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        contentType = ContentType.fromValue(
+                            getParameter(CONTENT_TYPE, exchange, IGNORE_BODY, NOT_OPTIONAL));
+                        // reuse SOBJECT_QUERY property
+                        soqlQuery = getParameter(SOBJECT_QUERY, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.createBatchQuery(jobId, soqlQuery, contentType,
+                        new BulkApiClient.BatchInfoResponseCallback() {
+                        @Override
+                        public void onResponse(BatchInfo batchInfo, SalesforceException ex) {
+                            processResponse(exchange, batchInfo, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_QUERY_RESULT_IDS:
+                    batchBody = exchange.getIn().getBody(BatchInfo.class);
+                    if (batchBody != null) {
+                        jobId = batchBody.getJobId();
+                        batchId = batchBody.getId();
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        batchId = getParameter(BATCH_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getQueryResultIds(jobId, batchId, new BulkApiClient.QueryResultIdsCallback() {
+                        @Override
+                        public void onResponse(List<String> ids, SalesforceException ex) {
+                            processResponse(exchange, ids, ex, callback);
+                        }
+                    });
+
+                    break;
+
+                case GET_QUERY_RESULT:
+                    batchBody = exchange.getIn().getBody(BatchInfo.class);
+                    String resultId;
+                    if (batchBody != null) {
+                        jobId = batchBody.getJobId();
+                        batchId = batchBody.getId();
+                        resultId = getParameter(RESULT_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                    } else {
+                        jobId = getParameter(JOB_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        batchId = getParameter(BATCH_ID, exchange, IGNORE_BODY, NOT_OPTIONAL);
+                        resultId = getParameter(RESULT_ID, exchange, USE_BODY, NOT_OPTIONAL);
+                    }
+                    bulkClient.getQueryResult(jobId, batchId, resultId, new BulkApiClient.StreamResponseCallback() {
+                        @Override
+                        public void onResponse(InputStream inputStream, SalesforceException ex) {
+                            StreamCache body = null;
+                            if (inputStream != null) {
+                                // read the result stream into a StreamCache temp file
+                                // ensures the connection is read
+                                try {
+                                    body = StreamCacheConverter.convertToStreamCache(inputStream, exchange);
+                                } catch (IOException e) {
+                                    String msg = "Error retrieving query result: " + e.getMessage();
+                                    ex = new SalesforceException(msg, e);
+                                } finally {
+                                    // close the input stream to release the Http connection
+                                    try {
+                                        inputStream.close();
+                                    } catch (IOException e) {
+                                        // ignore
+                                    }
+                                }
+                            }
+                            processResponse(exchange, body, ex, callback);
+                        }
+                    });
+
+                    break;
+            }
+
+        } catch (SalesforceException e) {
+            exchange.setException(new SalesforceException(
+                String.format("Error processing %s: [%s] \"%s\"",
+                    operationName, e.getStatusCode(), e.getMessage()),
+                e));
+            callback.done(true);
+            done = true;
+        } catch (InvalidPayloadException e) {
+            exchange.setException(new SalesforceException(
+                String.format("Unexpected Error processing %s: \"%s\"",
+                    operationName, e.getMessage()),
+                e));
+            callback.done(true);
+            done = true;
+        } catch (RuntimeException e) {
+            exchange.setException(new SalesforceException(
+                String.format("Unexpected Error processing %s: \"%s\"",
+                    operationName, e.getMessage()),
+                e));
+            callback.done(true);
+            done = true;
+        }
+
+        // continue routing asynchronously if false
+        return done;
+    }
+
+    private void processResponse(Exchange exchange, Object body, SalesforceException ex, AsyncCallback callback) {
+        final Message out = exchange.getOut();
+        if (ex != null) {
+            exchange.setException(ex);
+        } else {
+            out.setBody(body);
+        }
+
+        // copy headers and attachments
+        out.getHeaders().putAll(exchange.getIn().getHeaders());
+        out.getAttachments().putAll(exchange.getIn().getAttachments());
+
+        // signal exchange completion
+        callback.done(false);
+    }
+
+    @Override
+    public void start() throws Exception {
+        ServiceHelper.startService(bulkClient);
+    }
+
+    @Override
+    public void stop() throws Exception {
+        // stop the client
+        ServiceHelper.stopService(bulkClient);
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java
new file mode 100644
index 0000000..3ac32c6
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java
@@ -0,0 +1,180 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.SerializationConfig;
+import org.codehaus.jackson.type.TypeReference;
+import org.eclipse.jetty.util.StringUtil;
+import org.apache.camel.component.salesforce.SalesforceEndpoint;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.api.dto.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+public class JsonRestProcessor extends AbstractRestProcessor {
+
+    // it is ok to use a single thread safe ObjectMapper
+    private final ObjectMapper objectMapper;
+    private static final String RESPONSE_TYPE = JsonRestProcessor.class.getName() + ".responseType";
+
+    public JsonRestProcessor(SalesforceEndpoint endpoint) throws SalesforceException {
+        super(endpoint);
+
+        this.objectMapper = new ObjectMapper();
+        // enable date time support including Joda DateTime
+        this.objectMapper.configure(SerializationConfig.Feature.WRITE_DATES_AS_TIMESTAMPS, false);
+    }
+
+    @Override
+    protected void processRequest(Exchange exchange) {
+
+        switch (operationName) {
+            case GET_VERSIONS:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_TYPE, new TypeReference<List<Version>>() {});
+                break;
+
+            case GET_RESOURCES:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, RestResources.class);
+                break;
+
+            case GET_GLOBAL_OBJECTS:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, GlobalObjects.class);
+                break;
+
+            case GET_BASIC_INFO:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, SObjectBasicInfo.class);
+                break;
+
+            case GET_DESCRIPTION:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, SObjectDescription.class);
+                break;
+
+            case CREATE_SOBJECT:
+                // handle known response type
+                exchange.setProperty(RESPONSE_CLASS, CreateSObjectResult.class);
+                break;
+
+            case UPSERT_SOBJECT:
+                // handle known response type
+                exchange.setProperty(RESPONSE_CLASS, CreateSObjectResult.class);
+                break;
+
+            case SEARCH:
+                // handle known response type
+                exchange.setProperty(RESPONSE_TYPE, new TypeReference<List<SearchResult>>() {});
+                break;
+
+        }
+    }
+
+    @Override
+    protected InputStream getRequestStream(Exchange exchange) throws SalesforceException {
+        try {
+            InputStream request;
+            Message in = exchange.getIn();
+            request = in.getBody(InputStream.class);
+            if (request == null) {
+                AbstractSObjectBase sObject = in.getBody(AbstractSObjectBase.class);
+                if (sObject != null) {
+                    // marshall the SObject
+                    ByteArrayOutputStream out = new ByteArrayOutputStream();
+                    objectMapper.writeValue(out, sObject);
+                    request = new ByteArrayInputStream(out.toByteArray());
+                } else {
+                    // if all else fails, get body as String
+                    final String body = in.getBody(String.class);
+                    if (null == body) {
+                        String msg = "Unsupported request message body " +
+                            (in.getBody() == null ? null : in.getBody().getClass());
+                        throw new SalesforceException(msg, null);
+                    } else {
+                        request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8_CHARSET));
+                    }
+                }
+            }
+
+            return request;
+
+        } catch (IOException e) {
+            String msg = "Error marshaling request: " + e.getMessage();
+            throw new SalesforceException(msg, e);
+        }
+    }
+
+    @Override
+    protected void processResponse(Exchange exchange, InputStream responseEntity, SalesforceException ex, AsyncCallback callback) {
+
+        // process JSON response for TypeReference
+        try {
+            // do we need to un-marshal a response
+            if (responseEntity != null) {
+                Object response = null;
+                Class<?> responseClass = exchange.getProperty(RESPONSE_CLASS, Class.class);
+                if (responseClass != null) {
+                    response = objectMapper.readValue(responseEntity, responseClass);
+                } else {
+                    TypeReference<?> responseType = exchange.getProperty(RESPONSE_TYPE, TypeReference.class);
+                    if (responseType != null) {
+                        response = objectMapper.readValue(responseEntity, responseType);
+                    } else {
+                        // return the response as a stream, for getBlobField
+                        response = responseEntity;
+                    }
+                }
+                exchange.getOut().setBody(response);
+            } else {
+                exchange.setException(ex);
+            }
+            // copy headers and attachments
+            exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
+            exchange.getOut().getAttachments().putAll(exchange.getIn().getAttachments());
+        } catch (IOException e) {
+            String msg = "Error parsing JSON response: " + e.getMessage();
+            exchange.setException(new SalesforceException(msg, e));
+        } finally {
+            // cleanup temporary exchange headers
+            exchange.removeProperty(RESPONSE_CLASS);
+            exchange.removeProperty(RESPONSE_TYPE);
+
+            // consume response entity
+            try {
+                if (responseEntity != null) {
+                    responseEntity.close();
+                }
+            } catch (IOException ignored) {
+            }
+
+            // notify callback that exchange is done
+            callback.done(false);
+        }
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/SalesforceProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/SalesforceProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/SalesforceProcessor.java
new file mode 100644
index 0000000..36f77f6
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/SalesforceProcessor.java
@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Service;
+
+public interface SalesforceProcessor extends Service {
+
+    boolean process(Exchange exchange, AsyncCallback callback);
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/72a1767e/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/XmlRestProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/XmlRestProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/XmlRestProcessor.java
new file mode 100644
index 0000000..75449b6
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/XmlRestProcessor.java
@@ -0,0 +1,240 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.salesforce.internal.processor;
+
+import com.thoughtworks.xstream.XStream;
+import com.thoughtworks.xstream.XStreamException;
+import com.thoughtworks.xstream.io.HierarchicalStreamWriter;
+import com.thoughtworks.xstream.io.naming.NoNameCoder;
+import com.thoughtworks.xstream.io.xml.CompactWriter;
+import com.thoughtworks.xstream.io.xml.XppDriver;
+import com.thoughtworks.xstream.mapper.CachingMapper;
+import com.thoughtworks.xstream.mapper.CannotResolveClassException;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.eclipse.jetty.util.StringUtil;
+import org.apache.camel.component.salesforce.SalesforceEndpoint;
+import org.apache.camel.component.salesforce.api.JodaTimeConverter;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.api.dto.*;
+
+import java.io.*;
+
+import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.SOBJECT_NAME;
+
+public class XmlRestProcessor extends AbstractRestProcessor {
+
+    // although XStream is generally thread safe, because of the way we use aliases
+    // for GET_BASIC_INFO and GET_DESCRIPTION, we need to use a ThreadLocal
+    // not very efficient when both JSON and XML are used together with a single Thread pool
+    // but this will do for now
+    private static ThreadLocal<XStream> xStream =
+        new ThreadLocal<XStream>() {
+            @Override
+            protected XStream initialValue() {
+                // use NoNameCoder to avoid escaping __ in custom field names
+                // and CompactWriter to avoid pretty printing
+                XStream result = new XStream(new XppDriver(new NoNameCoder()) {
+                    @Override
+                    public HierarchicalStreamWriter createWriter(Writer out) {
+                        return new CompactWriter(out, getNameCoder());
+                    }
+
+                });
+                result.registerConverter(new JodaTimeConverter());
+                return result;
+            }
+        };
+
+    private static final String RESPONSE_ALIAS = XmlRestProcessor.class.getName() + ".responseAlias";
+
+    public XmlRestProcessor(SalesforceEndpoint endpoint) throws SalesforceException {
+        super(endpoint);
+
+    }
+
+    @Override
+    protected void processRequest(Exchange exchange) throws SalesforceException {
+
+        switch (operationName) {
+            case GET_VERSIONS:
+                exchange.setProperty(RESPONSE_CLASS, Versions.class);
+                break;
+
+            case GET_RESOURCES:
+                exchange.setProperty(RESPONSE_CLASS, RestResources.class);
+                break;
+
+            case GET_GLOBAL_OBJECTS:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, GlobalObjects.class);
+                break;
+
+            case GET_BASIC_INFO:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, SObjectBasicInfo.class);
+
+                // need to add alias for Salesforce XML that uses SObject name as root element
+                exchange.setProperty(RESPONSE_ALIAS,
+                    getParameter(SOBJECT_NAME, exchange, USE_BODY, NOT_OPTIONAL));
+                break;
+
+            case GET_DESCRIPTION:
+                // handle in built response types
+                exchange.setProperty(RESPONSE_CLASS, SObjectDescription.class);
+
+                // need to add alias for Salesforce XML that uses SObject name as root element
+                exchange.setProperty(RESPONSE_ALIAS,
+                    getParameter(SOBJECT_NAME, exchange, USE_BODY, NOT_OPTIONAL));
+                break;
+
+            case GET_SOBJECT:
+                // need to add alias for Salesforce XML that uses SObject name as root element
+                exchange.setProperty(RESPONSE_ALIAS,
+                    getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL));
+                break;
+
+            case CREATE_SOBJECT:
+                // handle known response type
+                exchange.setProperty(RESPONSE_CLASS, CreateSObjectResult.class);
+                break;
+
+            case GET_SOBJECT_WITH_ID:
+                // need to add alias for Salesforce XML that uses SObject name as root element
+                exchange.setProperty(RESPONSE_ALIAS,
+                    getParameter(SOBJECT_NAME, exchange, IGNORE_BODY, NOT_OPTIONAL));
+                break;
+
+            case UPSERT_SOBJECT:
+                // handle known response type
+                exchange.setProperty(RESPONSE_CLASS, CreateSObjectResult.class);
+                break;
+
+            case QUERY:
+            case QUERY_MORE:
+                // need to add alias for Salesforce XML that uses SObject name as root element
+                exchange.setProperty(RESPONSE_ALIAS,
+                    "QueryResult");
+                break;
+
+            case SEARCH:
+                // handle known response type
+                exchange.setProperty(RESPONSE_CLASS, SearchResults.class);
+                break;
+
+        }
+
+    }
+
+    protected InputStream getRequestStream(Exchange exchange) throws SalesforceException {
+        final XStream localXStream = xStream.get();
+        try {
+            // get request stream from In message
+            Message in = exchange.getIn();
+            InputStream request = in.getBody(InputStream.class);
+            if (request == null) {
+                AbstractSObjectBase sObject = in.getBody(AbstractSObjectBase.class);
+                if (sObject != null) {
+                    // marshall the SObject
+                    // first process annotations on the class, for things like alias, etc.
+                    localXStream.processAnnotations(sObject.getClass());
+                    ByteArrayOutputStream out = new ByteArrayOutputStream();
+                    // make sure we write the XML with the right encoding
+                    localXStream.toXML(sObject, new OutputStreamWriter(out, StringUtil.__UTF8_CHARSET));
+                    request = new ByteArrayInputStream(out.toByteArray());
+                } else {
+                    // if all else fails, get body as String
+                    final String body = in.getBody(String.class);
+                    if (null == body) {
+                        String msg = "Unsupported request message body " +
+                            (in.getBody() == null ? null : in.getBody().getClass());
+                        throw new SalesforceException(msg, null);
+                    } else {
+                        request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8_CHARSET));
+                    }
+                }
+            }
+            return request;
+        } catch (XStreamException e) {
+            String msg = "Error marshaling request: " + e.getMessage();
+            throw new SalesforceException(msg, e);
+        }
+    }
+
+    @Override
+    protected void processResponse(Exchange exchange, InputStream responseEntity,
+                                   SalesforceException exception, AsyncCallback callback) {
+        final XStream localXStream = xStream.get();
+        try {
+            // do we need to un-marshal a response
+            if (responseEntity != null) {
+                final Class<?> responseClass = exchange.getProperty(RESPONSE_CLASS, Class.class);
+                Object response;
+                if (responseClass != null) {
+                    // its ok to call this multiple times, as xstream ignores duplicate calls
+                    localXStream.processAnnotations(responseClass);
+                    final String responseAlias = exchange.getProperty(RESPONSE_ALIAS, String.class);
+                    if (responseAlias != null) {
+                        // extremely dirty, need to flush entire cache if its holding on to an old alias!!!
+                        final CachingMapper mapper = (CachingMapper) localXStream.getMapper();
+                        try {
+                            if (mapper.realClass(responseAlias) != responseClass) {
+                                mapper.flushCache();
+                            }
+                        } catch (CannotResolveClassException ignore) {
+                        }
+                        localXStream.alias(responseAlias, responseClass);
+                    }
+                    response = responseClass.newInstance();
+                    localXStream.fromXML(responseEntity, response);
+                } else {
+                    // return the response as a stream, for getBlobField
+                    response = responseEntity;
+                }
+                exchange.getOut().setBody(response);
+            } else {
+                exchange.setException(exception);
+            }
+            // copy headers and attachments
+            exchange.getOut().getHeaders().putAll(exchange.getIn().getHeaders());
+            exchange.getOut().getAttachments().putAll(exchange.getIn().getAttachments());
+        } catch (XStreamException e) {
+            String msg = "Error parsing XML response: " + e.getMessage();
+            exchange.setException(new SalesforceException(msg, e));
+        } catch (Exception e) {
+            String msg = "Error creating XML response: " + e.getMessage();
+            exchange.setException(new SalesforceException(msg, e));
+        } finally {
+            // cleanup temporary exchange headers
+            exchange.removeProperty(RESPONSE_CLASS);
+            exchange.removeProperty(RESPONSE_ALIAS);
+
+            // consume response entity
+            if (responseEntity != null) {
+                try {
+                    responseEntity.close();
+                } catch (IOException ignored) {
+                }
+            }
+
+            // notify callback that exchange is done
+            callback.done(false);
+        }
+    }
+
+}


Mime
View raw message