You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2019/07/02 06:38:54 UTC
[camel] branch async-aws-sns created (now 9605406)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch async-aws-sns
in repository https://gitbox.apache.org/repos/asf/camel.git.
at 9605406 https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883. Work in progress
This branch includes the following new commits:
new 9605406 https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883. Work in progress
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Re: [camel] 01/01: https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883.
Work in progress
Posted by Andrea Cosentino <an...@gmail.com>.
Personally, I think we have to add an option for using the async client and
leave to the end users the choice of using the plain or async client.
Il giorno mar 2 lug 2019 alle ore 08:38 <da...@apache.org> ha scritto:
> This is an automated email from the ASF dual-hosted git repository.
>
> davsclaus pushed a commit to branch async-aws-sns
> in repository https://gitbox.apache.org/repos/asf/camel.git
>
> commit 9605406b986b57edfd7415bd569654fab34a11d3
> Author: Claus Ibsen <cl...@gmail.com>
> AuthorDate: Tue Jul 2 08:38:09 2019 +0200
>
>
> https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883.
> Work in progress
> ---
> .../camel/component/aws/sns/SnsComponent.java | 3 +-
> .../camel/component/aws/sns/SnsConfiguration.java | 9 ++--
> .../camel/component/aws/sns/SnsEndpoint.java | 23 +++++----
> .../camel/component/aws/sns/SnsProducer.java | 56
> ++++++++++++----------
> 4 files changed, 53 insertions(+), 38 deletions(-)
>
> diff --git
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
> index aaa1bcd..2a488b3 100644
> ---
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
> +++
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
> @@ -22,6 +22,7 @@ import java.util.Set;
> import com.amazonaws.regions.Regions;
> import com.amazonaws.services.sns.AmazonSNS;
>
> +import com.amazonaws.services.sns.AmazonSNSAsync;
> import org.apache.camel.CamelContext;
> import org.apache.camel.Endpoint;
> import org.apache.camel.spi.Metadata;
> @@ -133,7 +134,7 @@ public class SnsComponent extends DefaultComponent {
> }
>
> private void checkAndSetRegistryClient(SnsConfiguration
> configuration) {
> - Set<AmazonSNS> clients =
> getCamelContext().getRegistry().findByType(AmazonSNS.class);
> + Set<AmazonSNSAsync> clients =
> getCamelContext().getRegistry().findByType(AmazonSNSAsync.class);
> if (clients.size() == 1) {
>
> configuration.setAmazonSNSClient(clients.stream().findFirst().get());
> }
> diff --git
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
> index 7690de8..a6bdc53 100644
> ---
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
> +++
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
> @@ -17,6 +17,9 @@
> package org.apache.camel.component.aws.sns;
>
> import com.amazonaws.services.sns.AmazonSNS;
> +import com.amazonaws.services.sns.AmazonSNSAsync;
> +import com.amazonaws.services.sns.AmazonSNSAsyncClient;
> +import com.amazonaws.services.sns.AmazonSNSClient;
> import com.amazonaws.services.sqs.AmazonSQS;
>
> import org.apache.camel.RuntimeCamelException;
> @@ -31,7 +34,7 @@ public class SnsConfiguration implements Cloneable {
> // Common properties
> private String topicName;
> @UriParam
> - private AmazonSNS amazonSNSClient;
> + private AmazonSNSAsync amazonSNSClient;
> @UriParam(label = "security", secret = true)
> private String accessKey;
> @UriParam(label = "security", secret = true)
> @@ -108,14 +111,14 @@ public class SnsConfiguration implements Cloneable {
> this.secretKey = secretKey;
> }
>
> - public AmazonSNS getAmazonSNSClient() {
> + public AmazonSNSAsync getAmazonSNSClient() {
> return amazonSNSClient;
> }
>
> /**
> * To use the AmazonSNS as the client
> */
> - public void setAmazonSNSClient(AmazonSNS amazonSNSClient) {
> + public void setAmazonSNSClient(AmazonSNSAsync amazonSNSClient) {
> this.amazonSNSClient = amazonSNSClient;
> }
>
> diff --git
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
> index 64636b4..b6e750b 100644
> ---
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
> +++
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
> @@ -27,6 +27,9 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider;
> import com.amazonaws.auth.BasicAWSCredentials;
> import com.amazonaws.regions.Regions;
> import com.amazonaws.services.sns.AmazonSNS;
> +import com.amazonaws.services.sns.AmazonSNSAsync;
> +import com.amazonaws.services.sns.AmazonSNSAsyncClient;
> +import com.amazonaws.services.sns.AmazonSNSAsyncClientBuilder;
> import com.amazonaws.services.sns.AmazonSNSClientBuilder;
> import com.amazonaws.services.sns.model.CreateTopicRequest;
> import com.amazonaws.services.sns.model.CreateTopicResult;
> @@ -55,7 +58,7 @@ import org.apache.camel.util.ObjectHelper;
> producerOnly = true, label = "cloud,mobile,messaging")
> public class SnsEndpoint extends DefaultEndpoint implements
> HeaderFilterStrategyAware {
>
> - private AmazonSNS snsClient;
> + private AmazonSNSAsync snsClient;
>
> @UriPath(description = "Topic name or ARN")
> @Metadata(required = true)
> @@ -178,11 +181,11 @@ public class SnsEndpoint extends DefaultEndpoint
> implements HeaderFilterStrategy
> this.configuration = configuration;
> }
>
> - public void setSNSClient(AmazonSNS snsClient) {
> + public void setSNSClient(AmazonSNSAsync snsClient) {
> this.snsClient = snsClient;
> }
>
> - public AmazonSNS getSNSClient() {
> + public AmazonSNSAsync getSNSClient() {
> return snsClient;
> }
>
> @@ -191,9 +194,9 @@ public class SnsEndpoint extends DefaultEndpoint
> implements HeaderFilterStrategy
> *
> * @return AmazonSNSClient
> */
> - AmazonSNS createSNSClient() {
> - AmazonSNS client = null;
> - AmazonSNSClientBuilder clientBuilder = null;
> + AmazonSNSAsync createSNSClient() {
> + AmazonSNSAsync client;
> + AmazonSNSAsyncClientBuilder clientBuilder;
> ClientConfiguration clientConfiguration = null;
> boolean isClientConfigFound = false;
> if (ObjectHelper.isNotEmpty(configuration.getProxyHost()) &&
> ObjectHelper.isNotEmpty(configuration.getProxyPort())) {
> @@ -206,15 +209,15 @@ public class SnsEndpoint extends DefaultEndpoint
> implements HeaderFilterStrategy
> AWSCredentials credentials = new
> BasicAWSCredentials(configuration.getAccessKey(),
> configuration.getSecretKey());
> AWSCredentialsProvider credentialsProvider = new
> AWSStaticCredentialsProvider(credentials);
> if (isClientConfigFound) {
> - clientBuilder =
> AmazonSNSClientBuilder.standard().withClientConfiguration(clientConfiguration).withCredentials(credentialsProvider);
> + clientBuilder =
> AmazonSNSAsyncClientBuilder.standard().withClientConfiguration(clientConfiguration).withCredentials(credentialsProvider);
> } else {
> - clientBuilder =
> AmazonSNSClientBuilder.standard().withCredentials(credentialsProvider);
> + clientBuilder =
> AmazonSNSAsyncClientBuilder.standard().withCredentials(credentialsProvider);
> }
> } else {
> if (isClientConfigFound) {
> - clientBuilder = AmazonSNSClientBuilder.standard();
> + clientBuilder = AmazonSNSAsyncClientBuilder.standard();
> } else {
> - clientBuilder =
> AmazonSNSClientBuilder.standard().withClientConfiguration(clientConfiguration);
> + clientBuilder =
> AmazonSNSAsyncClientBuilder.standard().withClientConfiguration(clientConfiguration);
> }
> }
> if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
> diff --git
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
> index 7eaea77..ac75ffc 100644
> ---
> a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
> +++
> b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
> @@ -22,22 +22,22 @@ import java.util.HashMap;
> import java.util.Map;
> import java.util.Map.Entry;
>
> +import com.amazonaws.handlers.AsyncHandler;
> import com.amazonaws.services.sns.model.MessageAttributeValue;
> import com.amazonaws.services.sns.model.PublishRequest;
> import com.amazonaws.services.sns.model.PublishResult;
> +import org.apache.camel.AsyncCallback;
> import org.apache.camel.Endpoint;
> import org.apache.camel.Exchange;
> -import org.apache.camel.Message;
> import org.apache.camel.spi.HeaderFilterStrategy;
> -import org.apache.camel.support.DefaultProducer;
> +import org.apache.camel.support.DefaultAsyncProducer;
> import org.apache.camel.util.URISupport;
>
> -
> /**
> * A Producer which sends messages to the Amazon Web Service Simple
> Notification Service
> * <a href="http://aws.amazon.com/sns/">AWS SNS</a>
> */
> -public class SnsProducer extends DefaultProducer {
> +public class SnsProducer extends DefaultAsyncProducer {
>
> private transient String snsProducerToString;
>
> @@ -45,23 +45,39 @@ public class SnsProducer extends DefaultProducer {
> super(endpoint);
> }
>
> - public void process(Exchange exchange) throws Exception {
> + @Override
> + public boolean process(final Exchange exchange, final AsyncCallback
> callback) {
> PublishRequest request = new PublishRequest();
> -
> - request.setTopicArn(getConfiguration().getTopicArn());
> - request.setSubject(determineSubject(exchange));
> - request.setMessageStructure(determineMessageStructure(exchange));
> - request.setMessage(exchange.getIn().getBody(String.class));
> -
> request.setMessageAttributes(this.translateAttributes(exchange.getIn().getHeaders(),
> exchange));
> + try {
> + request.setTopicArn(getConfiguration().getTopicArn());
> + request.setSubject(determineSubject(exchange));
> +
> request.setMessageStructure(determineMessageStructure(exchange));
> + request.setMessage(exchange.getIn().getBody(String.class));
> +
> request.setMessageAttributes(this.translateAttributes(exchange.getIn().getHeaders(),
> exchange));
> + } catch (Exception e) {
> + exchange.setException(e);
> + callback.done(true);
> + return true;
> + }
>
> log.trace("Sending request [{}] from exchange [{}]...", request,
> exchange);
> + getEndpoint().getSNSClient().publishAsync(request, new
> AsyncHandler<PublishRequest, PublishResult>() {
> + @Override
> + public void onError(Exception e) {
> + log.trace("Received error", e);
> + exchange.setException(e);
> + callback.done(false);
> + }
>
> - PublishResult result =
> getEndpoint().getSNSClient().publish(request);
> -
> - log.trace("Received result [{}]", result);
> + @Override
> + public void onSuccess(PublishRequest request, PublishResult
> result) {
> + log.trace("Received result [{}]", result);
> + exchange.getMessage().setHeader(SnsConstants.MESSAGE_ID,
> result.getMessageId());
> + callback.done(false);
> + }
> + });
>
> - Message message = getMessageForResponse(exchange);
> - message.setHeader(SnsConstants.MESSAGE_ID, result.getMessageId());
> + return false;
> }
>
> private String determineSubject(Exchange exchange) {
> @@ -130,12 +146,4 @@ public class SnsProducer extends DefaultProducer {
> return (SnsEndpoint) super.getEndpoint();
> }
>
> - public static Message getMessageForResponse(final Exchange exchange) {
> - if (exchange.getPattern().isOutCapable()) {
> - Message out = exchange.getOut();
> - out.copyFrom(exchange.getIn());
> - return out;
> - }
> - return exchange.getIn();
> - }
> }
> \ No newline at end of file
>
>
[camel] 01/01:
https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883.
Work in progress
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch async-aws-sns
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9605406b986b57edfd7415bd569654fab34a11d3
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Jul 2 08:38:09 2019 +0200
https://stackoverflow.com/questions/56775883/camel-sns-how-to-use-async-client-without-blocking?noredirect=1#comment100229953_56775883. Work in progress
---
.../camel/component/aws/sns/SnsComponent.java | 3 +-
.../camel/component/aws/sns/SnsConfiguration.java | 9 ++--
.../camel/component/aws/sns/SnsEndpoint.java | 23 +++++----
.../camel/component/aws/sns/SnsProducer.java | 56 ++++++++++++----------
4 files changed, 53 insertions(+), 38 deletions(-)
diff --git a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
index aaa1bcd..2a488b3 100644
--- a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
+++ b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsComponent.java
@@ -22,6 +22,7 @@ import java.util.Set;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sns.AmazonSNS;
+import com.amazonaws.services.sns.AmazonSNSAsync;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.spi.Metadata;
@@ -133,7 +134,7 @@ public class SnsComponent extends DefaultComponent {
}
private void checkAndSetRegistryClient(SnsConfiguration configuration) {
- Set<AmazonSNS> clients = getCamelContext().getRegistry().findByType(AmazonSNS.class);
+ Set<AmazonSNSAsync> clients = getCamelContext().getRegistry().findByType(AmazonSNSAsync.class);
if (clients.size() == 1) {
configuration.setAmazonSNSClient(clients.stream().findFirst().get());
}
diff --git a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
index 7690de8..a6bdc53 100644
--- a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
+++ b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsConfiguration.java
@@ -17,6 +17,9 @@
package org.apache.camel.component.aws.sns;
import com.amazonaws.services.sns.AmazonSNS;
+import com.amazonaws.services.sns.AmazonSNSAsync;
+import com.amazonaws.services.sns.AmazonSNSAsyncClient;
+import com.amazonaws.services.sns.AmazonSNSClient;
import com.amazonaws.services.sqs.AmazonSQS;
import org.apache.camel.RuntimeCamelException;
@@ -31,7 +34,7 @@ public class SnsConfiguration implements Cloneable {
// Common properties
private String topicName;
@UriParam
- private AmazonSNS amazonSNSClient;
+ private AmazonSNSAsync amazonSNSClient;
@UriParam(label = "security", secret = true)
private String accessKey;
@UriParam(label = "security", secret = true)
@@ -108,14 +111,14 @@ public class SnsConfiguration implements Cloneable {
this.secretKey = secretKey;
}
- public AmazonSNS getAmazonSNSClient() {
+ public AmazonSNSAsync getAmazonSNSClient() {
return amazonSNSClient;
}
/**
* To use the AmazonSNS as the client
*/
- public void setAmazonSNSClient(AmazonSNS amazonSNSClient) {
+ public void setAmazonSNSClient(AmazonSNSAsync amazonSNSClient) {
this.amazonSNSClient = amazonSNSClient;
}
diff --git a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
index 64636b4..b6e750b 100644
--- a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
+++ b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsEndpoint.java
@@ -27,6 +27,9 @@ import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sns.AmazonSNS;
+import com.amazonaws.services.sns.AmazonSNSAsync;
+import com.amazonaws.services.sns.AmazonSNSAsyncClient;
+import com.amazonaws.services.sns.AmazonSNSAsyncClientBuilder;
import com.amazonaws.services.sns.AmazonSNSClientBuilder;
import com.amazonaws.services.sns.model.CreateTopicRequest;
import com.amazonaws.services.sns.model.CreateTopicResult;
@@ -55,7 +58,7 @@ import org.apache.camel.util.ObjectHelper;
producerOnly = true, label = "cloud,mobile,messaging")
public class SnsEndpoint extends DefaultEndpoint implements HeaderFilterStrategyAware {
- private AmazonSNS snsClient;
+ private AmazonSNSAsync snsClient;
@UriPath(description = "Topic name or ARN")
@Metadata(required = true)
@@ -178,11 +181,11 @@ public class SnsEndpoint extends DefaultEndpoint implements HeaderFilterStrategy
this.configuration = configuration;
}
- public void setSNSClient(AmazonSNS snsClient) {
+ public void setSNSClient(AmazonSNSAsync snsClient) {
this.snsClient = snsClient;
}
- public AmazonSNS getSNSClient() {
+ public AmazonSNSAsync getSNSClient() {
return snsClient;
}
@@ -191,9 +194,9 @@ public class SnsEndpoint extends DefaultEndpoint implements HeaderFilterStrategy
*
* @return AmazonSNSClient
*/
- AmazonSNS createSNSClient() {
- AmazonSNS client = null;
- AmazonSNSClientBuilder clientBuilder = null;
+ AmazonSNSAsync createSNSClient() {
+ AmazonSNSAsync client;
+ AmazonSNSAsyncClientBuilder clientBuilder;
ClientConfiguration clientConfiguration = null;
boolean isClientConfigFound = false;
if (ObjectHelper.isNotEmpty(configuration.getProxyHost()) && ObjectHelper.isNotEmpty(configuration.getProxyPort())) {
@@ -206,15 +209,15 @@ public class SnsEndpoint extends DefaultEndpoint implements HeaderFilterStrategy
AWSCredentials credentials = new BasicAWSCredentials(configuration.getAccessKey(), configuration.getSecretKey());
AWSCredentialsProvider credentialsProvider = new AWSStaticCredentialsProvider(credentials);
if (isClientConfigFound) {
- clientBuilder = AmazonSNSClientBuilder.standard().withClientConfiguration(clientConfiguration).withCredentials(credentialsProvider);
+ clientBuilder = AmazonSNSAsyncClientBuilder.standard().withClientConfiguration(clientConfiguration).withCredentials(credentialsProvider);
} else {
- clientBuilder = AmazonSNSClientBuilder.standard().withCredentials(credentialsProvider);
+ clientBuilder = AmazonSNSAsyncClientBuilder.standard().withCredentials(credentialsProvider);
}
} else {
if (isClientConfigFound) {
- clientBuilder = AmazonSNSClientBuilder.standard();
+ clientBuilder = AmazonSNSAsyncClientBuilder.standard();
} else {
- clientBuilder = AmazonSNSClientBuilder.standard().withClientConfiguration(clientConfiguration);
+ clientBuilder = AmazonSNSAsyncClientBuilder.standard().withClientConfiguration(clientConfiguration);
}
}
if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
diff --git a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
index 7eaea77..ac75ffc 100644
--- a/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
+++ b/components/camel-aws-sns/src/main/java/org/apache/camel/component/aws/sns/SnsProducer.java
@@ -22,22 +22,22 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
+import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sns.model.MessageAttributeValue;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.PublishResult;
+import org.apache.camel.AsyncCallback;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
-import org.apache.camel.Message;
import org.apache.camel.spi.HeaderFilterStrategy;
-import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.DefaultAsyncProducer;
import org.apache.camel.util.URISupport;
-
/**
* A Producer which sends messages to the Amazon Web Service Simple Notification Service
* <a href="http://aws.amazon.com/sns/">AWS SNS</a>
*/
-public class SnsProducer extends DefaultProducer {
+public class SnsProducer extends DefaultAsyncProducer {
private transient String snsProducerToString;
@@ -45,23 +45,39 @@ public class SnsProducer extends DefaultProducer {
super(endpoint);
}
- public void process(Exchange exchange) throws Exception {
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
PublishRequest request = new PublishRequest();
-
- request.setTopicArn(getConfiguration().getTopicArn());
- request.setSubject(determineSubject(exchange));
- request.setMessageStructure(determineMessageStructure(exchange));
- request.setMessage(exchange.getIn().getBody(String.class));
- request.setMessageAttributes(this.translateAttributes(exchange.getIn().getHeaders(), exchange));
+ try {
+ request.setTopicArn(getConfiguration().getTopicArn());
+ request.setSubject(determineSubject(exchange));
+ request.setMessageStructure(determineMessageStructure(exchange));
+ request.setMessage(exchange.getIn().getBody(String.class));
+ request.setMessageAttributes(this.translateAttributes(exchange.getIn().getHeaders(), exchange));
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
log.trace("Sending request [{}] from exchange [{}]...", request, exchange);
+ getEndpoint().getSNSClient().publishAsync(request, new AsyncHandler<PublishRequest, PublishResult>() {
+ @Override
+ public void onError(Exception e) {
+ log.trace("Received error", e);
+ exchange.setException(e);
+ callback.done(false);
+ }
- PublishResult result = getEndpoint().getSNSClient().publish(request);
-
- log.trace("Received result [{}]", result);
+ @Override
+ public void onSuccess(PublishRequest request, PublishResult result) {
+ log.trace("Received result [{}]", result);
+ exchange.getMessage().setHeader(SnsConstants.MESSAGE_ID, result.getMessageId());
+ callback.done(false);
+ }
+ });
- Message message = getMessageForResponse(exchange);
- message.setHeader(SnsConstants.MESSAGE_ID, result.getMessageId());
+ return false;
}
private String determineSubject(Exchange exchange) {
@@ -130,12 +146,4 @@ public class SnsProducer extends DefaultProducer {
return (SnsEndpoint) super.getEndpoint();
}
- public static Message getMessageForResponse(final Exchange exchange) {
- if (exchange.getPattern().isOutCapable()) {
- Message out = exchange.getOut();
- out.copyFrom(exchange.getIn());
- return out;
- }
- return exchange.getIn();
- }
}
\ No newline at end of file