You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by dh...@apache.org on 2016/04/29 07:51:25 UTC
[2/4] camel git commit: CAMEL-9925: Updated Salesforce component to
use Jetty9 and cometd3
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRestClient.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRestClient.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRestClient.java
index 2dade18..9eb6e0c 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRestClient.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRestClient.java
@@ -26,6 +26,7 @@ import java.util.Map;
import com.thoughtworks.xstream.XStream;
+import org.apache.camel.component.salesforce.SalesforceHttpClient;
import org.apache.camel.component.salesforce.api.SalesforceException;
import org.apache.camel.component.salesforce.api.SalesforceMultipleChoicesException;
import org.apache.camel.component.salesforce.api.dto.RestError;
@@ -37,11 +38,11 @@ import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.URISupport;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;
-import org.eclipse.jetty.client.ContentExchange;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.HttpExchange;
-import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.http.HttpMethods;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.util.InputStreamContentProvider;
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.StringUtil;
@@ -56,7 +57,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
private ObjectMapper objectMapper;
private XStream xStream;
- public DefaultRestClient(HttpClient httpClient, String version, PayloadFormat format, SalesforceSession session)
+ public DefaultRestClient(SalesforceHttpClient httpClient, String version, PayloadFormat format, SalesforceSession session)
throws SalesforceException {
super(version, session, httpClient);
@@ -72,36 +73,32 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
}
@Override
- protected void doHttpRequest(ContentExchange request, ClientResponseCallback callback) {
+ protected void doHttpRequest(Request request, ClientResponseCallback callback) {
// set standard headers for all requests
final String contentType = PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8;
- request.setRequestHeader(HttpHeaders.ACCEPT, contentType);
- request.setRequestHeader(HttpHeaders.ACCEPT_CHARSET, StringUtil.__UTF8);
+ request.header(HttpHeader.ACCEPT, contentType);
+ request.header(HttpHeader.ACCEPT_CHARSET, StringUtil.__UTF8);
// request content type and charset is set by the request entity
super.doHttpRequest(request, callback);
}
@Override
- protected SalesforceException createRestException(ContentExchange httpExchange, String reason) {
+ protected SalesforceException createRestException(Response response, InputStream responseContent) {
// get status code and reason phrase
- final int statusCode = httpExchange.getResponseStatus();
+ final int statusCode = response.getStatus();
+ String reason = response.getReason();
if (reason == null || reason.isEmpty()) {
reason = HttpStatus.getMessage(statusCode);
}
// try parsing response according to format
- String responseContent = null;
try {
- responseContent = httpExchange.getResponseContent();
- if (responseContent != null && !responseContent.isEmpty()) {
+ if (responseContent != null && responseContent.available() > 0) {
final List<String> choices;
// return list of choices as error message for 300
if (statusCode == HttpStatus.MULTIPLE_CHOICES_300) {
if (PayloadFormat.JSON.equals(format)) {
- choices = objectMapper.readValue(
- responseContent, new TypeReference<List<String>>() {
- }
- );
+ choices = objectMapper.readValue(responseContent, new TypeReference<List<String>>() {});
} else {
RestChoices restChoices = new RestChoices();
xStream.fromXML(responseContent, restChoices);
@@ -142,7 +139,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getVersions(final ResponseCallback callback) {
- ContentExchange get = getContentExchange(HttpMethods.GET, servicesDataUrl());
+ Request get = getRequest(HttpMethod.GET, servicesDataUrl());
// does not require authorization token
doHttpRequest(get, new DelegatingClientCallback(callback));
@@ -150,7 +147,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getResources(ResponseCallback callback) {
- ContentExchange get = getContentExchange(HttpMethods.GET, versionUrl());
+ Request get = getRequest(HttpMethod.GET, versionUrl());
// requires authorization token
setAccessToken(get);
@@ -159,7 +156,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getGlobalObjects(ResponseCallback callback) {
- ContentExchange get = getContentExchange(HttpMethods.GET, sobjectsUrl(""));
+ Request get = getRequest(HttpMethod.GET, sobjectsUrl(""));
// requires authorization token
setAccessToken(get);
@@ -169,7 +166,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getBasicInfo(String sObjectName,
ResponseCallback callback) {
- ContentExchange get = getContentExchange(HttpMethods.GET, sobjectsUrl(sObjectName + "/"));
+ Request get = getRequest(HttpMethod.GET, sobjectsUrl(sObjectName + "/"));
// requires authorization token
setAccessToken(get);
@@ -179,7 +176,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getDescription(String sObjectName,
ResponseCallback callback) {
- ContentExchange get = getContentExchange(HttpMethods.GET, sobjectsUrl(sObjectName + "/describe/"));
+ Request get = getRequest(HttpMethod.GET, sobjectsUrl(sObjectName + "/describe/"));
// requires authorization token
setAccessToken(get);
@@ -202,7 +199,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
}
params = fieldsValue.toString();
}
- ContentExchange get = getContentExchange(HttpMethods.GET, sobjectsUrl(sObjectName + "/" + id + params));
+ Request get = getRequest(HttpMethod.GET, sobjectsUrl(sObjectName + "/" + id + params));
// requires authorization token
setAccessToken(get);
@@ -213,14 +210,14 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
public void createSObject(String sObjectName, InputStream sObject,
ResponseCallback callback) {
// post the sObject
- final ContentExchange post = getContentExchange(HttpMethods.POST, sobjectsUrl(sObjectName));
+ final Request post = getRequest(HttpMethod.POST, sobjectsUrl(sObjectName));
// authorization
setAccessToken(post);
// input stream as entity content
- post.setRequestContentSource(sObject);
- post.setRequestContentType(PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
+ post.content(new InputStreamContentProvider(sObject));
+ post.header(HttpHeader.CONTENT_TYPE, PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
doHttpRequest(post, new DelegatingClientCallback(callback));
}
@@ -228,13 +225,13 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void updateSObject(String sObjectName, String id, InputStream sObject,
ResponseCallback callback) {
- final ContentExchange patch = getContentExchange("PATCH", sobjectsUrl(sObjectName + "/" + id));
+ final Request patch = getRequest("PATCH", sobjectsUrl(sObjectName + "/" + id));
// requires authorization token
setAccessToken(patch);
// input stream as entity content
- patch.setRequestContentSource(sObject);
- patch.setRequestContentType(PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
+ patch.content(new InputStreamContentProvider(sObject));
+ patch.header(HttpHeader.CONTENT_TYPE, PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
doHttpRequest(patch, new DelegatingClientCallback(callback));
}
@@ -242,7 +239,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void deleteSObject(String sObjectName, String id,
ResponseCallback callback) {
- final ContentExchange delete = getContentExchange(HttpMethods.DELETE, sobjectsUrl(sObjectName + "/" + id));
+ final Request delete = getRequest(HttpMethod.DELETE, sobjectsUrl(sObjectName + "/" + id));
// requires authorization token
setAccessToken(delete);
@@ -253,7 +250,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getSObjectWithId(String sObjectName, String fieldName, String fieldValue,
ResponseCallback callback) {
- final ContentExchange get = getContentExchange(HttpMethods.GET,
+ final Request get = getRequest(HttpMethod.GET,
sobjectsExternalIdUrl(sObjectName, fieldName, fieldValue));
// requires authorization token
@@ -265,16 +262,16 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void upsertSObject(String sObjectName, String fieldName, String fieldValue, InputStream sObject,
ResponseCallback callback) {
- final ContentExchange patch = getContentExchange("PATCH",
+ final Request patch = getRequest("PATCH",
sobjectsExternalIdUrl(sObjectName, fieldName, fieldValue));
// requires authorization token
setAccessToken(patch);
// input stream as entity content
- patch.setRequestContentSource(sObject);
+ patch.content(new InputStreamContentProvider(sObject));
// TODO will the encoding always be UTF-8??
- patch.setRequestContentType(PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
+ patch.header(HttpHeader.CONTENT_TYPE, PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
doHttpRequest(patch, new DelegatingClientCallback(callback));
}
@@ -282,7 +279,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void deleteSObjectWithId(String sObjectName, String fieldName, String fieldValue,
ResponseCallback callback) {
- final ContentExchange delete = getContentExchange(HttpMethods.DELETE,
+ final Request delete = getRequest(HttpMethod.DELETE,
sobjectsExternalIdUrl(sObjectName, fieldName, fieldValue));
// requires authorization token
@@ -293,10 +290,10 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void getBlobField(String sObjectName, String id, String blobFieldName, ResponseCallback callback) {
- final ContentExchange get = getContentExchange(HttpMethods.GET,
+ final Request get = getRequest(HttpMethod.GET,
sobjectsUrl(sObjectName + "/" + id + "/" + blobFieldName));
// TODO this doesn't seem to be required, the response is always the content binary stream
- //get.setRequestHeader(HttpHeaders.ACCEPT_ENCODING, "base64");
+ //get.header(HttpHeader.ACCEPT_ENCODING, "base64");
// requires authorization token
setAccessToken(get);
@@ -309,7 +306,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
try {
String encodedQuery = urlEncode(soqlQuery);
- final ContentExchange get = getContentExchange(HttpMethods.GET, versionUrl() + "query/?q=" + encodedQuery);
+ final Request get = getRequest(HttpMethod.GET, versionUrl() + "query/?q=" + encodedQuery);
// requires authorization token
setAccessToken(get);
@@ -324,7 +321,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void queryMore(String nextRecordsUrl, ResponseCallback callback) {
- final ContentExchange get = getContentExchange(HttpMethods.GET, instanceUrl + nextRecordsUrl);
+ final Request get = getRequest(HttpMethod.GET, instanceUrl + nextRecordsUrl);
// requires authorization token
setAccessToken(get);
@@ -337,7 +334,7 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
try {
String encodedQuery = urlEncode(soslQuery);
- final ContentExchange get = getContentExchange(HttpMethods.GET, versionUrl() + "search/?q=" + encodedQuery);
+ final Request get = getRequest(HttpMethod.GET, versionUrl() + "search/?q=" + encodedQuery);
// requires authorization token
setAccessToken(get);
@@ -353,21 +350,21 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
@Override
public void apexCall(String httpMethod, String apexUrl,
Map<String, Object> queryParams, InputStream requestDto, ResponseCallback callback) {
- // create APEX call exchange
- final ContentExchange exchange;
+ // create APEX call request
+ final Request request;
try {
- exchange = getContentExchange(httpMethod, apexCallUrl(apexUrl, queryParams));
+ request = getRequest(httpMethod, apexCallUrl(apexUrl, queryParams));
// set request SObject and content type
if (requestDto != null) {
- exchange.setRequestContentSource(requestDto);
- exchange.setRequestContentType(
+ request.content(new InputStreamContentProvider(requestDto));
+ request.header(HttpHeader.CONTENT_TYPE,
PayloadFormat.JSON.equals(format) ? APPLICATION_JSON_UTF8 : APPLICATION_XML_UTF8);
}
// requires authorization token
- setAccessToken(exchange);
+ setAccessToken(request);
- doHttpRequest(exchange, new DelegatingClientCallback(callback));
+ doHttpRequest(request, new DelegatingClientCallback(callback));
} catch (UnsupportedEncodingException e) {
String msg = "Unexpected error: " + e.getMessage();
callback.onResponse(null, new SalesforceException(msg, e));
@@ -414,12 +411,13 @@ public class DefaultRestClient extends AbstractClientBase implements RestClient
}
}
- protected void setAccessToken(HttpExchange httpExchange) {
- httpExchange.setRequestHeader(TOKEN_HEADER, TOKEN_PREFIX + accessToken);
+ protected void setAccessToken(Request request) {
+ // replace old token
+ request.getHeaders().put(TOKEN_HEADER, TOKEN_PREFIX + accessToken);
}
private String urlEncode(String query) throws UnsupportedEncodingException {
- String encodedQuery = URLEncoder.encode(query, StringUtil.__UTF8_CHARSET.toString());
+ String encodedQuery = URLEncoder.encode(query, StringUtil.__UTF8);
// URLEncoder likes to use '+' for spaces
encodedQuery = encodedQuery.replace("+", "%20");
return encodedQuery;
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceExchange.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceExchange.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceExchange.java
deleted file mode 100644
index b17c5e1..0000000
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceExchange.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * 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.client;
-
-import org.eclipse.jetty.client.ContentExchange;
-
-/**
- * Wraps a Salesforce Http Exchange
- */
-public class SalesforceExchange extends ContentExchange {
-
- private AbstractClientBase client;
-
- public AbstractClientBase getClient() {
- return client;
- }
-
- public void setClient(AbstractClientBase client) {
- this.client = client;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceHttpRequest.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceHttpRequest.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceHttpRequest.java
new file mode 100644
index 0000000..743ec32
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceHttpRequest.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.client;
+
+import java.net.URI;
+
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpConversation;
+import org.eclipse.jetty.client.HttpRequest;
+
+/**
+ * Salesforce HTTP Request, exposes {@link HttpConversation} field.
+ */
+public class SalesforceHttpRequest extends HttpRequest {
+
+ public SalesforceHttpRequest(HttpClient client, HttpConversation conversation, URI uri) {
+ super(client, conversation, uri);
+ }
+
+ @Override
+ protected HttpConversation getConversation() {
+ return super.getConversation();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java
new file mode 100644
index 0000000..6a02b92
--- /dev/null
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java
@@ -0,0 +1,262 @@
+/**
+ * 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.client;
+
+import java.io.InputStream;
+
+import org.apache.camel.component.salesforce.SalesforceHttpClient;
+import org.apache.camel.component.salesforce.api.SalesforceException;
+import org.apache.camel.component.salesforce.internal.SalesforceSession;
+import org.eclipse.jetty.client.HttpContentResponse;
+import org.eclipse.jetty.client.HttpConversation;
+import org.eclipse.jetty.client.ProtocolHandler;
+import org.eclipse.jetty.client.ResponseNotifier;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.client.api.Response;
+import org.eclipse.jetty.client.api.Result;
+import org.eclipse.jetty.client.util.BufferingResponseListener;
+import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.http.HttpStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SalesforceSecurityHandler implements ProtocolHandler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SalesforceSecurityHandler.class);
+
+ private static final String AUTHENTICATION_RETRIES_ATTRIBUTE = SalesforceSecurityHandler.class.getName().concat(".retries");
+ static final String CLIENT_ATTRIBUTE = SalesforceSecurityHandler.class.getName().concat("camel-salesforce-client");
+ static final String AUTHENTICATION_REQUEST_ATTRIBUTE = SalesforceSecurityHandler.class.getName().concat(".request");
+
+ private final SalesforceHttpClient httpClient;
+ private final SalesforceSession session;
+ private final int maxAuthenticationRetries;
+ private final int maxContentLength;
+ private final ResponseNotifier notifier;
+
+ public SalesforceSecurityHandler(SalesforceHttpClient httpClient) {
+
+ this.httpClient = httpClient;
+ this.session = httpClient.getSession();
+
+ this.maxAuthenticationRetries = httpClient.getMaxRetries();
+ this.maxContentLength = httpClient.getMaxContentLength();
+ this.notifier = new ResponseNotifier();
+ }
+
+ @Override
+ public boolean accept(Request request, Response response) {
+
+ HttpConversation conversation = ((SalesforceHttpRequest) request).getConversation();
+ Integer retries = (Integer) conversation.getAttribute(AUTHENTICATION_RETRIES_ATTRIBUTE);
+
+ // is this an authentication response for a previously handled conversation?
+ if (conversation.getAttribute(AUTHENTICATION_REQUEST_ATTRIBUTE) != null
+ && (retries == null || retries <= maxAuthenticationRetries)) {
+ return true;
+ }
+
+ final int status = response.getStatus();
+ // handle UNAUTHORIZED and BAD_REQUEST for Bulk API,
+ // the actual InvalidSessionId Bulk API error is checked and handled in the listener
+ // also check retries haven't exceeded maxAuthenticationRetries
+ return (status == HttpStatus.UNAUTHORIZED_401 || status == HttpStatus.BAD_REQUEST_400)
+ && (retries == null || retries <= maxAuthenticationRetries);
+ }
+
+ @Override
+ public Response.Listener getResponseListener() {
+ return new SecurityListener(maxContentLength);
+ }
+
+ private class SecurityListener extends BufferingResponseListener {
+
+ public SecurityListener(int maxLength) {
+ super(maxLength);
+ }
+
+ @Override
+ public void onComplete(Result result) {
+
+ SalesforceHttpRequest request = (SalesforceHttpRequest)result.getRequest();
+ ContentResponse response = new HttpContentResponse(result.getResponse(), getContent(), getMediaType(), getEncoding());
+
+ // get number of retries
+ HttpConversation conversation = request.getConversation();
+ Integer retries = (Integer) conversation.getAttribute(AUTHENTICATION_RETRIES_ATTRIBUTE);
+ if (retries == null) {
+ retries = 0;
+ }
+
+ // get AbstractClientBase if request originated from one, for updating token and setting auth header
+ final AbstractClientBase client = (AbstractClientBase) conversation.getAttribute(CLIENT_ATTRIBUTE);
+
+ // exception response
+ if (result.isFailed()) {
+ Throwable failure = result.getFailure();
+ retryOnFailure(request, conversation, retries, client, failure);
+ return;
+ }
+
+ // response to a re-login request
+ SalesforceHttpRequest origRequest = (SalesforceHttpRequest) conversation.getAttribute(AUTHENTICATION_REQUEST_ATTRIBUTE);
+ if (origRequest != null) {
+
+ // parse response
+ try {
+ session.parseLoginResponse(response, response.getContentAsString());
+ } catch (SalesforceException e) {
+ // retry login request on error if we have login attempts left
+ if (retries < maxAuthenticationRetries) {
+ retryOnFailure(request, conversation, retries, client, e);
+ } else {
+ forwardFailureComplete(origRequest, null, response, e);
+ }
+ return;
+ }
+
+ // retry original request on success
+ conversation.removeAttribute(AUTHENTICATION_REQUEST_ATTRIBUTE);
+ retryRequest(origRequest, client, retries, conversation, true);
+ return;
+ }
+
+ // response to an original request
+ final int status = response.getStatus();
+ final String reason = response.getReason();
+
+ // check if login retries left
+ if (retries >= maxAuthenticationRetries) {
+ // forward current response
+ forwardSuccessComplete(request, response);
+ return;
+ }
+
+ // request failed authentication?
+ if (status == HttpStatus.UNAUTHORIZED_401) {
+
+ // REST token expiry
+ LOG.warn("Retrying on Salesforce authentication error [{}]: [{}]", status, reason);
+
+ // remember original request and send a relogin request in current conversation
+ retryLogin(request, retries);
+
+ } else if (status < HttpStatus.OK_200 || status >= HttpStatus.MULTIPLE_CHOICES_300) {
+
+ // HTTP failure status
+ // get detailed cause, if request comes from an AbstractClientBase
+ final InputStream inputStream = getContent().length == 0 ? null : getContentAsInputStream();
+ final SalesforceException cause = client != null ?
+ client.createRestException(response, inputStream) : null;
+
+ if (status == HttpStatus.BAD_REQUEST_400 && cause != null && isInvalidSessionError(cause)) {
+
+ // retry Bulk API call
+ LOG.warn("Retrying on Bulk API Salesforce authentication error [{}]: [{}]", status, reason);
+ retryLogin(request, retries);
+
+ } else {
+
+ // forward Salesforce HTTP failure!
+ forwardSuccessComplete(request, response);
+ }
+ }
+ }
+
+ protected void retryOnFailure(SalesforceHttpRequest request, HttpConversation conversation, Integer retries, AbstractClientBase client, Throwable failure) {
+ LOG.warn("Retrying on Salesforce authentication failure " + failure.getMessage(), failure);
+
+ // retry request
+ retryRequest(request, client, retries, conversation, true);
+ }
+
+ private boolean isInvalidSessionError(SalesforceException e) {
+ return e.getErrors() != null && e.getErrors().size() == 1
+ && "InvalidSessionId".equals(e.getErrors().get(0).getErrorCode());
+ }
+
+ private void retryLogin(SalesforceHttpRequest request, Integer retries) {
+
+ final HttpConversation conversation = request.getConversation();
+ // remember the original request to resend
+ conversation.setAttribute(AUTHENTICATION_REQUEST_ATTRIBUTE, request);
+
+ retryRequest((SalesforceHttpRequest)session.getLoginRequest(conversation), null, retries, conversation, false);
+ }
+
+ private void retryRequest(SalesforceHttpRequest request, AbstractClientBase client, Integer retries, HttpConversation conversation,
+ boolean copy) {
+ // copy the request to resend
+ // TODO handle a change in Salesforce instanceUrl, right now we retry with the same destination
+ final Request newRequest;
+ if (copy) {
+ newRequest = httpClient.copyRequest(request, request.getURI());
+ newRequest.method(request.getMethod());
+ } else {
+ newRequest = request;
+ }
+
+ conversation.setAttribute(AUTHENTICATION_RETRIES_ATTRIBUTE, ++retries);
+
+ LOG.debug("Retry attempt {} on authentication error for {}", retries, request);
+
+ // update currentToken
+ String currentToken = session.getAccessToken();
+ if (client != null) {
+ // update client cache for this and future requests
+ client.setAccessToken(currentToken);
+ client.setInstanceUrl(session.getInstanceUrl());
+ client.setAccessToken(newRequest);
+ } else {
+ // plain request not made by an AbstractClientBase
+ newRequest.header(HttpHeader.AUTHORIZATION, "OAuth " + currentToken);
+ }
+
+ // send new async request with a new delegate
+ conversation.updateResponseListeners(null);
+ newRequest.onRequestBegin(getRequestAbortListener(request));
+ newRequest.send(null);
+ }
+
+ private Request.BeginListener getRequestAbortListener(final SalesforceHttpRequest request) {
+ return new Request.BeginListener() {
+ @Override
+ public void onBegin(Request redirect) {
+ Throwable cause = request.getAbortCause();
+ if (cause != null) {
+ redirect.abort(cause);
+ }
+ }
+ };
+ }
+
+ private void forwardSuccessComplete(SalesforceHttpRequest request, Response response) {
+ HttpConversation conversation = request.getConversation();
+ conversation.updateResponseListeners(null);
+ notifier.forwardSuccessComplete(conversation.getResponseListeners(), request, response);
+ }
+
+ private void forwardFailureComplete(SalesforceHttpRequest request, Throwable requestFailure,
+ Response response, Throwable responseFailure) {
+ HttpConversation conversation = request.getConversation();
+ conversation.updateResponseListeners(null);
+ notifier.forwardFailureComplete(conversation.getResponseListeners(), request, requestFailure,
+ response, responseFailure);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityListener.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityListener.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityListener.java
deleted file mode 100644
index 09fde7a..0000000
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityListener.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/**
- * 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.client;
-
-import java.io.IOException;
-
-import org.apache.camel.component.salesforce.api.SalesforceException;
-import org.apache.camel.component.salesforce.internal.SalesforceSession;
-import org.eclipse.jetty.client.HttpDestination;
-import org.eclipse.jetty.client.HttpEventListenerWrapper;
-import org.eclipse.jetty.client.HttpExchange;
-import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.http.HttpStatus;
-import org.eclipse.jetty.io.Buffer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SalesforceSecurityListener extends HttpEventListenerWrapper {
-
- private static final Logger LOG = LoggerFactory.getLogger(SalesforceSecurityListener.class);
-
- private final HttpDestination destination;
- private final HttpExchange exchange;
- private final SalesforceSession session;
-
- private String currentToken;
- private int retries;
- private boolean retrying;
- private boolean requestComplete;
- private boolean responseComplete;
- private SalesforceException exceptionResponse;
-
- public SalesforceSecurityListener(HttpDestination destination, HttpExchange exchange,
- SalesforceSession session, String accessToken) {
- super(exchange.getEventListener(), true);
- this.destination = destination;
- this.exchange = exchange;
- this.session = session;
- this.currentToken = accessToken;
- }
-
- @Override
- public void onResponseStatus(Buffer version, int status, Buffer reason) throws IOException {
- if (status == HttpStatus.UNAUTHORIZED_401 && retries < destination.getHttpClient().maxRetries()) {
- LOG.warn("Retrying on Salesforce authentication error [{}]: [{}]", status, reason);
- setDelegatingRequests(false);
- setDelegatingResponses(false);
-
- retrying = true;
- }
- super.onResponseStatus(version, status, reason);
- }
-
- @Override
- public void onRequestComplete() throws IOException {
- requestComplete = true;
- if (checkExchangeComplete()) {
- super.onRequestComplete();
- }
- }
-
- @Override
- public void onResponseComplete() throws IOException {
- responseComplete = true;
-
- exceptionResponse = createExceptionResponse();
- if (!retrying && exceptionResponse != null && isInvalidSessionError(exceptionResponse)) {
- if (LOG.isWarnEnabled()) {
- LOG.warn("Retrying on Salesforce InvalidSessionId error: {}",
- getRootSalesforceException(exceptionResponse).getMessage());
- }
- retrying = true;
- }
-
- if (checkExchangeComplete()) {
- super.onResponseComplete();
- }
- }
-
- private boolean isInvalidSessionError(SalesforceException e) {
- e = getRootSalesforceException(e);
- return e.getErrors() != null && e.getErrors().size() == 1 && "InvalidSessionId".equals(e.getErrors().get(0).getErrorCode());
- }
-
- private SalesforceException getRootSalesforceException(SalesforceException e) {
- while (e.getCause() instanceof SalesforceException) {
- e = (SalesforceException) e.getCause();
- }
- return e;
- }
-
- protected SalesforceException createExceptionResponse() {
- return null;
- }
-
- private boolean checkExchangeComplete() throws IOException {
- if (retrying && requestComplete && responseComplete) {
- LOG.debug("Authentication Error, retrying: {}", exchange);
-
- requestComplete = false;
- responseComplete = false;
- exceptionResponse = null;
-
- setDelegatingRequests(true);
- setDelegatingResponses(true);
-
- try {
- // get a new token and retry
- currentToken = session.login(currentToken);
-
- if (exchange instanceof SalesforceExchange) {
- final SalesforceExchange salesforceExchange = (SalesforceExchange) exchange;
- final AbstractClientBase client = salesforceExchange.getClient();
-
- // update client cache for this and future requests
- client.setAccessToken(currentToken);
- client.setInstanceUrl(session.getInstanceUrl());
- client.setAccessToken(exchange);
- } else {
- exchange.setRequestHeader(HttpHeaders.AUTHORIZATION,
- "OAuth " + currentToken);
- }
-
- // TODO handle a change in Salesforce instanceUrl, right now we retry with the same destination
- destination.resend(exchange);
-
- // resending, exchange is not done
- return false;
-
- } catch (SalesforceException e) {
- // logging here, since login exception is not propagated!
- LOG.error(e.getMessage(), e);
-
- // the HTTP status and reason is pushed up
- setDelegationResult(false);
- }
- }
-
- return true;
- }
-
- @Override
- public void onRetry() {
- // ignore retries from other interceptors
- if (retrying) {
- retrying = false;
- retries++;
-
- setDelegatingRequests(true);
- setDelegatingResponses(true);
-
- requestComplete = false;
- responseComplete = false;
- exceptionResponse = null;
- }
- super.onRetry();
- }
-
- @Override
- public void onConnectionFailed(Throwable ex) {
- setDelegatingRequests(true);
- setDelegatingResponses(true);
- // delegate connection failures
- super.onConnectionFailed(ex);
- }
-
- @Override
- public void onException(Throwable ex) {
- setDelegatingRequests(true);
- setDelegatingResponses(true);
- // delegate exceptions
- super.onException(ex);
- }
-
- public SalesforceException getExceptionResponse() {
- return exceptionResponse;
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/XStreamUtils.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/XStreamUtils.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/XStreamUtils.java
index 182e411..43c66ad 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/XStreamUtils.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/XStreamUtils.java
@@ -29,7 +29,7 @@ import com.thoughtworks.xstream.security.WildcardTypePermission;
*/
public final class XStreamUtils {
private static final String PERMISSIONS_PROPERTY_KEY = "org.apache.camel.xstream.permissions";
- private static final String PERMISSIONS_PROPERTY_DEFAULT = "-*,java.lang.*,java.util.*";
+ private static final String PERMISSIONS_PROPERTY_DEFAULT = "java.lang.*,java.util.*";
private XStreamUtils() {
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/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
index 352005a..c8ceee7 100644
--- 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
@@ -35,7 +35,6 @@ 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 org.apache.camel.util.ServiceHelper;
-import org.eclipse.jetty.http.HttpMethods;
import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.APEX_METHOD;
import static org.apache.camel.component.salesforce.SalesforceEndpointConfig.APEX_QUERY_PARAM_PREFIX;
@@ -490,7 +489,7 @@ public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor
String apexMethod = getParameter(APEX_METHOD, exchange, IGNORE_BODY, IS_OPTIONAL);
// default to GET
if (apexMethod == null) {
- apexMethod = HttpMethods.GET;
+ apexMethod = "GET";
log.debug("Using HTTP GET method by default for APEX REST call for {}", apexUrl);
}
final Map<String, Object> queryParams = getQueryParams(exchange);
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/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
index 151d24d..76095ba 100644
--- 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
@@ -27,7 +27,7 @@ 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.eclipse.jetty.client.HttpClient;
+import org.apache.camel.component.salesforce.SalesforceHttpClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +45,7 @@ public abstract class AbstractSalesforceProcessor implements SalesforceProcessor
protected final OperationName operationName;
protected final SalesforceSession session;
- protected final HttpClient httpClient;
+ protected final SalesforceHttpClient httpClient;
public AbstractSalesforceProcessor(SalesforceEndpoint endpoint) {
this.endpoint = endpoint;
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AnalyticsApiProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AnalyticsApiProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AnalyticsApiProcessor.java
index 846bd62..cb01912 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AnalyticsApiProcessor.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AnalyticsApiProcessor.java
@@ -49,7 +49,8 @@ public class AnalyticsApiProcessor extends AbstractSalesforceProcessor {
super(endpoint);
this.analyticsClient = new DefaultAnalyticsApiClient(
- (String) endpointConfigMap.get(SalesforceEndpointConfig.API_VERSION), session, httpClient);
+ (String) endpointConfigMap.get(SalesforceEndpointConfig.API_VERSION), session,
+ httpClient);
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/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
index f3c8b4d..16dee3f 100644
--- 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
@@ -127,7 +127,7 @@ public class JsonRestProcessor extends AbstractRestProcessor {
+ (in.getBody() == null ? null : in.getBody().getClass());
throw new SalesforceException(msg, null);
} else {
- request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8_CHARSET));
+ request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8));
}
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/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
index 9e29a5d..a67bef5 100644
--- 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
@@ -21,6 +21,7 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
+import java.io.UnsupportedEncodingException;
import java.io.Writer;
import com.thoughtworks.xstream.XStream;
@@ -170,7 +171,7 @@ public class XmlRestProcessor extends AbstractRestProcessor {
localXStream.processAnnotations(dto.getClass());
ByteArrayOutputStream out = new ByteArrayOutputStream();
// make sure we write the XML with the right encoding
- localXStream.toXML(dto, new OutputStreamWriter(out, StringUtil.__UTF8_CHARSET));
+ localXStream.toXML(dto, new OutputStreamWriter(out, StringUtil.__UTF8));
request = new ByteArrayInputStream(out.toByteArray());
} else {
// if all else fails, get body as String
@@ -180,7 +181,7 @@ public class XmlRestProcessor extends AbstractRestProcessor {
+ (in.getBody() == null ? null : in.getBody().getClass());
throw new SalesforceException(msg, null);
} else {
- request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8_CHARSET));
+ request = new ByteArrayInputStream(body.getBytes(StringUtil.__UTF8));
}
}
}
@@ -188,6 +189,9 @@ public class XmlRestProcessor extends AbstractRestProcessor {
} catch (XStreamException e) {
String msg = "Error marshaling request: " + e.getMessage();
throw new SalesforceException(msg, e);
+ } catch (UnsupportedEncodingException e) {
+ String msg = "Error marshaling request: " + e.getMessage();
+ throw new SalesforceException(msg, e);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
index b0ed0d6..228177c 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
@@ -16,33 +16,29 @@
*/
package org.apache.camel.component.salesforce.internal.streaming;
-import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.concurrent.TimeUnit.SECONDS;
-
import org.apache.camel.CamelException;
import org.apache.camel.component.salesforce.SalesforceComponent;
import org.apache.camel.component.salesforce.SalesforceConsumer;
+import org.apache.camel.component.salesforce.SalesforceHttpClient;
import org.apache.camel.component.salesforce.internal.SalesforceSession;
-import org.apache.camel.component.salesforce.internal.client.SalesforceSecurityListener;
import org.apache.camel.support.ServiceSupport;
import org.cometd.bayeux.Message;
import org.cometd.bayeux.client.ClientSessionChannel;
import org.cometd.client.BayeuxClient;
import org.cometd.client.transport.ClientTransport;
import org.cometd.client.transport.LongPollingTransport;
-import org.eclipse.jetty.client.ContentExchange;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.http.HttpSchemes;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static org.cometd.bayeux.Channel.META_CONNECT;
import static org.cometd.bayeux.Channel.META_HANDSHAKE;
import static org.cometd.bayeux.Channel.META_SUBSCRIBE;
@@ -184,10 +180,10 @@ public class SubscriptionHelper extends ServiceSupport {
private BayeuxClient createClient() throws Exception {
// use default Jetty client from SalesforceComponent, its shared by all consumers
- final HttpClient httpClient = component.getConfig().getHttpClient();
+ final SalesforceHttpClient httpClient = component.getConfig().getHttpClient();
Map<String, Object> options = new HashMap<String, Object>();
- options.put(ClientTransport.TIMEOUT_OPTION, httpClient.getTimeout());
+ options.put(ClientTransport.MAX_NETWORK_DELAY_OPTION, httpClient.getTimeout());
// check login access token
if (session.getAccessToken() == null) {
@@ -197,29 +193,15 @@ public class SubscriptionHelper extends ServiceSupport {
LongPollingTransport transport = new LongPollingTransport(options, httpClient) {
@Override
- protected void customize(ContentExchange exchange) {
- super.customize(exchange);
- // add SalesforceSecurityListener to handle token expiry
- final String accessToken = session.getAccessToken();
- try {
- final boolean isHttps = HttpSchemes.HTTPS.equals(String.valueOf(exchange.getScheme()));
- exchange.setEventListener(new SalesforceSecurityListener(
- httpClient.getDestination(exchange.getAddress(), isHttps),
- exchange, session, accessToken));
- } catch (IOException e) {
- throw new RuntimeException(
- String.format("Error adding SalesforceSecurityListener to exchange %s", e.getMessage()),
- e);
- }
+ protected void customize(Request request) {
+ super.customize(request);
// add current security token obtained from session
- exchange.setRequestHeader(HttpHeaders.AUTHORIZATION,
- "OAuth " + accessToken);
+ request.header(HttpHeader.AUTHORIZATION, "OAuth " + session.getAccessToken());
}
};
BayeuxClient client = new BayeuxClient(getEndpointUrl(), transport);
- client.setDebugEnabled(false);
return client;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractBulkApiTestBase.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractBulkApiTestBase.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractBulkApiTestBase.java
index c48d143..2627535 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractBulkApiTestBase.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractBulkApiTestBase.java
@@ -26,7 +26,7 @@ import org.junit.runner.RunWith;
@RunWith(Theories.class)
public abstract class AbstractBulkApiTestBase extends AbstractSalesforceTestBase {
- protected JobInfo createJob(JobInfo jobInfo) throws InterruptedException {
+ protected JobInfo createJob(JobInfo jobInfo) {
jobInfo = template().requestBody("direct:createJob", jobInfo, JobInfo.class);
assertNotNull("Missing JobId", jobInfo.getId());
return jobInfo;
@@ -94,7 +94,7 @@ public abstract class AbstractBulkApiTestBase extends AbstractSalesforceTestBase
return !(state == BatchStateEnum.QUEUED || state == BatchStateEnum.IN_PROGRESS);
}
- protected BatchInfo getBatchInfo(BatchInfo batchInfo) throws InterruptedException {
+ protected BatchInfo getBatchInfo(BatchInfo batchInfo) {
batchInfo = template().requestBody("direct:getBatch", batchInfo, BatchInfo.class);
assertNotNull("Null batch", batchInfo);
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractSalesforceTestBase.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractSalesforceTestBase.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractSalesforceTestBase.java
index f05bbf9..3dbd36a 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractSalesforceTestBase.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/AbstractSalesforceTestBase.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.salesforce;
+import java.util.HashMap;
+
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.salesforce.dto.generated.Merchandise__c;
import org.apache.camel.test.junit4.CamelTestSupport;
@@ -46,6 +48,13 @@ public abstract class AbstractSalesforceTestBase extends CamelTestSupport {
component.setConfig(config);
component.setLoginConfig(LoginConfigHelper.getLoginConfig());
+ HashMap<String, Object> clientProperties = new HashMap<>();
+ clientProperties.put("timeout", "60000");
+ clientProperties.put("maxRetreis", "3");
+ // 4MB for RestApiIntegrationTest.testGetBlobField()
+ clientProperties.put("maxContentLength", String.valueOf(4 * 1024 * 1024));
+ component.setHttpClientProperties(clientProperties);
+
// set DTO package
component.setPackages(new String[] {
Merchandise__c.class.getPackage().getName()
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiIntegrationTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiIntegrationTest.java
index b8c6dfe..56b0c34 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiIntegrationTest.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/BulkApiIntegrationTest.java
@@ -16,16 +16,17 @@
*/
package org.apache.camel.component.salesforce;
+import java.util.concurrent.TimeUnit;
+
import org.apache.camel.component.salesforce.api.dto.bulk.ContentType;
import org.apache.camel.component.salesforce.api.dto.bulk.JobInfo;
import org.apache.camel.component.salesforce.api.dto.bulk.OperationEnum;
import org.apache.camel.component.salesforce.dto.generated.Merchandise__c;
import org.apache.camel.util.jsse.SSLContextParameters;
-import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.HttpExchange;
-import org.eclipse.jetty.client.RedirectListener;
-import org.eclipse.jetty.http.HttpMethods;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.junit.Test;
@@ -41,16 +42,15 @@ public class BulkApiIntegrationTest extends AbstractBulkApiTestBase {
sslContextFactory.setSslContext(new SSLContextParameters().createSSLContext());
HttpClient httpClient = new HttpClient(sslContextFactory);
httpClient.setConnectTimeout(60000);
- httpClient.setTimeout(60000);
- httpClient.registerListener(RedirectListener.class.getName());
httpClient.start();
- ContentExchange logoutGet = new ContentExchange(true);
- logoutGet.setURL(sf.getLoginConfig().getLoginUrl() + "/services/oauth2/revoke?token=" + accessToken);
- logoutGet.setMethod(HttpMethods.GET);
- httpClient.send(logoutGet);
- assertEquals(HttpExchange.STATUS_COMPLETED, logoutGet.waitForDone());
- assertEquals(HttpStatus.OK_200, logoutGet.getResponseStatus());
+ String uri = sf.getLoginConfig().getLoginUrl() + "/services/oauth2/revoke?token=" + accessToken;
+ Request logoutGet = httpClient.newRequest(uri)
+ .method(HttpMethod.GET)
+ .timeout(1, TimeUnit.MINUTES);
+
+ ContentResponse response = logoutGet.send();
+ assertEquals(HttpStatus.OK_200, response.getStatus());
JobInfo jobInfo = new JobInfo();
jobInfo.setOperation(OperationEnum.INSERT);
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/HttpProxyIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/HttpProxyIntegrationTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/HttpProxyIntegrationTest.java
index ab5e16b..d54b207 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/HttpProxyIntegrationTest.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/HttpProxyIntegrationTest.java
@@ -16,10 +16,8 @@
*/
package org.apache.camel.component.salesforce;
-import java.io.IOException;
import java.util.HashMap;
import java.util.List;
-import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@@ -27,28 +25,32 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.salesforce.api.dto.Version;
import org.apache.camel.component.salesforce.api.dto.Versions;
import org.apache.camel.test.junit4.CamelTestSupport;
-import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.server.Connector;
+import org.eclipse.jetty.proxy.ConnectHandler;
import org.eclipse.jetty.server.Server;
-import org.eclipse.jetty.server.handler.ConnectHandler;
-import org.eclipse.jetty.server.nio.SelectChannelConnector;
+import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.StringUtil;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.eclipse.jetty.http.HttpHeader.PROXY_AUTHENTICATE;
+import static org.eclipse.jetty.http.HttpHeader.PROXY_AUTHORIZATION;
+
/**
* Test HTTP proxy configuration for Salesforce component.
*/
+@Ignore("Bug in Jetty9 causes java.lang.IllegalArgumentException: Invalid protocol login.salesforce.com")
public class HttpProxyIntegrationTest extends AbstractSalesforceTestBase {
private static final Logger LOG = LoggerFactory.getLogger(HttpProxyIntegrationTest.class);
private static final String HTTP_PROXY_HOST = "localhost";
private static final String HTTP_PROXY_USER_NAME = "camel-user";
private static final String HTTP_PROXY_PASSWORD = "camel-user-password";
+ private static final String HTTP_PROXY_REALM = "proxy-realm";
private static Server server;
private static int httpProxyPort;
@@ -79,26 +81,36 @@ public class HttpProxyIntegrationTest extends AbstractSalesforceTestBase {
// start a local HTTP proxy using Jetty server
server = new Server();
- Connector connector = new SelectChannelConnector();
+/*
+ final SSLContextParameters contextParameters = new SSLContextParameters();
+ final SslContextFactory sslContextFactory = new SslContextFactory();
+ sslContextFactory.setSslContext(contextParameters.createSSLContext());
+ ServerConnector connector = new ServerConnector(server, sslContextFactory);
+*/
+ ServerConnector connector = new ServerConnector(server);
+
connector.setHost(HTTP_PROXY_HOST);
- server.setConnectors(new Connector[]{connector});
+ server.addConnector(connector);
final String authenticationString = "Basic "
+ B64Code.encode(HTTP_PROXY_USER_NAME + ":" + HTTP_PROXY_PASSWORD, StringUtil.__ISO_8859_1);
- ConnectHandler handler = new ConnectHandler() {
+ ConnectHandler connectHandler = new ConnectHandler() {
@Override
- protected boolean handleAuthentication(HttpServletRequest request, HttpServletResponse response, String address) throws ServletException, IOException {
+ protected boolean handleAuthentication(HttpServletRequest request, HttpServletResponse response, String address) {
// validate proxy-authentication header
- final String header = request.getHeader(HttpHeaders.PROXY_AUTHORIZATION);
+ final String header = request.getHeader(PROXY_AUTHORIZATION.toString());
if (!authenticationString.equals(header)) {
- throw new ServletException("Missing header " + HttpHeaders.PROXY_AUTHORIZATION);
+ LOG.warn("Missing header " + PROXY_AUTHORIZATION);
+ // ask for authentication header
+ response.setHeader(PROXY_AUTHENTICATE.toString(), String.format("Basic realm=\"%s\"", HTTP_PROXY_REALM));
+ return false;
}
- LOG.info("CONNECT exchange contains required header " + HttpHeaders.PROXY_AUTHORIZATION);
- return super.handleAuthentication(request, response, address);
+ LOG.info("Request contains required header " + PROXY_AUTHORIZATION);
+ return true;
}
};
- server.setHandler(handler);
+ server.setHandler(connectHandler);
LOG.info("Starting proxy server...");
server.start();
@@ -118,6 +130,8 @@ public class HttpProxyIntegrationTest extends AbstractSalesforceTestBase {
salesforce.setHttpProxyPort(httpProxyPort);
salesforce.setHttpProxyUsername(HTTP_PROXY_USER_NAME);
salesforce.setHttpProxyPassword(HTTP_PROXY_PASSWORD);
+ salesforce.setHttpProxyAuthUri(String.format("https://%s:%s", HTTP_PROXY_HOST, httpProxyPort));
+ salesforce.setHttpProxyRealm(HTTP_PROXY_REALM);
// set HTTP client properties
final HashMap<String, Object> properties = new HashMap<String, Object>();
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/RestApiIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/RestApiIntegrationTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/RestApiIntegrationTest.java
index e87a21f..ced55c5 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/RestApiIntegrationTest.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/RestApiIntegrationTest.java
@@ -23,6 +23,7 @@ import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.util.HashMap;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import com.thoughtworks.xstream.annotations.XStreamAlias;
@@ -45,11 +46,10 @@ import org.apache.camel.component.salesforce.dto.generated.Line_Item__c;
import org.apache.camel.component.salesforce.dto.generated.Merchandise__c;
import org.apache.camel.component.salesforce.dto.generated.QueryRecordsLine_Item__c;
import org.apache.camel.util.jsse.SSLContextParameters;
-import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.HttpExchange;
-import org.eclipse.jetty.client.RedirectListener;
-import org.eclipse.jetty.http.HttpMethods;
+import org.eclipse.jetty.client.api.ContentResponse;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.junit.Test;
@@ -74,21 +74,59 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
sslContextFactory.setSslContext(new SSLContextParameters().createSSLContext());
HttpClient httpClient = new HttpClient(sslContextFactory);
httpClient.setConnectTimeout(60000);
- httpClient.setTimeout(60000);
- httpClient.registerListener(RedirectListener.class.getName());
httpClient.start();
- ContentExchange logoutGet = new ContentExchange(true);
- logoutGet.setURL(sf.getLoginConfig().getLoginUrl() + "/services/oauth2/revoke?token=" + accessToken);
- logoutGet.setMethod(HttpMethods.GET);
- httpClient.send(logoutGet);
- assertEquals(HttpExchange.STATUS_COMPLETED, logoutGet.waitForDone());
- assertEquals(HttpStatus.OK_200, logoutGet.getResponseStatus());
+ String uri = sf.getLoginConfig().getLoginUrl() + "/services/oauth2/revoke?token=" + accessToken;
+ Request logoutGet = httpClient.newRequest(uri)
+ .method(HttpMethod.GET)
+ .timeout(1, TimeUnit.MINUTES);
+
+ ContentResponse response = logoutGet.send();
+ assertEquals(HttpStatus.OK_200, response.getStatus());
doTestGetGlobalObjects("");
}
@Test
+ public void testRetryFailure() throws Exception {
+ SalesforceComponent sf = context().getComponent("salesforce", SalesforceComponent.class);
+ String accessToken = sf.getSession().getAccessToken();
+
+ SslContextFactory sslContextFactory = new SslContextFactory();
+ sslContextFactory.setSslContext(new SSLContextParameters().createSSLContext());
+ HttpClient httpClient = new HttpClient(sslContextFactory);
+ httpClient.setConnectTimeout(60000);
+ httpClient.start();
+
+ String uri = sf.getLoginConfig().getLoginUrl() + "/services/oauth2/revoke?token=" + accessToken;
+ Request logoutGet = httpClient.newRequest(uri)
+ .method(HttpMethod.GET)
+ .timeout(1, TimeUnit.MINUTES);
+
+ ContentResponse response = logoutGet.send();
+ assertEquals(HttpStatus.OK_200, response.getStatus());
+
+ // set component config to bad password to cause relogin attempts to fail
+ final String password = sf.getLoginConfig().getPassword();
+ sf.getLoginConfig().setPassword("bad_password");
+
+ try {
+ doTestGetGlobalObjects("");
+ fail("Expected CamelExecutionException!");
+ } catch (CamelExecutionException e) {
+ if (e.getCause() instanceof SalesforceException) {
+ SalesforceException cause = (SalesforceException) e.getCause();
+ assertEquals("Expected 400 on authentication retry failure", HttpStatus.BAD_REQUEST_400, cause.getStatusCode());
+ } else {
+ fail("Expected SalesforceException!");
+ }
+ } finally {
+ // reset password and retries to allow other tests to pass
+ sf.getLoginConfig().setPassword(password);
+ }
+ }
+
+ @Test
public void testGetVersions() throws Exception {
doTestGetVersions("");
doTestGetVersions("Xml");
@@ -197,7 +235,7 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
doTestCreateUpdateDelete("Xml");
}
- private void doTestCreateUpdateDelete(String suffix) throws InterruptedException {
+ private void doTestCreateUpdateDelete(String suffix) throws Exception {
Merchandise__c merchandise = new Merchandise__c();
merchandise.setName("Wee Wee Wee Plane");
merchandise.setDescription__c("Microlite plane");
@@ -232,7 +270,7 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
doTestCreateUpdateDeleteWithId("Xml");
}
- private void doTestCreateUpdateDeleteWithId(String suffix) throws InterruptedException {
+ private void doTestCreateUpdateDeleteWithId(String suffix) throws Exception {
// get line item with Name 1
Line_Item__c lineItem = template().requestBody("direct:getSObjectWithId" + suffix, TEST_LINE_ITEM_ID,
Line_Item__c.class);
@@ -273,8 +311,13 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
@Test
public void testGetBlobField() throws Exception {
- doTestGetBlobField("");
- doTestGetBlobField("Xml");
+ SalesforceComponent component = context().getComponent("salesforce", SalesforceComponent.class);
+ try {
+ doTestGetBlobField("");
+ doTestGetBlobField("Xml");
+ } finally {
+ // reset response content buffer size
+ }
}
public void doTestGetBlobField(String suffix) throws Exception {
@@ -305,7 +348,7 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
doTestQuery("Xml");
}
- private void doTestQuery(String suffix) throws InterruptedException {
+ private void doTestQuery(String suffix) throws Exception {
QueryRecordsLine_Item__c queryRecords = template().requestBody("direct:query" + suffix, null,
QueryRecordsLine_Item__c.class);
assertNotNull(queryRecords);
@@ -320,7 +363,7 @@ public class RestApiIntegrationTest extends AbstractSalesforceTestBase {
}
@SuppressWarnings("unchecked")
- private void doTestSearch(String suffix) throws InterruptedException {
+ private void doTestSearch(String suffix) throws Exception {
Object obj = template().requestBody("direct:search" + suffix, (Object) null);
List<SearchResult> searchResults = null;
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/SessionIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/SessionIntegrationTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/SessionIntegrationTest.java
index a25ad52..c78720a 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/SessionIntegrationTest.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/SessionIntegrationTest.java
@@ -17,9 +17,8 @@
package org.apache.camel.component.salesforce.internal;
import org.apache.camel.component.salesforce.LoginConfigHelper;
+import org.apache.camel.component.salesforce.SalesforceHttpClient;
import org.apache.camel.util.jsse.SSLContextParameters;
-import org.eclipse.jetty.client.HttpClient;
-import org.eclipse.jetty.client.RedirectListener;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.junit.Assert;
import org.junit.Test;
@@ -41,16 +40,15 @@ public class SessionIntegrationTest extends Assert implements SalesforceSession.
final SslContextFactory sslContextFactory = new SslContextFactory();
sslContextFactory.setSslContext(new SSLContextParameters().createSSLContext());
- final HttpClient httpClient = new HttpClient(sslContextFactory);
+ final SalesforceHttpClient httpClient = new SalesforceHttpClient(sslContextFactory);
httpClient.setConnectTimeout(TIMEOUT);
- httpClient.setTimeout(TIMEOUT);
- httpClient.registerListener(RedirectListener.class.getName());
- httpClient.start();
final SalesforceSession session = new SalesforceSession(
- httpClient, LoginConfigHelper.getLoginConfig());
+ httpClient, TIMEOUT, LoginConfigHelper.getLoginConfig());
session.addListener(this);
+ httpClient.setSession(session);
+ httpClient.start();
try {
String loginToken = session.login(session.getAccessToken());
LOG.info("First token " + loginToken);
http://git-wip-us.apache.org/repos/asf/camel/blob/8dfd66bd/components/camel-salesforce/camel-salesforce-maven-plugin/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-salesforce/camel-salesforce-maven-plugin/pom.xml b/components/camel-salesforce/camel-salesforce-maven-plugin/pom.xml
index 95cdcf9..0250d1e 100644
--- a/components/camel-salesforce/camel-salesforce-maven-plugin/pom.xml
+++ b/components/camel-salesforce/camel-salesforce-maven-plugin/pom.xml
@@ -101,6 +101,12 @@
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-salesforce</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.cometd.java</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.velocity</groupId>
@@ -128,7 +134,19 @@
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
- <version>${jetty-version}</version>
+ <version>${jetty9-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-servlet</artifactId>
+ <version>${jetty9-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-proxy</artifactId>
+ <version>${jetty9-version}</version>
<scope>test</scope>
</dependency>
<dependency>