This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch exchange-factory in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/exchange-factory by this push: new 3609376 CAMEL-16222: PooledExchangeFactory experiment 3609376 is described below commit 36093762e671bdd29156bf32092c0d6cb92798f8 Author: Claus Ibsen <claus.ib...@gmail.com> AuthorDate: Sun Feb 21 19:40:18 2021 +0100 CAMEL-16222: PooledExchangeFactory experiment --- .../component/git/consumer/GitBranchConsumer.java | 2 +- .../component/git/consumer/GitCommitConsumer.java | 2 +- .../component/git/consumer/GitTagConsumer.java | 2 +- .../component/github/consumer/CommitConsumer.java | 2 +- .../component/github/consumer/EventsConsumer.java | 2 +- .../consumer/PullRequestCommentConsumer.java | 2 +- .../github/consumer/PullRequestConsumer.java | 2 +- .../component/github/consumer/TagConsumer.java | 2 +- .../mail/stream/GoogleMailStreamConsumer.java | 43 ++++++++++++++- .../mail/stream/GoogleMailStreamEndpoint.java | 42 -------------- .../google/pubsub/GooglePubsubConsumer.java | 8 +-- .../pubsub/consumer/CamelMessageReceiver.java | 9 ++- .../sheets/stream/GoogleSheetsStreamConsumer.java | 48 ++++++++++++++-- .../sheets/stream/GoogleSheetsStreamEndpoint.java | 37 ------------- .../google/storage/GoogleCloudStorageConsumer.java | 64 +++++++++++++++++++++- .../google/storage/GoogleCloudStorageEndpoint.java | 59 -------------------- .../apache/camel/component/gora/GoraConsumer.java | 10 +--- .../guava/eventbus/CamelEventHandler.java | 18 ++++-- .../guava/eventbus/FilteringCamelEventHandler.java | 5 +- .../guava/eventbus/GuavaEventBusConsumer.java | 8 +-- .../guava/eventbus/GuavaEventBusEndpoint.java | 7 --- 21 files changed, 187 insertions(+), 187 deletions(-) diff --git a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java index 49d9456..6dc075d 100644 --- a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java +++ b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitBranchConsumer.java @@ -40,7 +40,7 @@ public class GitBranchConsumer extends AbstractGitConsumer { List<Ref> call = getGit().branchList().setListMode(ListMode.ALL).call(); for (Ref ref : call) { if (!branchesConsumed.contains(ref.getName())) { - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getMessage().setBody(ref.getName()); e.getMessage().setHeader(GitConstants.GIT_BRANCH_LEAF, ref.getLeaf().getName()); e.getMessage().setHeader(GitConstants.GIT_BRANCH_OBJECT_ID, ref.getObjectId().getName()); diff --git a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java index 06e203b..122a85a 100644 --- a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java +++ b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitCommitConsumer.java @@ -46,7 +46,7 @@ public class GitCommitConsumer extends AbstractGitConsumer { } for (RevCommit commit : commits) { if (!commitsConsumed.contains(commit.getId())) { - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getMessage().setBody(commit.getFullMessage()); e.getMessage().setHeader(GitConstants.GIT_COMMIT_ID, commit.getId()); e.getMessage().setHeader(GitConstants.GIT_COMMIT_AUTHOR_NAME, commit.getAuthorIdent().getName()); diff --git a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java index 31ea39a..7052764 100644 --- a/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java +++ b/components/camel-git/src/main/java/org/apache/camel/component/git/consumer/GitTagConsumer.java @@ -39,7 +39,7 @@ public class GitTagConsumer extends AbstractGitConsumer { List<Ref> call = getGit().tagList().call(); for (Ref ref : call) { if (!tagsConsumed.contains(ref.getName())) { - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getMessage().setBody(ref.getName()); e.getMessage().setHeader(GitConstants.GIT_BRANCH_LEAF, ref.getLeaf().getName()); e.getMessage().setHeader(GitConstants.GIT_BRANCH_OBJECT_ID, ref.getObjectId().getName()); diff --git a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/CommitConsumer.java b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/CommitConsumer.java index ac92e01..5a4bbf7 100644 --- a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/CommitConsumer.java +++ b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/CommitConsumer.java @@ -71,7 +71,7 @@ public class CommitConsumer extends AbstractGitHubConsumer { while (!newCommits.empty()) { RepositoryCommit newCommit = newCommits.pop(); - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getMessage().setHeader(GitHubConstants.GITHUB_COMMIT_AUTHOR, newCommit.getAuthor().getName()); e.getMessage().setHeader(GitHubConstants.GITHUB_COMMIT_COMMITTER, newCommit.getCommitter().getName()); e.getMessage().setHeader(GitHubConstants.GITHUB_COMMIT_SHA, newCommit.getSha()); diff --git a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/EventsConsumer.java b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/EventsConsumer.java index 343fe45..a47e763 100644 --- a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/EventsConsumer.java +++ b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/EventsConsumer.java @@ -78,7 +78,7 @@ public class EventsConsumer extends AbstractGitHubConsumer { lastEventId = Long.parseLong(latestEvent.getId()); for (Event event : newEvents) { - Exchange exchange = getEndpoint().createExchange(); + Exchange exchange = createExchange(true); exchange.getMessage().setBody(event.getType()); exchange.getMessage().setHeader(GitHubConstants.GITHUB_EVENT_PAYLOAD, event.getPayload()); getProcessor().process(exchange); diff --git a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestCommentConsumer.java b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestCommentConsumer.java index 7f2164b..fc4fa03 100644 --- a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestCommentConsumer.java +++ b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestCommentConsumer.java @@ -110,7 +110,7 @@ public class PullRequestCommentConsumer extends AbstractGitHubConsumer { while (!newComments.empty()) { Comment newComment = newComments.pop(); - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getIn().setBody(newComment); // Required by the producers. Set it here for convenience. diff --git a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestConsumer.java b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestConsumer.java index 9d6b804..61d839c 100644 --- a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestConsumer.java +++ b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/PullRequestConsumer.java @@ -76,7 +76,7 @@ public class PullRequestConsumer extends AbstractGitHubConsumer { while (!newPullRequests.empty()) { PullRequest newPullRequest = newPullRequests.pop(); - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getIn().setBody(newPullRequest); diff --git a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/TagConsumer.java b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/TagConsumer.java index 559cbba..91a7107 100644 --- a/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/TagConsumer.java +++ b/components/camel-github/src/main/java/org/apache/camel/component/github/consumer/TagConsumer.java @@ -56,7 +56,7 @@ public class TagConsumer extends AbstractGitHubConsumer { while (!newTags.empty()) { RepositoryTag newTag = newTags.pop(); - Exchange e = getEndpoint().createExchange(); + Exchange e = createExchange(true); e.getIn().setBody(newTag); getProcessor().process(e); } diff --git a/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java b/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java index 4d4f5e2..da70219 100644 --- a/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java +++ b/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamConsumer.java @@ -16,17 +16,22 @@ */ package org.apache.camel.component.google.mail.stream; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.LinkedList; import java.util.List; import java.util.Queue; +import com.google.api.client.util.Base64; import com.google.api.services.gmail.Gmail; import com.google.api.services.gmail.model.ListMessagesResponse; import com.google.api.services.gmail.model.Message; +import com.google.api.services.gmail.model.MessagePart; +import com.google.api.services.gmail.model.MessagePartHeader; import com.google.api.services.gmail.model.ModifyMessageRequest; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; import org.apache.camel.ExtendedExchange; import org.apache.camel.Processor; import org.apache.camel.spi.Synchronization; @@ -83,7 +88,7 @@ public class GoogleMailStreamConsumer extends ScheduledBatchPollingConsumer { if (c.getMessages() != null) { for (Message message : c.getMessages()) { Message mess = getClient().users().messages().get("me", message.getId()).setFormat("FULL").execute(); - Exchange exchange = getEndpoint().createExchange(getEndpoint().getExchangePattern(), mess); + Exchange exchange = createExchange(getEndpoint().getExchangePattern(), mess); answer.add(exchange); } } @@ -170,4 +175,40 @@ public class GoogleMailStreamConsumer extends ScheduledBatchPollingConsumer { } } + public Exchange createExchange(ExchangePattern pattern, com.google.api.services.gmail.model.Message mail) { + Exchange exchange = createExchange(true); + exchange.setPattern(pattern); + org.apache.camel.Message message = exchange.getIn(); + exchange.getIn().setHeader(GoogleMailStreamConstants.MAIL_ID, mail.getId()); + List<MessagePart> parts = mail.getPayload().getParts(); + if (parts != null && parts.get(0).getBody().getData() != null) { + byte[] bodyBytes = Base64.decodeBase64(parts.get(0).getBody().getData().trim()); + String body = new String(bodyBytes, StandardCharsets.UTF_8); + message.setBody(body); + } + configureHeaders(message, mail.getPayload().getHeaders()); + return exchange; + } + + private void configureHeaders(org.apache.camel.Message message, List<MessagePartHeader> headers) { + for (MessagePartHeader header : headers) { + String headerName = header.getName(); + if ("SUBJECT".equalsIgnoreCase(headerName)) { + message.setHeader(GoogleMailStreamConstants.MAIL_SUBJECT, header.getValue()); + } + if ("TO".equalsIgnoreCase(headerName)) { + message.setHeader(GoogleMailStreamConstants.MAIL_TO, header.getValue()); + } + if ("FROM".equalsIgnoreCase(headerName)) { + message.setHeader(GoogleMailStreamConstants.MAIL_FROM, header.getValue()); + } + if ("CC".equalsIgnoreCase(headerName)) { + message.setHeader(GoogleMailStreamConstants.MAIL_CC, header.getValue()); + } + if ("BCC".equalsIgnoreCase(headerName)) { + message.setHeader(GoogleMailStreamConstants.MAIL_BCC, header.getValue()); + } + } + } + } diff --git a/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamEndpoint.java b/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamEndpoint.java index 8c47b31..947f69a2 100644 --- a/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamEndpoint.java +++ b/components/camel-google-mail/src/main/java/org/apache/camel/component/google/mail/stream/GoogleMailStreamEndpoint.java @@ -16,22 +16,15 @@ */ package org.apache.camel.component.google.mail.stream; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; -import com.google.api.client.util.Base64; import com.google.api.services.gmail.Gmail; import com.google.api.services.gmail.model.Label; import com.google.api.services.gmail.model.ListLabelsResponse; -import com.google.api.services.gmail.model.MessagePart; -import com.google.api.services.gmail.model.MessagePartHeader; import com.google.common.base.Splitter; import org.apache.camel.Category; import org.apache.camel.Consumer; -import org.apache.camel.Exchange; -import org.apache.camel.ExchangePattern; -import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.Producer; import org.apache.camel.component.google.mail.GoogleMailClientFactory; @@ -108,41 +101,6 @@ public class GoogleMailStreamEndpoint extends ScheduledPollEndpoint { return configuration; } - public Exchange createExchange(ExchangePattern pattern, com.google.api.services.gmail.model.Message mail) { - Exchange exchange = super.createExchange(pattern); - Message message = exchange.getIn(); - exchange.getIn().setHeader(GoogleMailStreamConstants.MAIL_ID, mail.getId()); - List<MessagePart> parts = mail.getPayload().getParts(); - if (parts != null && parts.get(0).getBody().getData() != null) { - byte[] bodyBytes = Base64.decodeBase64(parts.get(0).getBody().getData().trim()); - String body = new String(bodyBytes, StandardCharsets.UTF_8); - message.setBody(body); - } - setHeaders(message, mail.getPayload().getHeaders()); - return exchange; - } - - private void setHeaders(Message message, List<MessagePartHeader> headers) { - for (MessagePartHeader header : headers) { - String headerName = header.getName(); - if ("SUBJECT".equalsIgnoreCase(headerName)) { - message.setHeader(GoogleMailStreamConstants.MAIL_SUBJECT, header.getValue()); - } - if ("TO".equalsIgnoreCase(headerName)) { - message.setHeader(GoogleMailStreamConstants.MAIL_TO, header.getValue()); - } - if ("FROM".equalsIgnoreCase(headerName)) { - message.setHeader(GoogleMailStreamConstants.MAIL_FROM, header.getValue()); - } - if ("CC".equalsIgnoreCase(headerName)) { - message.setHeader(GoogleMailStreamConstants.MAIL_CC, header.getValue()); - } - if ("BCC".equalsIgnoreCase(headerName)) { - message.setHeader(GoogleMailStreamConstants.MAIL_BCC, header.getValue()); - } - } - } - private List<String> splitLabels(String labels) { return Splitter.on(',').splitToList(labels); } diff --git a/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java b/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java index 360dadc..8adc5e1 100644 --- a/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java +++ b/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java @@ -40,7 +40,7 @@ import org.apache.camel.support.DefaultConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -class GooglePubsubConsumer extends DefaultConsumer { +public class GooglePubsubConsumer extends DefaultConsumer { private Logger localLog; @@ -123,7 +123,7 @@ class GooglePubsubConsumer extends DefaultConsumer { private void asynchronousPull(String subscriptionName) { while (isRunAllowed() && !isSuspendingOrSuspended()) { - MessageReceiver messageReceiver = new CamelMessageReceiver(endpoint, processor); + MessageReceiver messageReceiver = new CamelMessageReceiver(GooglePubsubConsumer.this, endpoint, processor); Subscriber subscriber = endpoint.getComponent().getSubscriber(subscriptionName, messageReceiver); try { @@ -152,7 +152,7 @@ class GooglePubsubConsumer extends DefaultConsumer { PullResponse pullResponse = subscriber.pullCallable().call(pullRequest); for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) { PubsubMessage pubsubMessage = message.getMessage(); - Exchange exchange = endpoint.createExchange(); + Exchange exchange = createExchange(true); exchange.getIn().setBody(pubsubMessage.getData().toByteArray()); exchange.getIn().setHeader(GooglePubsubConstants.ACK_ID, message.getAckId()); @@ -171,7 +171,7 @@ class GooglePubsubConsumer extends DefaultConsumer { try { processor.process(exchange); } catch (Exception e) { - exchange.setException(e); + getExceptionHandler().handleException(e); } } } catch (IOException e) { diff --git a/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/CamelMessageReceiver.java b/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/CamelMessageReceiver.java index 0b441b0..d26d649 100644 --- a/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/CamelMessageReceiver.java +++ b/components/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/CamelMessageReceiver.java @@ -24,6 +24,7 @@ import org.apache.camel.Exchange; import org.apache.camel.ExtendedExchange; import org.apache.camel.Processor; import org.apache.camel.component.google.pubsub.GooglePubsubConstants; +import org.apache.camel.component.google.pubsub.GooglePubsubConsumer; import org.apache.camel.component.google.pubsub.GooglePubsubEndpoint; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,10 +32,12 @@ import org.slf4j.LoggerFactory; public class CamelMessageReceiver implements MessageReceiver { private final Logger localLog; + private final GooglePubsubConsumer consumer; private final GooglePubsubEndpoint endpoint; private final Processor processor; - public CamelMessageReceiver(GooglePubsubEndpoint endpoint, Processor processor) { + public CamelMessageReceiver(GooglePubsubConsumer consumer, GooglePubsubEndpoint endpoint, Processor processor) { + this.consumer = consumer; this.endpoint = endpoint; this.processor = processor; String loggerId = endpoint.getLoggerId(); @@ -50,7 +53,7 @@ public class CamelMessageReceiver implements MessageReceiver { localLog.trace("Received message ID : {}", pubsubMessage.getMessageId()); } - Exchange exchange = endpoint.createExchange(); + Exchange exchange = consumer.createExchange(true); exchange.getIn().setBody(pubsubMessage.getData().toByteArray()); exchange.getIn().setHeader(GooglePubsubConstants.MESSAGE_ID, pubsubMessage.getMessageId()); @@ -67,7 +70,7 @@ public class CamelMessageReceiver implements MessageReceiver { try { processor.process(exchange); } catch (Exception e) { - exchange.setException(e); + consumer.getExceptionHandler().handleException(e); } } } diff --git a/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java b/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java index 5d382db..eb04b5d 100644 --- a/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java +++ b/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java @@ -19,6 +19,7 @@ package org.apache.camel.component.google.sheets.stream; import java.util.ArrayDeque; import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.Queue; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -29,6 +30,7 @@ import com.google.api.services.sheets.v4.model.Spreadsheet; import com.google.api.services.sheets.v4.model.ValueRange; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.support.ScheduledBatchPollingConsumer; import org.apache.camel.util.CastUtils; @@ -89,12 +91,12 @@ public class GoogleSheetsStreamConsumer extends ScheduledBatchPollingConsumer { if (getConfiguration().getMaxResults() > 0) { valueRange.getValues().stream() .limit(getConfiguration().getMaxResults()) - .map(values -> getEndpoint().createExchange(rangeIndex.get(), valueIndex.incrementAndGet(), + .map(values -> createExchange(rangeIndex.get(), valueIndex.incrementAndGet(), valueRange.getRange(), valueRange.getMajorDimension(), values)) .forEach(answer::add); } else { valueRange.getValues().stream() - .map(values -> getEndpoint().createExchange(rangeIndex.get(), valueIndex.incrementAndGet(), + .map(values -> createExchange(rangeIndex.get(), valueIndex.incrementAndGet(), valueRange.getRange(), valueRange.getMajorDimension(), values)) .forEach(answer::add); } @@ -112,7 +114,7 @@ public class GoogleSheetsStreamConsumer extends ScheduledBatchPollingConsumer { .collect(Collectors.toList())); } }) - .map(valueRange -> getEndpoint().createExchange(rangeIndex.incrementAndGet(), valueRange)) + .map(valueRange -> createExchange(rangeIndex.incrementAndGet(), valueRange)) .forEach(answer::add); } } @@ -122,7 +124,7 @@ public class GoogleSheetsStreamConsumer extends ScheduledBatchPollingConsumer { request.setIncludeGridData(getConfiguration().isIncludeGridData()); Spreadsheet spreadsheet = request.execute(); - answer.add(getEndpoint().createExchange(spreadsheet)); + answer.add(createExchange(spreadsheet)); } return processBatch(CastUtils.cast(answer)); @@ -148,4 +150,42 @@ public class GoogleSheetsStreamConsumer extends ScheduledBatchPollingConsumer { return total; } + + public Exchange createExchange(int rangeIndex, ValueRange valueRange) { + Exchange exchange = createExchange(true); + exchange.setPattern(getEndpoint().getExchangePattern()); + Message message = exchange.getIn(); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, + getEndpoint().getConfiguration().getSpreadsheetId()); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE, valueRange.getRange()); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE_INDEX, rangeIndex); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.MAJOR_DIMENSION, valueRange.getMajorDimension()); + message.setBody(valueRange); + return exchange; + } + + public Exchange createExchange(int rangeIndex, int valueIndex, String range, String majorDimension, List<Object> values) { + Exchange exchange = createExchange(true); + exchange.setPattern(getEndpoint().getExchangePattern()); + Message message = exchange.getIn(); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, + getEndpoint().getConfiguration().getSpreadsheetId()); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE_INDEX, rangeIndex); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.VALUE_INDEX, valueIndex); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE, range); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.MAJOR_DIMENSION, majorDimension); + message.setBody(values); + return exchange; + } + + public Exchange createExchange(Spreadsheet spreadsheet) { + Exchange exchange = createExchange(true); + exchange.setPattern(getEndpoint().getExchangePattern()); + Message message = exchange.getIn(); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, spreadsheet.getSpreadsheetId()); + exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_URL, spreadsheet.getSpreadsheetUrl()); + message.setBody(spreadsheet); + return exchange; + } + } diff --git a/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamEndpoint.java b/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamEndpoint.java index 18062f3..c17f148 100644 --- a/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamEndpoint.java +++ b/components/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamEndpoint.java @@ -16,15 +16,9 @@ */ package org.apache.camel.component.google.sheets.stream; -import java.util.List; - import com.google.api.services.sheets.v4.Sheets; -import com.google.api.services.sheets.v4.model.Spreadsheet; -import com.google.api.services.sheets.v4.model.ValueRange; import org.apache.camel.Category; import org.apache.camel.Consumer; -import org.apache.camel.Exchange; -import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.Producer; import org.apache.camel.component.google.sheets.GoogleSheetsClientFactory; @@ -80,35 +74,4 @@ public class GoogleSheetsStreamEndpoint extends ScheduledPollEndpoint { return configuration; } - public Exchange createExchange(int rangeIndex, ValueRange valueRange) { - Exchange exchange = super.createExchange(getExchangePattern()); - Message message = exchange.getIn(); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, configuration.getSpreadsheetId()); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE, valueRange.getRange()); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE_INDEX, rangeIndex); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.MAJOR_DIMENSION, valueRange.getMajorDimension()); - message.setBody(valueRange); - return exchange; - } - - public Exchange createExchange(int rangeIndex, int valueIndex, String range, String majorDimension, List<Object> values) { - Exchange exchange = super.createExchange(getExchangePattern()); - Message message = exchange.getIn(); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, configuration.getSpreadsheetId()); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE_INDEX, rangeIndex); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.VALUE_INDEX, valueIndex); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.RANGE, range); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.MAJOR_DIMENSION, majorDimension); - message.setBody(values); - return exchange; - } - - public Exchange createExchange(Spreadsheet spreadsheet) { - Exchange exchange = super.createExchange(getExchangePattern()); - Message message = exchange.getIn(); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_ID, spreadsheet.getSpreadsheetId()); - exchange.getIn().setHeader(GoogleSheetsStreamConstants.SPREADSHEET_URL, spreadsheet.getSpreadsheetUrl()); - message.setBody(spreadsheet); - return exchange; - } } diff --git a/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java b/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java index 935b7e6..b8cede8 100644 --- a/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java +++ b/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java @@ -16,6 +16,8 @@ */ package org.apache.camel.component.google.storage; +import java.io.ByteArrayOutputStream; +import java.util.Date; import java.util.LinkedList; import java.util.List; import java.util.Queue; @@ -28,8 +30,11 @@ import com.google.cloud.storage.Storage; import com.google.cloud.storage.Storage.CopyRequest; import org.apache.camel.AsyncCallback; import org.apache.camel.Exchange; +import org.apache.camel.ExchangePattern; import org.apache.camel.ExtendedExchange; +import org.apache.camel.Message; import org.apache.camel.Processor; +import org.apache.camel.RuntimeCamelException; import org.apache.camel.spi.Synchronization; import org.apache.camel.support.ScheduledBatchPollingConsumer; import org.apache.camel.util.CastUtils; @@ -73,7 +78,7 @@ public class GoogleCloudStorageConsumer extends ScheduledBatchPollingConsumer { String fileName = getConfiguration().getObjectName(); String bucketName = getConfiguration().getBucketName(); - Queue<Exchange> exchanges = new LinkedList<>(); + Queue<Exchange> exchanges; if (fileName != null) { LOG.trace("Getting object in bucket [{}] with file name [{}]...", bucketName, fileName); @@ -101,7 +106,7 @@ public class GoogleCloudStorageConsumer extends ScheduledBatchPollingConsumer { protected Queue<Exchange> createExchanges(Blob blob, String key) { Queue<Exchange> answer = new LinkedList<>(); - Exchange exchange = getEndpoint().createExchange(blob, key); + Exchange exchange = createExchange(blob, key); answer.add(exchange); return answer; } @@ -115,7 +120,7 @@ public class GoogleCloudStorageConsumer extends ScheduledBatchPollingConsumer { try { for (Blob blob : blobList) { if (includeObject(blob)) { - Exchange exchange = getEndpoint().createExchange(blob, blob.getBlobId().getName()); + Exchange exchange = createExchange(blob, blob.getBlobId().getName()); answer.add(exchange); } } @@ -248,4 +253,57 @@ public class GoogleCloudStorageConsumer extends ScheduledBatchPollingConsumer { public GoogleCloudStorageEndpoint getEndpoint() { return (GoogleCloudStorageEndpoint) super.getEndpoint(); } + + public Exchange createExchange(Blob blob, String key) { + return createExchange(getEndpoint().getExchangePattern(), blob, key); + } + + public Exchange createExchange(ExchangePattern pattern, Blob blob, String key) { + if (LOG.isTraceEnabled()) { + LOG.trace("Getting object with key [{}] from bucket [{}]...", key, getConfiguration().getBucketName()); + LOG.trace("Got object [{}]", blob); + } + + Exchange exchange = createExchange(true); + exchange.setPattern(pattern); + Message message = exchange.getIn(); + + if (getConfiguration().isIncludeBody()) { + try { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + blob.downloadTo(baos); + message.setBody(baos.toByteArray()); + } catch (Exception e) { + throw new RuntimeCamelException(e); + } + } else { + message.setBody(blob); + } + + message.setHeader(GoogleCloudStorageConstants.OBJECT_NAME, key); + message.setHeader(GoogleCloudStorageConstants.BUCKET_NAME, getConfiguration().getBucketName()); + //OTHER METADATA + message.setHeader(GoogleCloudStorageConstants.CACHE_CONTROL, blob.getCacheControl()); + message.setHeader(GoogleCloudStorageConstants.METADATA_COMPONENT_COUNT, blob.getComponentCount()); + message.setHeader(GoogleCloudStorageConstants.CONTENT_DISPOSITION, blob.getContentDisposition()); + message.setHeader(GoogleCloudStorageConstants.CONTENT_ENCODING, blob.getContentEncoding()); + message.setHeader(GoogleCloudStorageConstants.METADATA_CONTENT_LANGUAGE, blob.getContentLanguage()); + message.setHeader(GoogleCloudStorageConstants.CONTENT_TYPE, blob.getContentType()); + message.setHeader(GoogleCloudStorageConstants.METADATA_CUSTOM_TIME, blob.getCustomTime()); + message.setHeader(GoogleCloudStorageConstants.METADATA_CRC32C_HEX, blob.getCrc32cToHexString()); + message.setHeader(GoogleCloudStorageConstants.METADATA_ETAG, blob.getEtag()); + message.setHeader(GoogleCloudStorageConstants.METADATA_GENERATION, blob.getGeneration()); + message.setHeader(GoogleCloudStorageConstants.METADATA_BLOB_ID, blob.getBlobId()); + message.setHeader(GoogleCloudStorageConstants.METADATA_KMS_KEY_NAME, blob.getKmsKeyName()); + message.setHeader(GoogleCloudStorageConstants.CONTENT_MD5, blob.getMd5ToHexString()); + message.setHeader(GoogleCloudStorageConstants.METADATA_MEDIA_LINK, blob.getMediaLink()); + message.setHeader(GoogleCloudStorageConstants.METADATA_METAGENERATION, blob.getMetageneration()); + message.setHeader(GoogleCloudStorageConstants.CONTENT_LENGTH, blob.getSize()); + message.setHeader(GoogleCloudStorageConstants.METADATA_STORAGE_CLASS, blob.getStorageClass()); + message.setHeader(GoogleCloudStorageConstants.METADATA_CREATE_TIME, blob.getCreateTime()); + message.setHeader(GoogleCloudStorageConstants.METADATA_LAST_UPDATE, new Date(blob.getUpdateTime())); + + return exchange; + } + } diff --git a/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageEndpoint.java b/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageEndpoint.java index f3fe59a..79da387 100644 --- a/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageEndpoint.java +++ b/components/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageEndpoint.java @@ -16,10 +16,6 @@ */ package org.apache.camel.component.google.storage; -import java.io.ByteArrayOutputStream; -import java.util.Date; - -import com.google.cloud.storage.Blob; import com.google.cloud.storage.Bucket; import com.google.cloud.storage.BucketInfo; import com.google.cloud.storage.BucketInfo.Builder; @@ -27,12 +23,8 @@ import com.google.cloud.storage.Storage; import com.google.cloud.storage.StorageClass; import org.apache.camel.Category; import org.apache.camel.Consumer; -import org.apache.camel.Exchange; -import org.apache.camel.ExchangePattern; -import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.Producer; -import org.apache.camel.RuntimeCamelException; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.support.ScheduledPollEndpoint; @@ -123,55 +115,4 @@ public class GoogleCloudStorageEndpoint extends ScheduledPollEndpoint { return storageClient; } - public Exchange createExchange(Blob blob, String key) { - return createExchange(getExchangePattern(), blob, key); - } - - public Exchange createExchange(ExchangePattern pattern, Blob blob, String key) { - if (LOG.isTraceEnabled()) { - LOG.trace("Getting object with key [{}] from bucket [{}]...", key, getConfiguration().getBucketName()); - LOG.trace("Got object [{}]", blob); - } - - Exchange exchange = super.createExchange(pattern); - Message message = exchange.getIn(); - - if (configuration.isIncludeBody()) { - try { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - blob.downloadTo(baos); - message.setBody(baos.toByteArray()); - } catch (Exception e) { - throw new RuntimeCamelException(e); - } - } else { - message.setBody(blob); - } - - message.setHeader(GoogleCloudStorageConstants.OBJECT_NAME, key); - message.setHeader(GoogleCloudStorageConstants.BUCKET_NAME, getConfiguration().getBucketName()); - //OTHER METADATA - message.setHeader(GoogleCloudStorageConstants.CACHE_CONTROL, blob.getCacheControl()); - message.setHeader(GoogleCloudStorageConstants.METADATA_COMPONENT_COUNT, blob.getComponentCount()); - message.setHeader(GoogleCloudStorageConstants.CONTENT_DISPOSITION, blob.getContentDisposition()); - message.setHeader(GoogleCloudStorageConstants.CONTENT_ENCODING, blob.getContentEncoding()); - message.setHeader(GoogleCloudStorageConstants.METADATA_CONTENT_LANGUAGE, blob.getContentLanguage()); - message.setHeader(GoogleCloudStorageConstants.CONTENT_TYPE, blob.getContentType()); - message.setHeader(GoogleCloudStorageConstants.METADATA_CUSTOM_TIME, blob.getCustomTime()); - message.setHeader(GoogleCloudStorageConstants.METADATA_CRC32C_HEX, blob.getCrc32cToHexString()); - message.setHeader(GoogleCloudStorageConstants.METADATA_ETAG, blob.getEtag()); - message.setHeader(GoogleCloudStorageConstants.METADATA_GENERATION, blob.getGeneration()); - message.setHeader(GoogleCloudStorageConstants.METADATA_BLOB_ID, blob.getBlobId()); - message.setHeader(GoogleCloudStorageConstants.METADATA_KMS_KEY_NAME, blob.getKmsKeyName()); - message.setHeader(GoogleCloudStorageConstants.CONTENT_MD5, blob.getMd5ToHexString()); - message.setHeader(GoogleCloudStorageConstants.METADATA_MEDIA_LINK, blob.getMediaLink()); - message.setHeader(GoogleCloudStorageConstants.METADATA_METAGENERATION, blob.getMetageneration()); - message.setHeader(GoogleCloudStorageConstants.CONTENT_LENGTH, blob.getSize()); - message.setHeader(GoogleCloudStorageConstants.METADATA_STORAGE_CLASS, blob.getStorageClass()); - message.setHeader(GoogleCloudStorageConstants.METADATA_CREATE_TIME, blob.getCreateTime()); - message.setHeader(GoogleCloudStorageConstants.METADATA_LAST_UPDATE, new Date(blob.getUpdateTime())); - - return exchange; - } - } diff --git a/components/camel-gora/src/main/java/org/apache/camel/component/gora/GoraConsumer.java b/components/camel-gora/src/main/java/org/apache/camel/component/gora/GoraConsumer.java index 8bd0fc4..876c109 100644 --- a/components/camel-gora/src/main/java/org/apache/camel/component/gora/GoraConsumer.java +++ b/components/camel-gora/src/main/java/org/apache/camel/component/gora/GoraConsumer.java @@ -76,7 +76,7 @@ public class GoraConsumer extends ScheduledPollConsumer { @Override protected int poll() throws Exception { - final Exchange exchange = this.getEndpoint().createExchange(); + final Exchange exchange = createExchange(true); // compute time (approx) since last update if (firstRun) { @@ -88,13 +88,7 @@ public class GoraConsumer extends ScheduledPollConsumer { //proceed with query final Result result = query.execute(); - try { - getProcessor().process(exchange); - } finally { - if (exchange.getException() != null) { - getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException()); - } - } + getProcessor().process(exchange); return Long.valueOf(result.getOffset()).intValue(); } diff --git a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/CamelEventHandler.java b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/CamelEventHandler.java index b5d8bcc..9b53f57 100644 --- a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/CamelEventHandler.java +++ b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/CamelEventHandler.java @@ -31,14 +31,16 @@ import org.slf4j.LoggerFactory; public class CamelEventHandler { protected final Logger log = LoggerFactory.getLogger(CamelEventHandler.class); - protected final GuavaEventBusEndpoint eventBusEndpoint; + protected final GuavaEventBusConsumer consumer; + protected final GuavaEventBusEndpoint endpoint; protected final AsyncProcessor processor; - public CamelEventHandler(GuavaEventBusEndpoint eventBusEndpoint, Processor processor) { - ObjectHelper.notNull(eventBusEndpoint, "eventBusEndpoint"); + public CamelEventHandler(GuavaEventBusConsumer consumer, GuavaEventBusEndpoint endpoint, Processor processor) { + ObjectHelper.notNull(endpoint, "eventBusEndpoint"); ObjectHelper.notNull(processor, "processor"); - this.eventBusEndpoint = eventBusEndpoint; + this.consumer = consumer; + this.endpoint = endpoint; this.processor = AsyncProcessorConverterHelper.convert(processor); } @@ -49,7 +51,7 @@ public class CamelEventHandler { */ public void doEventReceived(Object event) { log.trace("Received event: {}", event); - final Exchange exchange = eventBusEndpoint.createExchange(event); + final Exchange exchange = createExchange(event); log.debug("Processing event: {}", event); // use async processor to support async routing engine processor.process(exchange, new AsyncCallback() { @@ -60,4 +62,10 @@ public class CamelEventHandler { }); } + public Exchange createExchange(Object event) { + Exchange exchange = consumer.createExchange(true); + exchange.getIn().setBody(event); + return exchange; + } + } diff --git a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/FilteringCamelEventHandler.java b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/FilteringCamelEventHandler.java index 7b01d3b..c950d3f 100644 --- a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/FilteringCamelEventHandler.java +++ b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/FilteringCamelEventHandler.java @@ -27,8 +27,9 @@ public class FilteringCamelEventHandler extends CamelEventHandler { private final Class<?> eventClass; - public FilteringCamelEventHandler(GuavaEventBusEndpoint eventBusEndpoint, Processor processor, Class<?> eventClass) { - super(eventBusEndpoint, processor); + public FilteringCamelEventHandler(GuavaEventBusConsumer consumer, GuavaEventBusEndpoint endpoint, Processor processor, + Class<?> eventClass) { + super(consumer, endpoint, processor); this.eventClass = eventClass; } diff --git a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusConsumer.java b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusConsumer.java index 1740117..bdf9500 100644 --- a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusConsumer.java +++ b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusConsumer.java @@ -50,7 +50,7 @@ public class GuavaEventBusConsumer extends DefaultConsumer { if (listenerInterface != null) { this.eventHandler = createListenerInterfaceProxy(endpoint, processor, listenerInterface); } else { - this.eventHandler = new FilteringCamelEventHandler(endpoint, processor, eventClass); + this.eventHandler = new FilteringCamelEventHandler(this, endpoint, processor, eventClass); } } @@ -72,7 +72,7 @@ public class GuavaEventBusConsumer extends DefaultConsumer { GuavaEventBusEndpoint endpoint, Processor processor, Class<?> listenerInterface) { ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); return Proxy.newProxyInstance(classLoader, new Class[] { listenerInterface }, - new ListenerInterfaceHandler(endpoint, processor)); + new ListenerInterfaceHandler(this, endpoint, processor)); } private static final class ListenerInterfaceHandler implements InvocationHandler { @@ -81,8 +81,8 @@ public class GuavaEventBusConsumer extends DefaultConsumer { private final CamelEventHandler delegateHandler; - private ListenerInterfaceHandler(GuavaEventBusEndpoint endpoint, Processor processor) { - this.delegateHandler = new CamelEventHandler(endpoint, processor); + private ListenerInterfaceHandler(GuavaEventBusConsumer consumer, GuavaEventBusEndpoint endpoint, Processor processor) { + this.delegateHandler = new CamelEventHandler(consumer, endpoint, processor); } @Override diff --git a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusEndpoint.java b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusEndpoint.java index d246eb6..83181b8 100644 --- a/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusEndpoint.java +++ b/components/camel-guava-eventbus/src/main/java/org/apache/camel/component/guava/eventbus/GuavaEventBusEndpoint.java @@ -20,7 +20,6 @@ import com.google.common.eventbus.EventBus; import org.apache.camel.Category; import org.apache.camel.Component; import org.apache.camel.Consumer; -import org.apache.camel.Exchange; import org.apache.camel.MultipleConsumersSupport; import org.apache.camel.Processor; import org.apache.camel.Producer; @@ -69,12 +68,6 @@ public class GuavaEventBusEndpoint extends DefaultEndpoint implements MultipleCo return true; } - public Exchange createExchange(Object event) { - Exchange exchange = createExchange(); - exchange.getIn().setBody(event); - return exchange; - } - public String getEventBusRef() { return eventBusRef; }