You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@camel.apache.org by Andrea Cosentino <an...@gmail.com> on 2019/07/02 07:00:30 UTC
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
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
>
>