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
commit 2c612eb53bc40fff40fc71faa679236cf4304c07 Author: Claus Ibsen <claus.ib...@gmail.com> AuthorDate: Mon Feb 22 08:04:44 2021 +0100 CAMEL-16222: PooledExchangeFactory experiment --- .../apache/camel/component/mail/MailConsumer.java | 11 ++- .../apache/camel/component/mail/MailEndpoint.java | 7 -- .../component/milo/client/MiloClientConsumer.java | 5 +- .../component/milo/server/MiloServerConsumer.java | 4 +- .../apache/camel/component/mina/MinaConsumer.java | 99 +++++++++++++--------- .../apache/camel/component/mina/MinaEndpoint.java | 11 --- .../mina/MinaTransferExchangeOptionTest.java | 1 - .../camel/component/minio/MinioConsumer.java | 36 +++++++- .../camel/component/minio/MinioEndpoint.java | 40 +-------- .../apache/camel/component/mllp/MllpEndpoint.java | 2 +- .../component/mllp/MllpTcpServerConsumer.java | 8 +- .../component/mongodb/gridfs/GridFsConsumer.java | 2 +- .../mongodb/MongoDbChangeStreamsConsumer.java | 4 + .../mongodb/MongoDbChangeStreamsThread.java | 14 ++- .../camel/component/mongodb/MongoDbEndpoint.java | 12 --- .../component/mongodb/MongoDbTailingThread.java | 14 ++- .../camel/component/mybatis/MyBatisConsumer.java | 21 +++-- .../mybatis/MyBatisConsumerIsolatedTest.java | 53 ------------ 18 files changed, 157 insertions(+), 187 deletions(-) diff --git a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java index dd541b2..6fe7825 100644 --- a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java +++ b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java @@ -385,7 +385,7 @@ public class MailConsumer extends ScheduledBatchPollingConsumer { } if (!message.getFlags().contains(Flags.Flag.DELETED)) { - Exchange exchange = getEndpoint().createExchange(message); + Exchange exchange = createExchange(message); if (getEndpoint().getConfiguration().isMapMailMessage()) { // ensure the mail message is mapped, which can be ensured by touching the body/header/attachment LOG.trace("Mapping #{} from javax.mail.Message to Camel MailMessage", i); @@ -399,6 +399,8 @@ public class MailConsumer extends ScheduledBatchPollingConsumer { exchange.getIn(AttachmentMessage.class).setAttachmentObjects(att); } } catch (MessagingException | IOException e) { + // must release exchange before throwing exception + releaseExchange(exchange, true); throw new RuntimeCamelException("Error accessing attachments due to: " + e.getMessage(), e); } } @@ -511,6 +513,13 @@ public class MailConsumer extends ScheduledBatchPollingConsumer { } } + private Exchange createExchange(Message message) { + Exchange exchange = createExchange(true); + exchange.setProperty(Exchange.BINDING, getEndpoint().getBinding()); + exchange.setIn(new MailMessage(exchange, message, getEndpoint().getConfiguration().isMapMailMessage())); + return exchange; + } + private void copyOrMoveMessageIfRequired( MailConfiguration config, Message mail, String destinationFolder, boolean moveMessage) throws MessagingException { diff --git a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailEndpoint.java b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailEndpoint.java index c7723cd..1ce1be9 100644 --- a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailEndpoint.java +++ b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailEndpoint.java @@ -126,13 +126,6 @@ public class MailEndpoint extends ScheduledPollEndpoint implements HeaderFilterS return answer; } - public Exchange createExchange(Message message) { - Exchange exchange = super.createExchange(); - exchange.setProperty(Exchange.BINDING, getBinding()); - exchange.setIn(new MailMessage(exchange, message, getConfiguration().isMapMailMessage())); - return exchange; - } - // Properties // ------------------------------------------------------------------------- diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java index 408af8c..6068910 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java @@ -69,10 +69,9 @@ public class MiloClientConsumer extends DefaultConsumer { private void handleValueUpdate(final DataValue value) { LOG.debug("Handle item update - {} = {}", node, value); - final Exchange exchange = getEndpoint().createExchange(); - mapToMessage(value, exchange.getMessage()); - + final Exchange exchange = createExchange(true); try { + mapToMessage(value, exchange.getMessage()); getProcessor().process(exchange); } catch (final Exception e) { getExceptionHandler().handleException("Error processing exchange", e); diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/server/MiloServerConsumer.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/server/MiloServerConsumer.java index f958974..7677e17 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/server/MiloServerConsumer.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/server/MiloServerConsumer.java @@ -54,10 +54,10 @@ public class MiloServerConsumer extends DefaultConsumer { } protected void performWrite(final DataValue value) { - Exchange exchange = getEndpoint().createExchange(); - mapToMessage(value, exchange.getMessage()); + Exchange exchange = createExchange(true); try { + mapToMessage(value, exchange.getMessage()); getProcessor().process(exchange); } catch (Exception e) { getExceptionHandler().handleException("Error processing exchange", e); diff --git a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java index 1152f8f..76917e3 100644 --- a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java +++ b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaConsumer.java @@ -373,6 +373,21 @@ public class MinaConsumer extends DefaultConsumer { this.acceptor = acceptor; } + private Exchange createExchange(IoSession session, Object payload) { + Exchange exchange; + if (configuration.isTransferExchange()) { + // do not release + exchange = getEndpoint().createExchange(); + } else { + exchange = createExchange(false); + } + exchange.getIn().setHeader(MinaConstants.MINA_IOSESSION, session); + exchange.getIn().setHeader(MinaConstants.MINA_LOCAL_ADDRESS, session.getLocalAddress()); + exchange.getIn().setHeader(MinaConstants.MINA_REMOTE_ADDRESS, session.getRemoteAddress()); + MinaPayloadHelper.setIn(exchange, payload); + return exchange; + } + /** * Handles consuming messages and replying if the exchange is out capable. */ @@ -406,7 +421,7 @@ public class MinaConsumer extends DefaultConsumer { LOG.debug("Received body: {}", in); } - Exchange exchange = getEndpoint().createExchange(session, object); + Exchange exchange = createExchange(session, object); //Set the exchange charset property for converting if (getEndpoint().getConfiguration().getCharsetName() != null) { exchange.setProperty(Exchange.CHARSET_NAME, @@ -419,50 +434,54 @@ public class MinaConsumer extends DefaultConsumer { getExceptionHandler().handleException(e); } - // - // If there's a response to send, send it. - // - boolean disconnect = getEndpoint().getConfiguration().isDisconnect(); - Object response = null; - if (exchange.hasOut()) { - response = MinaPayloadHelper.getOut(getEndpoint(), exchange); - } else { - response = MinaPayloadHelper.getIn(getEndpoint(), exchange); - } - - boolean failed = exchange.isFailed(); - if (failed && !getEndpoint().getConfiguration().isTransferExchange()) { - if (exchange.getException() != null) { - response = exchange.getException(); + try { + // + // If there's a response to send, send it. + // + boolean disconnect = getEndpoint().getConfiguration().isDisconnect(); + Object response; + if (exchange.hasOut()) { + response = MinaPayloadHelper.getOut(getEndpoint(), exchange); } else { - // failed and no exception, must be a fault - response = exchange.getOut().getBody(); + response = MinaPayloadHelper.getIn(getEndpoint(), exchange); } - } - if (response != null) { - LOG.debug("Writing body: {}", response); - MinaHelper.writeBody(session, response, exchange, configuration.getWriteTimeout()); - } else { - LOG.debug("Writing no response"); - disconnect = Boolean.TRUE; - } + boolean failed = exchange.isFailed(); + if (failed && !getEndpoint().getConfiguration().isTransferExchange()) { + if (exchange.getException() != null) { + response = exchange.getException(); + } else { + // failed and no exception, must be a fault + response = exchange.getOut().getBody(); + } + } - // should session be closed after complete? - Boolean close; - if (ExchangeHelper.isOutCapable(exchange)) { - close = exchange.getOut().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); - } else { - close = exchange.getIn().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); - } + if (response != null) { + LOG.debug("Writing body: {}", response); + MinaHelper.writeBody(session, response, exchange, configuration.getWriteTimeout()); + } else { + LOG.debug("Writing no response"); + disconnect = Boolean.TRUE; + } - // should we disconnect, the header can override the configuration - if (close != null) { - disconnect = close; - } - if (disconnect) { - LOG.debug("Closing session when complete at address: {}", address); - session.closeNow(); + // should session be closed after complete? + Boolean close; + if (ExchangeHelper.isOutCapable(exchange)) { + close = exchange.getOut().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); + } else { + close = exchange.getIn().getHeader(MinaConstants.MINA_CLOSE_SESSION_WHEN_COMPLETE, Boolean.class); + } + + // should we disconnect, the header can override the configuration + if (close != null) { + disconnect = close; + } + if (disconnect) { + LOG.debug("Closing session when complete at address: {}", address); + session.closeNow(); + } + } finally { + releaseExchange(exchange, false); } } } diff --git a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java index 2d3784a..416579a 100644 --- a/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java +++ b/components/camel-mina/src/main/java/org/apache/camel/component/mina/MinaEndpoint.java @@ -19,7 +19,6 @@ package org.apache.camel.component.mina; 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; @@ -27,7 +26,6 @@ import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.support.DefaultEndpoint; import org.apache.camel.util.ObjectHelper; -import org.apache.mina.core.session.IoSession; /** * Socket level networking using TCP or UDP with Apache Mina 2.x. @@ -68,15 +66,6 @@ public class MinaEndpoint extends DefaultEndpoint implements MultipleConsumersSu return answer; } - public Exchange createExchange(IoSession session, Object payload) { - Exchange exchange = createExchange(); - exchange.getIn().setHeader(MinaConstants.MINA_IOSESSION, session); - exchange.getIn().setHeader(MinaConstants.MINA_LOCAL_ADDRESS, session.getLocalAddress()); - exchange.getIn().setHeader(MinaConstants.MINA_REMOTE_ADDRESS, session.getRemoteAddress()); - MinaPayloadHelper.setIn(exchange, payload); - return exchange; - } - @Override public boolean isMultipleConsumersSupported() { // only datagram should allow multiple consumers diff --git a/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaTransferExchangeOptionTest.java b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaTransferExchangeOptionTest.java index c0354ae..140617b 100644 --- a/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaTransferExchangeOptionTest.java +++ b/components/camel-mina/src/test/java/org/apache/camel/component/mina/MinaTransferExchangeOptionTest.java @@ -58,7 +58,6 @@ public class MinaTransferExchangeOptionTest extends BaseMinaTest { String.format("mina:tcp://localhost:%1$s?sync=true&encoding=UTF-8&transferExchange=true", getPort())); Producer producer = endpoint.createProducer(); Exchange exchange = endpoint.createExchange(); - //Exchange exchange = endpoint.createExchange(); Message message = exchange.getIn(); message.setBody("Hello!"); diff --git a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java index df0a7d9..e98b295 100644 --- a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java +++ b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java @@ -39,9 +39,11 @@ import io.minio.errors.MinioException; import io.minio.messages.Item; import org.apache.camel.Exchange; import org.apache.camel.ExtendedExchange; +import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.spi.Synchronization; import org.apache.camel.support.ScheduledBatchPollingConsumer; +import org.apache.camel.support.SynchronizationAdapter; import org.apache.camel.util.CastUtils; import org.apache.camel.util.IOHelper; import org.apache.camel.util.URISupport; @@ -182,7 +184,7 @@ public class MinioConsumer extends ScheduledBatchPollingConsumer { protected Queue<Exchange> createExchanges(InputStream objectStream, String objectName) throws Exception { Queue<Exchange> answer = new LinkedList<>(); - Exchange exchange = getEndpoint().createExchange(objectStream, objectName); + Exchange exchange = createExchange(objectStream, objectName); answer.add(exchange); IOHelper.close(objectStream); return answer; @@ -200,7 +202,7 @@ public class MinioConsumer extends ScheduledBatchPollingConsumer { Item minioObjectSummary = minioObjectSummaries.next().get(); InputStream minioObject = getObject(bucketName, getMinioClient(), minioObjectSummary.objectName()); minioObjects.add(minioObject); - Exchange exchange = getEndpoint().createExchange(minioObject, minioObjectSummary.objectName()); + Exchange exchange = createExchange(minioObject, minioObjectSummary.objectName()); answer.add(exchange); continuationToken = minioObjectSummary.objectName(); } while (minioObjectSummaries.hasNext()); @@ -212,7 +214,7 @@ public class MinioConsumer extends ScheduledBatchPollingConsumer { if (!minioObjectSummary.isDir()) { InputStream minioObject = getObject(bucketName, getMinioClient(), minioObjectSummary.objectName()); minioObjects.add(minioObject); - Exchange exchange = getEndpoint().createExchange(minioObject, minioObjectSummary.objectName()); + Exchange exchange = createExchange(minioObject, minioObjectSummary.objectName()); answer.add(exchange); continuationToken = minioObjectSummary.objectName(); } @@ -401,6 +403,34 @@ public class MinioConsumer extends ScheduledBatchPollingConsumer { return (MinioEndpoint) super.getEndpoint(); } + private Exchange createExchange(InputStream minioObject, String objectName) throws Exception { + LOG.trace("Getting object with objectName {} from bucket {}...", objectName, getConfiguration().getBucketName()); + + Exchange exchange = createExchange(true); + exchange.setPattern(getEndpoint().getExchangePattern()); + Message message = exchange.getIn(); + LOG.trace("Got object!"); + + getEndpoint().getObjectStat(objectName, message); + + if (getConfiguration().isIncludeBody()) { + message.setBody(getEndpoint().readInputStream(minioObject)); + if (getConfiguration().isAutoCloseBody()) { + exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() { + @Override + public void onDone(Exchange exchange) { + IOHelper.close(minioObject); + } + }); + } + } else { + message.setBody(null); + IOHelper.close(minioObject); + } + + return exchange; + } + @Override public String toString() { if (isEmpty(minioConsumerToString)) { diff --git a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioEndpoint.java b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioEndpoint.java index e85cf8f..bb67186 100644 --- a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioEndpoint.java +++ b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioEndpoint.java @@ -32,17 +32,12 @@ import io.minio.StatObjectResponse; import org.apache.camel.Category; import org.apache.camel.Component; import org.apache.camel.Consumer; -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.Producer; import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.support.ScheduledPollEndpoint; -import org.apache.camel.support.SynchronizationAdapter; -import org.apache.camel.util.IOHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -123,37 +118,6 @@ public class MinioEndpoint extends ScheduledPollEndpoint { super.doStop(); } - public Exchange createExchange(InputStream minioObject, String objectName) throws Exception { - return createExchange(getExchangePattern(), minioObject, objectName); - } - - public Exchange createExchange(ExchangePattern pattern, InputStream minioObject, String objectName) throws Exception { - LOG.trace("Getting object with objectName {} from bucket {}...", objectName, getConfiguration().getBucketName()); - - Exchange exchange = super.createExchange(pattern); - Message message = exchange.getIn(); - LOG.trace("Got object!"); - - getObjectStat(objectName, message); - - if (getConfiguration().isIncludeBody()) { - message.setBody(readInputStream(minioObject)); - if (getConfiguration().isAutoCloseBody()) { - exchange.adapt(ExtendedExchange.class).addOnCompletion(new SynchronizationAdapter() { - @Override - public void onDone(Exchange exchange) { - IOHelper.close(minioObject); - } - }); - } - } else { - message.setBody(null); - IOHelper.close(minioObject); - } - - return exchange; - } - public MinioConfiguration getConfiguration() { return configuration; } @@ -196,7 +160,7 @@ public class MinioEndpoint extends ScheduledPollEndpoint { } } - private String readInputStream(InputStream minioObject) throws IOException { + String readInputStream(InputStream minioObject) throws IOException { StringBuilder textBuilder = new StringBuilder(); try (Reader reader = new BufferedReader(new InputStreamReader(minioObject, StandardCharsets.UTF_8))) { int c; @@ -227,7 +191,7 @@ public class MinioEndpoint extends ScheduledPollEndpoint { LOG.trace("Bucket policy updated"); } - private void getObjectStat(String objectName, Message message) throws Exception { + void getObjectStat(String objectName, Message message) throws Exception { String bucketName = getConfiguration().getBucketName(); StatObjectArgs.Builder statObjectRequest = StatObjectArgs.builder().bucket(bucketName).object(objectName); diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java index d1e3a4a..35719f0 100644 --- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java +++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java @@ -109,7 +109,7 @@ public class MllpEndpoint extends DefaultEndpoint { super.setBridgeErrorHandler(configuration.isBridgeErrorHandler()); } - private void setExchangeProperties(Exchange mllpExchange) { + void setExchangeProperties(Exchange mllpExchange) { if (configuration.hasCharsetName()) { mllpExchange.setProperty(Exchange.CHARSET_NAME, configuration.getCharsetName()); } diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java index d43c356..2ca75bb 100644 --- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java +++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java @@ -228,7 +228,8 @@ public class MllpTcpServerConsumer extends DefaultConsumer { // Send the message on to Camel for processing and wait for the response log.debug("processMessage(hl7MessageBytes[{}], {}) - populating the exchange with received payload", hl7MessageBytes == null ? -1 : hl7MessageBytes.length, consumerRunnable.getSocket()); - Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOut); + Exchange exchange = createExchange(false); + exchange.setPattern(ExchangePattern.InOut); if (getConfiguration().hasCharsetName()) { exchange.setProperty(Exchange.CHARSET_NAME, getConfiguration().getCharsetName()); } @@ -283,9 +284,8 @@ public class MllpTcpServerConsumer extends DefaultConsumer { "processMessage(byte[], TcpSocketConsumerRunnable) - Unexpected exception creating Unit of Work", exchange, uowEx); } finally { - if (exchange != null) { - doneUoW(exchange); - } + doneUoW(exchange); + releaseExchange(exchange, false); } } diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java index 60eff1d..74fe66a 100644 --- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java +++ b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java @@ -142,7 +142,7 @@ public class GridFsConsumer extends DefaultConsumer implements Runnable { forig = endpoint.getFilesCollection().findOneAndUpdate(filter, update, options); } if (forig != null) { - Exchange exchange = endpoint.createExchange(); + Exchange exchange = createExchange(true); GridFSDownloadStream downloadStream = endpoint.getGridFsBucket().openDownloadStream(file.getFilename()); file = downloadStream.getGridFSFile(); diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java index f625205..4fd3b23 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsConsumer.java @@ -19,10 +19,13 @@ package org.apache.camel.component.mongodb; import java.util.List; import java.util.concurrent.ExecutorService; +import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.support.DefaultConsumer; import org.apache.camel.util.ObjectHelper; import org.bson.BsonDocument; +import org.bson.Document; import static java.util.Collections.singletonList; @@ -69,4 +72,5 @@ public class MongoDbChangeStreamsConsumer extends DefaultConsumer { changeStreamsThread.init(); executor.execute(changeStreamsThread); } + } diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java index 116b072..8757793 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbChangeStreamsThread.java @@ -24,6 +24,7 @@ import com.mongodb.client.MongoCursor; import com.mongodb.client.model.changestream.ChangeStreamDocument; import com.mongodb.client.model.changestream.OperationType; import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.bson.BsonDocument; import org.bson.Document; import org.bson.types.ObjectId; @@ -70,7 +71,7 @@ class MongoDbChangeStreamsThread extends MongoAbstractConsumerThread { try { while (cursor.hasNext() && keepRunning) { ChangeStreamDocument<Document> dbObj = (ChangeStreamDocument<Document>) cursor.next(); - Exchange exchange = endpoint.createMongoDbExchange(dbObj.getFullDocument()); + Exchange exchange = createMongoDbExchange(dbObj.getFullDocument()); ObjectId documentId = dbObj.getDocumentKey().getObjectId(MONGO_ID).getValue(); OperationType operationType = dbObj.getOperationType(); @@ -101,4 +102,15 @@ class MongoDbChangeStreamsThread extends MongoAbstractConsumerThread { } } } + + private Exchange createMongoDbExchange(Document dbObj) { + Exchange exchange = consumer.createExchange(true); + Message message = exchange.getIn(); + message.setHeader(MongoDbConstants.DATABASE, endpoint.getDatabase()); + message.setHeader(MongoDbConstants.COLLECTION, endpoint.getCollection()); + message.setHeader(MongoDbConstants.FROM_TAILABLE, true); + message.setBody(dbObj); + return exchange; + } + } diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java index 76ddc9a..a118fea 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbEndpoint.java @@ -31,8 +31,6 @@ import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoDatabase; 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.spi.Metadata; @@ -318,16 +316,6 @@ public class MongoDbEndpoint extends DefaultEndpoint { } } - public Exchange createMongoDbExchange(Document dbObj) { - Exchange exchange = super.createExchange(); - Message message = exchange.getIn(); - message.setHeader(MongoDbConstants.DATABASE, database); - message.setHeader(MongoDbConstants.COLLECTION, collection); - message.setHeader(MongoDbConstants.FROM_TAILABLE, true); - message.setBody(dbObj); - return exchange; - } - @Override protected void doStart() throws Exception { if (mongoConnection == null) { diff --git a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbTailingThread.java b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbTailingThread.java index 5f870d1..b2c7747 100644 --- a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbTailingThread.java +++ b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbTailingThread.java @@ -20,6 +20,7 @@ import com.mongodb.CursorType; import com.mongodb.MongoCursorNotFoundException; import com.mongodb.client.MongoCursor; import org.apache.camel.Exchange; +import org.apache.camel.Message; import org.bson.Document; import static com.mongodb.client.model.Filters.gt; @@ -110,7 +111,7 @@ class MongoDbTailingThread extends MongoAbstractConsumerThread { try { while (cursor.hasNext() && keepRunning) { Document dbObj = (Document) cursor.next(); - Exchange exchange = endpoint.createMongoDbExchange(dbObj); + Exchange exchange = createMongoDbExchange(dbObj); try { if (log.isTraceEnabled()) { log.trace("Sending exchange: {}, ObjectId: {}", exchange, dbObj.get(MONGO_ID)); @@ -145,4 +146,15 @@ class MongoDbTailingThread extends MongoAbstractConsumerThread { tailTracking.persistToStore(); } } + + Exchange createMongoDbExchange(Document dbObj) { + Exchange exchange = consumer.createExchange(true); + Message message = exchange.getIn(); + message.setHeader(MongoDbConstants.DATABASE, endpoint.getDatabase()); + message.setHeader(MongoDbConstants.COLLECTION, endpoint.getCollection()); + message.setHeader(MongoDbConstants.FROM_TAILABLE, true); + message.setBody(dbObj); + return exchange; + } + } diff --git a/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java b/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java index 45bfca1..bfc721d 100644 --- a/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java +++ b/components/camel-mybatis/src/main/java/org/apache/camel/component/mybatis/MyBatisConsumer.java @@ -130,9 +130,10 @@ public class MyBatisConsumer extends ScheduledBatchPollingConsumer { // process the current exchange LOG.debug("Processing exchange: {} with properties: {}", exchange, exchange.getProperties()); - getProcessor().process(exchange); - + Exception cause = null; try { + getProcessor().process(exchange); + if (onConsume != null) { endpoint.getProcessingStrategy().commit(endpoint, exchange, data, onConsume); } @@ -142,13 +143,16 @@ public class MyBatisConsumer extends ScheduledBatchPollingConsumer { if (getEndpoint().isTransacted() && exchange.isFailed()) { // break out as we are transacted and should rollback - Exception cause = exchange.getException(); - if (cause != null) { - throw cause; - } else { - throw new RollbackExchangeException("Rollback transaction due error processing exchange", exchange); + cause = exchange.getException(); + if (cause == null) { + cause = new RollbackExchangeException("Rollback transaction due error processing exchange", null); } } + releaseExchange(exchange, false); + + if (cause != null) { + throw cause; + } } return total; @@ -156,7 +160,8 @@ public class MyBatisConsumer extends ScheduledBatchPollingConsumer { private Exchange createExchange(Object data) { final MyBatisEndpoint endpoint = getEndpoint(); - final Exchange exchange = endpoint.createExchange(ExchangePattern.InOnly); + final Exchange exchange = createExchange(false); + exchange.setPattern(ExchangePattern.InOnly); final String outputHeader = getEndpoint().getOutputHeader(); Message msg = exchange.getIn(); diff --git a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisConsumerIsolatedTest.java b/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisConsumerIsolatedTest.java deleted file mode 100644 index 27ad1db..0000000 --- a/components/camel-mybatis/src/test/java/org/apache/camel/component/mybatis/MyBatisConsumerIsolatedTest.java +++ /dev/null @@ -1,53 +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.mybatis; - -import java.util.ArrayDeque; -import java.util.Queue; - -import org.apache.camel.CamelContext; -import org.apache.camel.Processor; -import org.apache.camel.support.DefaultExchange; -import org.junit.jupiter.api.Test; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.mockito.Mockito.mock; - -public class MyBatisConsumerIsolatedTest { - - @Test - public void shouldRespectBatchSize() throws Exception { - // Given - int batchSize = 5; - MyBatisConsumer consumer = new MyBatisConsumer(mock(MyBatisEndpoint.class), mock(Processor.class)); - consumer.setMaxMessagesPerPoll(batchSize); - - Queue<Object> emptyMessageQueue = new ArrayDeque<>(); - for (int i = 0; i < 10; i++) { - MyBatisConsumer.DataHolder dataHolder = new MyBatisConsumer.DataHolder(); - dataHolder.exchange = new DefaultExchange(mock(CamelContext.class)); - emptyMessageQueue.add(dataHolder); - } - - // When - int processedMessages = consumer.processBatch(emptyMessageQueue); - - // Then - assertEquals(batchSize, processedMessages); - } - -}