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 2013/08/21 15:06:42 UTC
[3/4] git commit: CAMEL-5932 updated consumer and producer for direct
EntityManager usage
CAMEL-5932 updated consumer and producer for direct EntityManager usage
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1efc537f
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1efc537f
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1efc537f
Branch: refs/heads/master
Commit: 1efc537f8752772f54c1e2e1ff84bf733ba1d055
Parents: 708c4cd
Author: Brett Meyer <br...@3riverdev.com>
Authored: Thu Aug 8 14:07:33 2013 -0400
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Aug 21 15:06:32 2013 +0200
----------------------------------------------------------------------
.../camel/component/jpa/JpaConstants.java | 2 +-
.../apache/camel/component/jpa/JpaConsumer.java | 28 ++++--
.../apache/camel/component/jpa/JpaEndpoint.java | 45 +++++-----
.../apache/camel/component/jpa/JpaProducer.java | 31 ++++---
.../jpa/JpaTemplateTransactionStrategy.java | 89 --------------------
.../component/jpa/TransactionStrategy.java | 33 --------
.../idempotent/jpa/JpaMessageIdRepository.java | 54 +++++++-----
7 files changed, 96 insertions(+), 186 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConstants.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConstants.java
index ed5b922..3e42d9f 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConstants.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConstants.java
@@ -23,7 +23,7 @@ package org.apache.camel.component.jpa;
*/
public final class JpaConstants {
- public static final String JPA_TEMPLATE = "CamelJpaTemplate";
+ public static final String ENTITYMANAGER = "CamelEntityManager";
private JpaConstants() {
// utility class
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
index 5f81041..10cdd2d 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
@@ -36,7 +36,9 @@ import org.apache.camel.util.CastUtils;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.orm.jpa.JpaCallback;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
/**
* @version
@@ -45,7 +47,8 @@ public class JpaConsumer extends ScheduledBatchPollingConsumer {
private static final Logger LOG = LoggerFactory.getLogger(JpaConsumer.class);
private final JpaEndpoint endpoint;
- private final TransactionStrategy template;
+ private final EntityManager entityManager;
+ private final TransactionTemplate transactionTemplate;
private QueryFactory queryFactory;
private DeleteHandler<Object> deleteHandler;
private String query;
@@ -66,7 +69,8 @@ public class JpaConsumer extends ScheduledBatchPollingConsumer {
public JpaConsumer(JpaEndpoint endpoint, Processor processor) {
super(endpoint, processor);
this.endpoint = endpoint;
- this.template = endpoint.createTransactionStrategy();
+ this.entityManager = endpoint.createEntityManager();
+ this.transactionTemplate = endpoint.createTransactionTemplate();
}
@Override
@@ -75,9 +79,11 @@ public class JpaConsumer extends ScheduledBatchPollingConsumer {
shutdownRunningTask = null;
pendingExchanges = 0;
- Object messagePolled = template.execute(new JpaCallback<Object>() {
- public Object doInJpa(EntityManager entityManager) throws PersistenceException {
- Queue<DataHolder> answer = new LinkedList<DataHolder>();
+ Object messagePolled = transactionTemplate.execute(new TransactionCallback<Object>() {
+ public Object doInTransaction(TransactionStatus status) {
+ entityManager.joinTransaction();
+
+ Queue<DataHolder> answer = new LinkedList<DataHolder>();
Query query = getQueryFactory().createQuery(entityManager);
configureParameters(query);
@@ -374,7 +380,15 @@ public class JpaConsumer extends ScheduledBatchPollingConsumer {
protected Exchange createExchange(Object result) {
Exchange exchange = endpoint.createExchange();
exchange.getIn().setBody(result);
- exchange.getIn().setHeader(JpaConstants.JPA_TEMPLATE, endpoint.getTemplate());
+ exchange.getIn().setHeader(JpaConstants.ENTITYMANAGER, entityManager);
return exchange;
}
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ // TODO: This should probably happen, but hitting an open transaction or flush in progress.
+ // Is there a thread holding onto it?
+// entityManager.close();
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaEndpoint.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaEndpoint.java
index 19ea714..cee83f2 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaEndpoint.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaEndpoint.java
@@ -17,9 +17,9 @@
package org.apache.camel.component.jpa;
import java.util.Map;
+
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
-import javax.persistence.Persistence;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
@@ -33,20 +33,21 @@ import org.apache.camel.support.ExpressionAdapter;
import org.apache.camel.util.CastUtils;
import org.apache.camel.util.IntrospectionSupport;
import org.apache.camel.util.ObjectHelper;
-import org.springframework.orm.jpa.JpaTemplate;
import org.springframework.orm.jpa.JpaTransactionManager;
+import org.springframework.orm.jpa.LocalEntityManagerFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
public class JpaEndpoint extends ScheduledPollEndpoint {
private EntityManagerFactory entityManagerFactory;
private PlatformTransactionManager transactionManager;
private String persistenceUnit = "camel";
- private JpaTemplate template;
private Expression producerExpression;
private int maximumResults = -1;
private Class<?> entityType;
- private Map<Object, Object> entityManagerProperties;
+ private Map<String, Object> entityManagerProperties;
private boolean consumeDelete = true;
private boolean consumeLockEntity = true;
private boolean flushOnSend = true;
@@ -116,7 +117,7 @@ public class JpaEndpoint extends ScheduledPollEndpoint {
super.configureProperties(options);
Map<String, Object> emProperties = IntrospectionSupport.extractProperties(options, "emf.");
if (emProperties != null) {
- setEntityManagerProperties(CastUtils.cast(emProperties));
+ setEntityManagerProperties(emProperties);
}
}
@@ -132,17 +133,6 @@ public class JpaEndpoint extends ScheduledPollEndpoint {
// Properties
// -------------------------------------------------------------------------
- public JpaTemplate getTemplate() {
- if (template == null) {
- template = createTemplate();
- }
- return template;
- }
-
- public void setTemplate(JpaTemplate template) {
- this.template = template;
- }
-
public Expression getProducerExpression() {
if (producerExpression == null) {
producerExpression = createProducerExpression();
@@ -192,14 +182,14 @@ public class JpaEndpoint extends ScheduledPollEndpoint {
this.transactionManager = transactionManager;
}
- public Map<Object, Object> getEntityManagerProperties() {
+ public Map<String, Object> getEntityManagerProperties() {
if (entityManagerProperties == null) {
- entityManagerProperties = System.getProperties();
+ entityManagerProperties = CastUtils.cast(System.getProperties());
}
return entityManagerProperties;
}
- public void setEntityManagerProperties(Map<Object, Object> entityManagerProperties) {
+ public void setEntityManagerProperties(Map<String, Object> entityManagerProperties) {
this.entityManagerProperties = entityManagerProperties;
}
@@ -258,12 +248,12 @@ public class JpaEndpoint extends ScheduledPollEndpoint {
ObjectHelper.notNull(getEntityManagerFactory(), "entityManagerFactory");
}
- protected JpaTemplate createTemplate() {
- return new JpaTemplate(getEntityManagerFactory());
- }
-
protected EntityManagerFactory createEntityManagerFactory() {
- return Persistence.createEntityManagerFactory(persistenceUnit, getEntityManagerProperties());
+ LocalEntityManagerFactoryBean emfBean = new LocalEntityManagerFactoryBean();
+ emfBean.setPersistenceUnitName(persistenceUnit);
+ emfBean.setJpaPropertyMap(getEntityManagerProperties());
+ emfBean.afterPropertiesSet();
+ return emfBean.getObject();
}
protected PlatformTransactionManager createTransactionManager() {
@@ -276,8 +266,11 @@ public class JpaEndpoint extends ScheduledPollEndpoint {
return getEntityManagerFactory().createEntityManager();
}
- protected TransactionStrategy createTransactionStrategy() {
- return JpaTemplateTransactionStrategy.newInstance(getTransactionManager(), getTemplate());
+ protected TransactionTemplate createTransactionTemplate() {
+ TransactionTemplate transactionTemplate = new TransactionTemplate(getTransactionManager());
+ transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
+ transactionTemplate.afterPropertiesSet();
+ return transactionTemplate;
}
protected Expression createProducerExpression() {
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
index 7e8f2da..c06a1b8 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
@@ -21,18 +21,20 @@ import java.util.Collection;
import java.util.List;
import javax.persistence.EntityManager;
-import javax.persistence.PersistenceException;
import org.apache.camel.Exchange;
import org.apache.camel.Expression;
import org.apache.camel.impl.DefaultProducer;
-import org.springframework.orm.jpa.JpaCallback;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
/**
* @version
*/
public class JpaProducer extends DefaultProducer {
- private final TransactionStrategy template;
+ private final EntityManager entityManager;
+ private final TransactionTemplate transactionTemplate;
private final JpaEndpoint endpoint;
private final Expression expression;
@@ -40,17 +42,19 @@ public class JpaProducer extends DefaultProducer {
super(endpoint);
this.endpoint = endpoint;
this.expression = expression;
- this.template = endpoint.createTransactionStrategy();
+ this.entityManager = endpoint.createEntityManager();
+ this.transactionTemplate = endpoint.createTransactionTemplate();
}
public void process(final Exchange exchange) {
- exchange.getIn().setHeader(JpaConstants.JPA_TEMPLATE, endpoint.getTemplate());
+ exchange.getIn().setHeader(JpaConstants.ENTITYMANAGER, entityManager);
final Object values = expression.evaluate(exchange, Object.class);
if (values != null) {
- template.execute(new JpaCallback<Object>() {
-
- public Object doInJpa(EntityManager entityManager) throws PersistenceException {
- if (values.getClass().isArray()) {
+ transactionTemplate.execute(new TransactionCallback<Object>() {
+ public Object doInTransaction(TransactionStatus status) {
+ entityManager.joinTransaction();
+
+ if (values.getClass().isArray()) {
Object[] array = (Object[]) values;
for (int index = 0; index < array.length; index++) {
Object managedEntity = save(array[index], entityManager);
@@ -80,6 +84,7 @@ public class JpaProducer extends DefaultProducer {
if (endpoint.isFlushOnSend()) {
entityManager.flush();
}
+
return null;
}
@@ -100,6 +105,12 @@ public class JpaProducer extends DefaultProducer {
}
});
}
- exchange.getIn().removeHeader(JpaConstants.JPA_TEMPLATE);
+ exchange.getIn().removeHeader(JpaConstants.ENTITYMANAGER);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ entityManager.close();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaTemplateTransactionStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaTemplateTransactionStrategy.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaTemplateTransactionStrategy.java
deleted file mode 100644
index 789fa46..0000000
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaTemplateTransactionStrategy.java
+++ /dev/null
@@ -1,89 +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.jpa;
-
-import javax.persistence.EntityManager;
-import javax.persistence.EntityManagerFactory;
-import javax.persistence.PersistenceException;
-
-import org.springframework.orm.jpa.JpaCallback;
-import org.springframework.orm.jpa.JpaTemplate;
-import org.springframework.orm.jpa.JpaTransactionManager;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionStatus;
-import org.springframework.transaction.support.TransactionCallback;
-import org.springframework.transaction.support.TransactionTemplate;
-
-/**
- * Delegates the strategy to the {@link JpaTemplate} and {@link TransactionTemplate} for transaction handling
- *
- * @version
- */
-public class JpaTemplateTransactionStrategy implements TransactionStrategy {
- private final JpaTemplate jpaTemplate;
- private final TransactionTemplate transactionTemplate;
-
- public JpaTemplateTransactionStrategy(JpaTemplate jpaTemplate, TransactionTemplate transactionTemplate) {
- this.jpaTemplate = jpaTemplate;
- this.transactionTemplate = transactionTemplate;
- }
-
- /**
- * Creates a new implementation from the given JPA factory
- */
- public static JpaTemplateTransactionStrategy newInstance(EntityManagerFactory emf) {
- JpaTemplate template = new JpaTemplate(emf);
- return newInstance(emf, template);
- }
-
- /**
- * Creates a new implementation from the given JPA factory and JPA template
- */
- public static JpaTemplateTransactionStrategy newInstance(EntityManagerFactory emf, JpaTemplate template) {
- JpaTransactionManager transactionManager = new JpaTransactionManager(emf);
- transactionManager.afterPropertiesSet();
-
- TransactionTemplate tranasctionTemplate = new TransactionTemplate(transactionManager);
- tranasctionTemplate.afterPropertiesSet();
-
- return new JpaTemplateTransactionStrategy(template, tranasctionTemplate);
- }
-
- /**
- * Creates a new implementation from the given Transaction Manager and JPA template
- */
- public static JpaTemplateTransactionStrategy newInstance(PlatformTransactionManager transactionManager, JpaTemplate template) {
- TransactionTemplate tranasctionTemplate = new TransactionTemplate(transactionManager);
- tranasctionTemplate.afterPropertiesSet();
-
- return new JpaTemplateTransactionStrategy(template, tranasctionTemplate);
- }
-
-
- public Object execute(final JpaCallback<?> callback) {
- return transactionTemplate.execute(new TransactionCallback<Object>() {
- public Object doInTransaction(TransactionStatus status) {
- return jpaTemplate.execute(new JpaCallback<Object>() {
- public Object doInJpa(EntityManager entityManager) throws PersistenceException {
- return callback.doInJpa(entityManager);
- }
- });
- }
- });
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/TransactionStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/TransactionStrategy.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/TransactionStrategy.java
deleted file mode 100644
index e49a742..0000000
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/TransactionStrategy.java
+++ /dev/null
@@ -1,33 +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.jpa;
-
-import org.springframework.orm.jpa.JpaCallback;
-
-/**
- * @version
- */
-public interface TransactionStrategy {
-
- /**
- * Executes in a transaction.
- *
- * @param callback the callback
- * @return the result
- */
- Object execute(JpaCallback<?> callback);
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/1efc537f/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java b/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
index b8a49e9..a46f61c 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
@@ -19,15 +19,16 @@ package org.apache.camel.processor.idempotent.jpa;
import java.util.Date;
import java.util.List;
+import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.Persistence;
+import javax.persistence.Query;
import org.apache.camel.api.management.ManagedAttribute;
import org.apache.camel.api.management.ManagedOperation;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.support.ServiceSupport;
-import org.springframework.orm.jpa.JpaTemplate;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -40,32 +41,31 @@ import org.springframework.transaction.support.TransactionTemplate;
@ManagedResource(description = "JPA based message id repository")
public class JpaMessageIdRepository extends ServiceSupport implements IdempotentRepository<String> {
protected static final String QUERY_STRING = "select x from " + MessageProcessed.class.getName() + " x where x.processorName = ?1 and x.messageId = ?2";
- private final JpaTemplate jpaTemplate;
private final String processorName;
+ private final EntityManager entityManager;
private final TransactionTemplate transactionTemplate;
- public JpaMessageIdRepository(JpaTemplate template, String processorName) {
- this(template, createTransactionTemplate(template), processorName);
+ public JpaMessageIdRepository(EntityManagerFactory entityManagerFactory, String processorName) {
+ this(entityManagerFactory, createTransactionTemplate(entityManagerFactory), processorName);
}
- public JpaMessageIdRepository(JpaTemplate template, TransactionTemplate transactionTemplate, String processorName) {
- this.jpaTemplate = template;
+ public JpaMessageIdRepository(EntityManagerFactory entityManagerFactory, TransactionTemplate transactionTemplate, String processorName) {
+ this.entityManager = entityManagerFactory.createEntityManager();
this.processorName = processorName;
this.transactionTemplate = transactionTemplate;
}
public static JpaMessageIdRepository jpaMessageIdRepository(String persistenceUnit, String processorName) {
- EntityManagerFactory entityManagerFactory = Persistence.createEntityManagerFactory(persistenceUnit);
- return jpaMessageIdRepository(new JpaTemplate(entityManagerFactory), processorName);
+ return jpaMessageIdRepository(Persistence.createEntityManagerFactory(persistenceUnit), processorName);
}
- public static JpaMessageIdRepository jpaMessageIdRepository(JpaTemplate jpaTemplate, String processorName) {
- return new JpaMessageIdRepository(jpaTemplate, processorName);
+ public static JpaMessageIdRepository jpaMessageIdRepository(EntityManagerFactory entityManagerFactory, String processorName) {
+ return new JpaMessageIdRepository(entityManagerFactory, processorName);
}
- private static TransactionTemplate createTransactionTemplate(JpaTemplate jpaTemplate) {
+ private static TransactionTemplate createTransactionTemplate(EntityManagerFactory entityManagerFactory) {
TransactionTemplate transactionTemplate = new TransactionTemplate();
- transactionTemplate.setTransactionManager(new JpaTransactionManager(jpaTemplate.getEntityManagerFactory()));
+ transactionTemplate.setTransactionManager(new JpaTransactionManager(entityManagerFactory));
transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
return transactionTemplate;
}
@@ -75,14 +75,16 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
// Run this in single transaction.
Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus arg0) {
- List<?> list = jpaTemplate.find(QUERY_STRING, processorName, messageId);
+ entityManager.joinTransaction();
+
+ List<?> list = query(messageId);
if (list.isEmpty()) {
MessageProcessed processed = new MessageProcessed();
processed.setProcessorName(processorName);
processed.setMessageId(messageId);
processed.setCreatedAt(new Date());
- jpaTemplate.persist(processed);
- jpaTemplate.flush();
+ entityManager.persist(processed);
+ entityManager.flush();
return Boolean.TRUE;
} else {
return Boolean.FALSE;
@@ -97,7 +99,9 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
// Run this in single transaction.
Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus arg0) {
- List<?> list = jpaTemplate.find(QUERY_STRING, processorName, messageId);
+ entityManager.joinTransaction();
+
+ List<?> list = query(messageId);
if (list.isEmpty()) {
return Boolean.FALSE;
} else {
@@ -112,19 +116,28 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
public boolean remove(final String messageId) {
Boolean rc = transactionTemplate.execute(new TransactionCallback<Boolean>() {
public Boolean doInTransaction(TransactionStatus arg0) {
- List<?> list = jpaTemplate.find(QUERY_STRING, processorName, messageId);
+ entityManager.joinTransaction();
+
+ List<?> list = query(messageId);
if (list.isEmpty()) {
return Boolean.FALSE;
} else {
- MessageProcessed processoed = (MessageProcessed) list.get(0);
- jpaTemplate.remove(processoed);
- jpaTemplate.flush();
+ MessageProcessed processed = (MessageProcessed) list.get(0);
+ entityManager.remove(processed);
+ entityManager.flush();
return Boolean.TRUE;
}
}
});
return rc.booleanValue();
}
+
+ private List<?> query(final String messageId) {
+ Query query = entityManager.createQuery(QUERY_STRING);
+ query.setParameter(1, processorName);
+ query.setParameter(2, messageId);
+ return query.getResultList();
+ }
public boolean confirm(String s) {
// noop
@@ -142,5 +155,6 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
@Override
protected void doStop() throws Exception {
+ entityManager.close();
}
}