CAMEL-7118: Hazelcast-based Recoverable Aggregation Repository

camel-core/pom.xml

@@ -137,7 +137,6 @@
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
-      <version>${junit-version}</version>
       <scope>test</scope>
     </dependency>
 

camel-core/src/main/java/org/apache/camel/component/file/FileEndpoint.java

@@ -41,6 +41,8 @@ public class FileEndpoint extends GenericFileEndpoint<File> {
     @UriParam
     private boolean copyAndDeleteOnRenameFail = true;
     @UriParam
+    private boolean renameUsingCopy;
+    @UriParam
     private boolean forceWrites = true;
 
     public FileEndpoint() {
@@ -174,6 +176,14 @@ public void setCopyAndDeleteOnRenameFail(boolean copyAndDeleteOnRenameFail) {
         this.copyAndDeleteOnRenameFail = copyAndDeleteOnRenameFail;
     }
 
+    public boolean isRenameUsingCopy() {
+        return renameUsingCopy;
+    }
+
+    public void setRenameUsingCopy(boolean renameUsingCopy) {
+        this.renameUsingCopy = renameUsingCopy;
+    }
+
     public boolean isForceWrites() {
         return forceWrites;
     }

camel-core/src/main/java/org/apache/camel/component/file/FileOperations.java

@@ -64,13 +64,20 @@ public boolean deleteFile(String name) throws GenericFileOperationFailedExceptio
     }
 
     public boolean renameFile(String from, String to) throws GenericFileOperationFailedException {
+        boolean renamed = false;
         File file = new File(from);
         File target = new File(to);
         try {
-            return FileUtil.renameFile(file, target, endpoint.isCopyAndDeleteOnRenameFail());
+            if (endpoint.isRenameUsingCopy()) {
+                renamed = FileUtil.renameFileUsingCopy(file, target);
+            } else {
+                renamed = FileUtil.renameFile(file, target, endpoint.isCopyAndDeleteOnRenameFail());
+            }
         } catch (IOException e) {
             throw new GenericFileOperationFailedException("Error renaming file from " + from + " to " + to, e);
         }
+
+        return renamed;
     }
 
     public boolean existsFile(String name) throws GenericFileOperationFailedException {

camel-core/src/main/java/org/apache/camel/component/log/LogComponent.java

@@ -22,13 +22,11 @@
 import org.apache.camel.Endpoint;
 import org.apache.camel.LoggingLevel;
 import org.apache.camel.Processor;
-import org.apache.camel.ResolveEndpointFailedException;
 import org.apache.camel.impl.UriEndpointComponent;
 import org.apache.camel.processor.CamelLogProcessor;
 import org.apache.camel.processor.DefaultExchangeFormatter;
 import org.apache.camel.processor.ThroughputLogger;
 import org.apache.camel.spi.ExchangeFormatter;
-import org.apache.camel.util.CamelContextHelper;
 import org.apache.camel.util.CamelLogger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -57,6 +55,7 @@ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Obje
             Map<String, Logger> availableLoggers = getCamelContext().getRegistry().findByTypeWithName(Logger.class);
             if (availableLoggers.size() == 1) {
                 providedLogger = availableLoggers.values().iterator().next();
+                LOG.info("Using custom Logger: {}", providedLogger);
             } else if (availableLoggers.size() > 1) {
                 LOG.info("More than one {} instance found in the registry. Falling back to creating logger from URI {}.", Logger.class.getName(), uri);
             }

camel-core/src/main/java/org/apache/camel/impl/DefaultEndpoint.java

@@ -136,7 +136,13 @@ public boolean equals(Object object) {
 
     @Override
     public String toString() {
-        return String.format("Endpoint[%s]", URISupport.sanitizeUri(getEndpointUri()));
+        String value = null;
+        try {
+            value = getEndpointUri();
+        } catch (RuntimeException e) {
+            // ignore any exception and use null for building the string value
+        }
+        return String.format("Endpoint[%s]", URISupport.sanitizeUri(value));
     }
 
     /**

camel-core/src/main/java/org/apache/camel/model/LoadBalanceDefinition.java

@@ -31,13 +31,15 @@
 
 import org.apache.camel.Expression;
 import org.apache.camel.Processor;
+import org.apache.camel.model.loadbalancer.CircuitBreakerLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.CustomLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.FailoverLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.RandomLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.RoundRobinLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.StickyLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.TopicLoadBalancerDefinition;
 import org.apache.camel.model.loadbalancer.WeightedLoadBalancerDefinition;
+import org.apache.camel.processor.loadbalancer.CircuitBreakerLoadBalancer;
 import org.apache.camel.processor.loadbalancer.FailOverLoadBalancer;
 import org.apache.camel.processor.loadbalancer.LoadBalancer;
 import org.apache.camel.processor.loadbalancer.RandomLoadBalancer;
@@ -66,7 +68,8 @@ public class LoadBalanceDefinition extends ProcessorDefinition<LoadBalanceDefini
             @XmlElement(required = false, name = "roundRobin", type = RoundRobinLoadBalancerDefinition.class),
             @XmlElement(required = false, name = "sticky", type = StickyLoadBalancerDefinition.class),
             @XmlElement(required = false, name = "topic", type = TopicLoadBalancerDefinition.class),
-            @XmlElement(required = false, name = "weighted", type = WeightedLoadBalancerDefinition.class)}
+            @XmlElement(required = false, name = "weighted", type = WeightedLoadBalancerDefinition.class),
+            @XmlElement(required = false, name = "circuitBreaker", type = CircuitBreakerLoadBalancerDefinition.class)}
     )
     private LoadBalancerDefinition loadBalancerType;
     @XmlElementRef
@@ -213,6 +216,23 @@ public LoadBalanceDefinition failover(int maximumFailoverAttempts, boolean inher
     public LoadBalanceDefinition weighted(boolean roundRobin, String distributionRatio) {
         return weighted(roundRobin, distributionRatio, ",");
     }
+
+    /**
+     * Uses circuitBreaker load balancer
+     *
+     * @param threshold         number of errors before failure.
+     * @param halfOpenAfter     time interval in milliseconds for half open state.
+     * @param exceptions        exception classes which we want to break if one of them was thrown
+     * @return the builder
+     */
+    public LoadBalanceDefinition circuitBreaker(int threshold, long halfOpenAfter, Class<?>... exceptions) {
+        CircuitBreakerLoadBalancer breakerLoadBalancer = new CircuitBreakerLoadBalancer(Arrays.asList(exceptions));
+        breakerLoadBalancer.setThreshold(threshold);
+        breakerLoadBalancer.setHalfOpenAfter(halfOpenAfter);
+
+        setLoadBalancerType(new LoadBalancerDefinition(breakerLoadBalancer));
+        return this;
+    }
 
     /**
      * Uses weighted load balancer

camel-core/src/main/java/org/apache/camel/model/LogDefinition.java

@@ -16,18 +16,24 @@
  */
 package org.apache.camel.model;
 
+import java.util.Map;
+
 import javax.xml.bind.annotation.XmlAccessType;
 import javax.xml.bind.annotation.XmlAccessorType;
 import javax.xml.bind.annotation.XmlAttribute;
 import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.XmlTransient;
 
 import org.apache.camel.Expression;
 import org.apache.camel.LoggingLevel;
 import org.apache.camel.Processor;
 import org.apache.camel.processor.LogProcessor;
 import org.apache.camel.spi.RouteContext;
+import org.apache.camel.util.CamelContextHelper;
 import org.apache.camel.util.CamelLogger;
 import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Represents an XML <log/> element
@@ -37,6 +43,8 @@
 @XmlRootElement(name = "log")
 @XmlAccessorType(XmlAccessType.FIELD)
 public class LogDefinition extends NoOutputDefinition<LogDefinition> {
+    @XmlTransient
+    private static final Logger LOG = LoggerFactory.getLogger(LogDefinition.class);
     @XmlAttribute(required = true)
     private String message;
     @XmlAttribute
@@ -45,6 +53,10 @@ public class LogDefinition extends NoOutputDefinition<LogDefinition> {
     private String logName;
     @XmlAttribute
     private String marker;
+    @XmlAttribute
+    private String loggerRef;
+    @XmlTransient
+    private Logger logger;
 
     public LogDefinition() {
     }
@@ -75,15 +87,40 @@ public Processor createProcessor(RouteContext routeContext) throws Exception {
         // use simple language for the message string to give it more power
         Expression exp = routeContext.getCamelContext().resolveLanguage("simple").createExpression(message);
 
-        String name = getLogName();
-        if (name == null) {
-            name = routeContext.getRoute().getId();
+        // get logger explicitely set in the definition
+        Logger logger = this.getLogger();
+
+        // get logger which may be set in XML definition
+        if (logger == null && ObjectHelper.isNotEmpty(loggerRef)) {
+            logger = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), loggerRef, Logger.class);
+        }
+
+        if (logger == null) {
+            // first - try to lookup single instance in the registry, just like LogComponent
+            Map<String, Logger> availableLoggers = routeContext.lookupByType(Logger.class);
+            if (availableLoggers.size() == 1) {
+                logger = availableLoggers.values().iterator().next();
+                LOG.info("Using custom Logger: {}", logger);
+            } else if (availableLoggers.size() > 1) {
+                // we should log about this somewhere...
+                LOG.info("More than one {} instance found in the registry. Falling back to create logger by name.", Logger.class.getName());
+            }
+        }
+
+        if (logger == null) {
+            String name = getLogName();
+            if (name == null) {
+                name = routeContext.getRoute().getId();
+                LOG.info("The LogName is null. Falling back to create logger by using the route id {}.", name);
+            }
+            logger = LoggerFactory.getLogger(name);
         }
+
         // should be INFO by default
         LoggingLevel level = getLoggingLevel() != null ? getLoggingLevel() : LoggingLevel.INFO;
-        CamelLogger logger = new CamelLogger(name, level, getMarker());
+        CamelLogger camelLogger = new CamelLogger(logger, level, getMarker());
 
-        return new LogProcessor(exp, logger);
+        return new LogProcessor(exp, camelLogger);
     }
 
     @Override
@@ -123,4 +160,20 @@ public String getMarker() {
     public void setMarker(String marker) {
         this.marker = marker;
     }
+
+    public String getLoggerRef() {
+        return loggerRef;
+    }
+
+    public void setLoggerRef(String loggerRef) {
+        this.loggerRef = loggerRef;
+    }
+
+    public Logger getLogger() {
+        return logger;
+    }
+
+    public void setLogger(Logger logger) {
+        this.logger = logger;
+    }
 }

camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java

@@ -1539,6 +1539,23 @@ public Type log(LoggingLevel loggingLevel, String logName, String message) {
         return (Type) this;
     }
 
+    /**
+     * Creates a log message to be logged at the given level using provided logger.
+     *
+     * @param loggingLevel the logging level to use
+     * @param logger the logger to use
+     * @param message the log message, (you can use {@link org.apache.camel.language.simple.SimpleLanguage} syntax)
+     * @return the builder
+     */
+    @SuppressWarnings("unchecked")
+    public Type log(LoggingLevel loggingLevel, Logger logger, String message) {
+        LogDefinition answer = new LogDefinition(message);
+        answer.setLoggingLevel(loggingLevel);
+        answer.setLogger(logger);
+        addOutput(answer);
+        return (Type) this;
+    }
+
     /**
      * Creates a log message to be logged at the given level and name.
      *
@@ -1559,6 +1576,26 @@ public Type log(LoggingLevel loggingLevel, String logName, String marker, String
         return (Type) this;
     }
 
+    /**
+     * Creates a log message to be logged at the given level using provided logger.
+     *
+     *
+     * @param loggingLevel the logging level to use
+     * @param logger the logger to use
+     * @param marker  log marker name
+     * @param message the log message, (you can use {@link org.apache.camel.language.simple.SimpleLanguage} syntax)
+     * @return the builder
+     */
+    @SuppressWarnings("unchecked")
+    public Type log(LoggingLevel loggingLevel, Logger logger, String marker, String message) {
+        LogDefinition answer = new LogDefinition(message);
+        answer.setLoggingLevel(loggingLevel);
+        answer.setLogger(logger);
+        answer.setMarker(marker);
+        addOutput(answer);
+        return (Type) this;
+    }
+
     /**
      * <a href="http://camel.apache.org/content-based-router.html">Content Based Router EIP:</a>
      * Creates a choice of one or more predicates with an otherwise clause

camel-core/src/main/java/org/apache/camel/model/RouteContextRefDefinition.java

@@ -50,8 +50,7 @@ public String getRef() {
     public void setRef(String ref) {
         this.ref = ref;
     }
-
-    @SuppressWarnings({"unchecked", "rawtypes"})
+
     public List<RouteDefinition> lookupRoutes(CamelContext camelContext) {
         return RouteContextRefDefinitionHelper.lookupRoutes(camelContext, ref);
     }