camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r950375 - in /camel/trunk: camel-core/src/main/java/org/apache/camel/component/file/strategy/ camel-core/src/main/java/org/apache/camel/impl/ camel-core/src/main/java/org/apache/camel/management/mbean/ camel-core/src/main/java/org/apache/ca...
Date Wed, 02 Jun 2010 05:37:59 GMT
Author: davsclaus
Date: Wed Jun  2 05:37:59 2010
New Revision: 950375

URL: http://svn.apache.org/viewvc?rev=950375&view=rev
Log:
CAMEL-2736: Polished code a bit about the warm up the producer for the SendProcessor.

Modified:
    camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileProcessStrategyFactory.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedProcessor.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedSendProcessor.java
    camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
    camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedSendProcessorTest.java
    camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpComponent.java

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileProcessStrategyFactory.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileProcessStrategyFactory.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileProcessStrategyFactory.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/strategy/FileProcessStrategyFactory.java
Wed Jun  2 05:37:59 2010
@@ -93,7 +93,7 @@ public final class FileProcessStrategyFa
             return strategy;
         }
 
-        // no explicit stategy set then fallback to readLock option
+        // no explicit strategy set then fallback to readLock option
         String readLock = (String) params.get("readLock");
         if (ObjectHelper.isNotEmpty(readLock)) {
             if ("none".equals(readLock) || "false".equals(readLock)) {

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ProducerCache.java Wed Jun
 2 05:37:59 2010
@@ -122,6 +122,21 @@ public class ProducerCache extends Servi
     }
 
     /**
+     * Starts the {@link Producer} to be used for sending to the given endpoint
+     * <p/>
+     * This can be used to early start the {@link Producer} to ensure it can be created,
+     * such as when Camel is started. This allows to fail fast in case the {@link Producer}
+     * could not be started.
+     *
+     * @param endpoint the endpoint to send the exchange to
+     * @throws Exception is thrown if failed to create or start the {@link Producer}
+     */
+    public void startProducer(Endpoint endpoint) throws Exception {
+        Producer producer = acquireProducer(endpoint);
+        releaseProducer(endpoint, producer);
+    }
+
+    /**
      * Sends an exchange to an endpoint using a supplied
      * {@link Processor} to populate the exchange
      *
@@ -238,7 +253,7 @@ public class ProducerCache extends Servi
         });
     }
 
-    public synchronized Producer doGetProducer(Endpoint endpoint, boolean pooled) {
+    protected synchronized Producer doGetProducer(Endpoint endpoint, boolean pooled) {
         String key = endpoint.getEndpointUri();
         Producer answer = producers.get(key);
         if (pooled && answer == null) {

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedProcessor.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedProcessor.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedProcessor.java
Wed Jun  2 05:37:59 2010
@@ -23,7 +23,9 @@ import org.apache.camel.Route;
 import org.apache.camel.ServiceStatus;
 import org.apache.camel.impl.ServiceSupport;
 import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.util.ServiceHelper;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
 
 /**
@@ -102,4 +104,14 @@ public class ManagedProcessor extends Ma
         return id;
     }
 
+    @ManagedOperation(description = "Start Processor")
+    public void start() throws Exception {
+        ServiceHelper.startService(getProcessor());
+    }
+
+    @ManagedOperation(description = "Stop Processor")
+    public void stop() throws Exception {
+        ServiceHelper.stopService(getProcessor());
+    }
+
 }

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedSendProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedSendProcessor.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedSendProcessor.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedSendProcessor.java
Wed Jun  2 05:37:59 2010
@@ -46,6 +46,12 @@ public class ManagedSendProcessor extend
         return processor.getDestination().getEndpointUri();
     }
 
+    @ManagedAttribute(description = "Destination as Endpoint Uri")
+    public void setDestination(String uri) {
+        Endpoint endpoint = getContext().getEndpoint(uri);
+        processor.setDestination(endpoint);
+    }
+
     @ManagedAttribute(description = "Message Exchange Pattern")
     public String getMessageExchangePattern() {
         if (processor.getPattern() != null) {
@@ -55,9 +61,4 @@ public class ManagedSendProcessor extend
         }
     }
 
-    @ManagedOperation(description = "Change Destination Endpoint Uri")
-    public void changeDestination(String uri) throws Exception {
-        Endpoint endpoint = getContext().getEndpoint(uri);
-        processor.setDestination(endpoint);
-    }
 }

Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java Wed
Jun  2 05:37:59 2010
@@ -42,7 +42,6 @@ public class SendProcessor extends Servi
     protected ProducerCache producerCache;
     protected Endpoint destination;
     protected ExchangePattern pattern;
-    private boolean init;
 
     public SendProcessor(Endpoint destination) {
         ObjectHelper.notNull(destination, "destination");
@@ -61,9 +60,8 @@ public class SendProcessor extends Servi
         return "sendTo(" + destination + (pattern != null ? " " + pattern : "") + ")";
     }
 
-    public synchronized void setDestination(Endpoint destination) {
+    public void setDestination(Endpoint destination) {
         this.destination = destination;
-        this.init = false;
     }
 
     public String getTraceLabel() {
@@ -71,7 +69,6 @@ public class SendProcessor extends Servi
     }
 
     public void process(final Exchange exchange) throws Exception {
-        
         doProcess(exchange);
     }
 
@@ -121,23 +118,18 @@ public class SendProcessor extends Servi
             camelContext.addService(producerCache);
         }
         ServiceHelper.startService(producerCache);
+
         // the destination could since have been intercepted by a interceptSendToEndpoint
so we got to
         // init this before we can use the destination
-        if (!init) {
-            init = true;
-            Endpoint lookup = camelContext.hasEndpoint(destination.getEndpointKey());
-            if (lookup instanceof InterceptSendToEndpoint) {
-                if (LOG.isDebugEnabled()) {
-                    LOG.debug("SendTo is intercepted using a interceptSendToEndpoint: " +
lookup.getEndpointUri());
-                }
-                destination = lookup;
+        Endpoint lookup = camelContext.hasEndpoint(destination.getEndpointKey());
+        if (lookup instanceof InterceptSendToEndpoint) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("SendTo is intercepted using a interceptSendToEndpoint: " + lookup.getEndpointUri());
             }
+            destination = lookup;
         }
-        // get the producer when the send processor is starting
-        Producer producer = producerCache.doGetProducer(destination, true);
-        if (producer == null) {            
-            throw new IllegalStateException("No producer, this processor has not been started:
" + this);
-        }
+        // warm up the producer by starting it so we can fail fast if there was a problem
+        producerCache.startProducer(destination);
     }
 
     protected void doStop() throws Exception {

Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedSendProcessorTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedSendProcessorTest.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedSendProcessorTest.java
(original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedSendProcessorTest.java
Wed Jun  2 05:37:59 2010
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.management;
 
+import javax.management.Attribute;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
@@ -76,8 +77,14 @@ public class ManagedSendProcessorTest ex
         String pattern = (String) mbeanServer.getAttribute(on, "MessageExchangePattern");
         assertNull(pattern);
 
+        // we must stop it to change the destination
+        mbeanServer.invoke(on, "stop", null, null);
+
         // send it somewhere else
-        mbeanServer.invoke(on, "changeDestination", new Object[]{"direct:foo"}, new String[]{"java.lang.String"});
+        mbeanServer.setAttribute(on, new Attribute("Destination", "direct:foo"));
+
+        // start it
+        mbeanServer.invoke(on, "start", null, null);
 
         // prepare mocks
         result.reset();

Modified: camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpComponent.java?rev=950375&r1=950374&r2=950375&view=diff
==============================================================================
--- camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpComponent.java
(original)
+++ camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpComponent.java
Wed Jun  2 05:37:59 2010
@@ -55,9 +55,6 @@ public class FtpComponent extends Remote
      * get the base uri part before the options as they can be non URI valid such as the
expression using $ chars
      * and the URI constructor will regard $ as an illegal character and we dont want to
enforce end users to
      * to escape the $ for the expression (file language)
-     * 
-     * @param uri
-     * @return baseUri
      */
     protected String getBaseUri(String uri) {
         String baseUri = uri;



Mime
View raw message