Coverage Report - org.apache.camel.bam.processor.BamProcessorSupport
 
Classes in this File Line Coverage Branch Coverage Complexity
BamProcessorSupport
37% 
40% 
0
 
 1  
 /**
 2  
  * Licensed to the Apache Software Foundation (ASF) under one or more
 3  
  * contributor license agreements.  See the NOTICE file distributed with
 4  
  * this work for additional information regarding copyright ownership.
 5  
  * The ASF licenses this file to You under the Apache License, Version 2.0
 6  
  * (the "License"); you may not use this file except in compliance with
 7  
  * the License.  You may obtain a copy of the License at
 8  
  *
 9  
  *      http://www.apache.org/licenses/LICENSE-2.0
 10  
  *
 11  
  * Unless required by applicable law or agreed to in writing, software
 12  
  * distributed under the License is distributed on an "AS IS" BASIS,
 13  
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 14  
  * See the License for the specific language governing permissions and
 15  
  * limitations under the License.
 16  
  */
 17  
 package org.apache.camel.bam.processor;
 18  
 
 19  
 import org.apache.camel.Exchange;
 20  
 import org.apache.camel.Expression;
 21  
 import org.apache.camel.Processor;
 22  
 import org.apache.camel.RuntimeCamelException;
 23  
 import org.apache.commons.logging.Log;
 24  
 import org.apache.commons.logging.LogFactory;
 25  
 import org.springframework.dao.DataIntegrityViolationException;
 26  
 import org.springframework.orm.jpa.JpaSystemException;
 27  
 import org.springframework.transaction.TransactionStatus;
 28  
 import org.springframework.transaction.support.TransactionCallback;
 29  
 import org.springframework.transaction.support.TransactionTemplate;
 30  
 
 31  
 import javax.persistence.EntityExistsException;
 32  
 import java.lang.reflect.ParameterizedType;
 33  
 import java.lang.reflect.Type;
 34  
 
 35  
 /**
 36  
  * A base {@link Processor} for working on <a
 37  
  * href="http://activemq.apache.org/camel/bam.html">BAM</a> which a derived
 38  
  * class would do the actual persistence such as the {@link JpaBamProcessor}
 39  
  *
 40  
  * @version $Revision: $
 41  
  */
 42  6
 public abstract class BamProcessorSupport<T> implements Processor {
 43  2
     private static final transient Log LOG = LogFactory.getLog(BamProcessorSupport.class);
 44  
     private Class<T> entityType;
 45  
     private Expression<Exchange> correlationKeyExpression;
 46  
     private TransactionTemplate transactionTemplate;
 47  4
     private int maximumRetries = 30;
 48  
 
 49  
     public int getMaximumRetries() {
 50  0
         return maximumRetries;
 51  
     }
 52  
 
 53  
     public void setMaximumRetries(int maximumRetries) {
 54  0
         this.maximumRetries = maximumRetries;
 55  0
     }
 56  
 
 57  0
     protected BamProcessorSupport(TransactionTemplate transactionTemplate, Expression<Exchange> correlationKeyExpression) {
 58  0
         this.transactionTemplate = transactionTemplate;
 59  0
         this.correlationKeyExpression = correlationKeyExpression;
 60  
 
 61  0
         Type type = getClass().getGenericSuperclass();
 62  0
         if (type instanceof ParameterizedType) {
 63  0
             ParameterizedType parameterizedType = (ParameterizedType) type;
 64  0
             Type[] arguments = parameterizedType.getActualTypeArguments();
 65  0
             if (arguments.length > 0) {
 66  0
                 Type argumentType = arguments[0];
 67  0
                 if (argumentType instanceof Class) {
 68  0
                     this.entityType = (Class<T>) argumentType;
 69  
                 }
 70  
             }
 71  
         }
 72  0
         if (entityType == null) {
 73  0
             throw new IllegalArgumentException("Could not infer the entity type!");
 74  
         }
 75  0
     }
 76  
 
 77  4
     protected BamProcessorSupport(TransactionTemplate transactionTemplate, Expression<Exchange> correlationKeyExpression, Class<T> entitytype) {
 78  4
         this.transactionTemplate = transactionTemplate;
 79  4
         this.entityType = entitytype;
 80  4
         this.correlationKeyExpression = correlationKeyExpression;
 81  4
     }
 82  
 
 83  
     public void process(final Exchange exchange) {
 84  6
         Object entity = null;
 85  12
         for (int i = 0; entity == null && i < maximumRetries; i++) {
 86  6
             if (i > 0) {
 87  0
                 LOG.info("Retry attempt due to duplicate row: " + i);
 88  
             }
 89  6
             entity = transactionTemplate.execute(new TransactionCallback() {
 90  6
                 public Object doInTransaction(TransactionStatus status) {
 91  
                     try {
 92  6
                         Object key = getCorrelationKey(exchange);
 93  
 
 94  6
                         T entity = loadEntity(exchange, key);
 95  
 
 96  6
                         if (LOG.isDebugEnabled()) {
 97  0
                             LOG.debug("Correlation key: " + key + " with entity: " + entity);
 98  
                         }
 99  6
                         processEntity(exchange, entity);
 100  
 
 101  6
                         return entity;
 102  
                     }
 103  0
                     catch (JpaSystemException e) {
 104  0
                         if (LOG.isDebugEnabled()) {
 105  0
                             LOG.debug("Likely exception is due to duplicate row in concurrent setting: " + e, e);
 106  
                         }
 107  0
                         LOG.info("Attempt to insert duplicate row due to concurrency issue, so retrying: " + e);
 108  0
                         return retryDueToDuplicate(status);
 109  
                     }
 110  0
                     catch (DataIntegrityViolationException e) {
 111  0
                         Throwable throwable = e.getCause();
 112  0
                         if (throwable instanceof EntityExistsException) {
 113  0
                             LOG.info("Attempt to insert duplicate row due to concurrency issue, so retrying: " + throwable);
 114  0
                             return retryDueToDuplicate(status);
 115  
                         }
 116  0
                         return onError(status, throwable);
 117  
                     }
 118  0
                     catch (Throwable e) {
 119  0
                         return onError(status, e);
 120  
                     }
 121  
                 }
 122  
             });
 123  
         }
 124  6
     }
 125  
 
 126  
     // Properties
 127  
     // -----------------------------------------------------------------------
 128  
     public Expression<Exchange> getCorrelationKeyExpression() {
 129  0
         return correlationKeyExpression;
 130  
     }
 131  
 
 132  
     public Class<T> getEntityType() {
 133  10
         return entityType;
 134  
     }
 135  
 
 136  
     // Implemenation methods
 137  
     // -----------------------------------------------------------------------
 138  
     protected abstract void processEntity(Exchange exchange, T entity) throws Exception;
 139  
 
 140  
     protected abstract T loadEntity(Exchange exchange, Object key);
 141  
 
 142  
     protected Object getCorrelationKey(Exchange exchange) throws NoCorrelationKeyException {
 143  6
         Object value = correlationKeyExpression.evaluate(exchange);
 144  6
         if (value == null) {
 145  0
             throw new NoCorrelationKeyException(this, exchange);
 146  
         }
 147  6
         return value;
 148  
     }
 149  
 
 150  
     protected Object retryDueToDuplicate(TransactionStatus status) {
 151  0
         status.setRollbackOnly();
 152  0
         return null;
 153  
     }
 154  
 
 155  
     protected Object onError(TransactionStatus status, Throwable e) {
 156  0
         status.setRollbackOnly();
 157  0
         LOG.error("Caught: " + e, e);
 158  0
         throw new RuntimeCamelException(e);
 159  
     }
 160  
 }