camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r1377416 - in /camel/branches/camel-2.9.x: ./ components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/ components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/
Date Sun, 26 Aug 2012 10:49:42 GMT
Author: davsclaus
Date: Sun Aug 26 10:49:41 2012
New Revision: 1377416

URL: http://svn.apache.org/viewvc?rev=1377416&view=rev
Log:
CAMEL-5480: Quickfix engines is deferred to be started after CamelContext has been fully started.
To avoid any side effects with engines firing too fast.

Modified:
    camel/branches/camel-2.9.x/   (props changed)
    camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
    camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
    camel/branches/camel-2.9.x/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjComponentTest.java

Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
  Merged /camel/trunk:r1377413
  Merged /camel/branches/camel-2.10.x:r1377415

Propchange: camel/branches/camel-2.9.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.

Modified: camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java?rev=1377416&r1=1377415&r2=1377416&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
Sun Aug 26 10:49:41 2012
@@ -22,7 +22,9 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.camel.CamelContext;
 import org.apache.camel.Endpoint;
+import org.apache.camel.StartupListener;
 import org.apache.camel.impl.DefaultComponent;
 import org.apache.camel.util.UnsafeUriCharactersEncoder;
 import org.slf4j.Logger;
@@ -32,7 +34,7 @@ import quickfix.MessageFactory;
 import quickfix.MessageStoreFactory;
 import quickfix.SessionSettings;
 
-public class QuickfixjComponent extends DefaultComponent {
+public class QuickfixjComponent extends DefaultComponent implements StartupListener {
     private static final Logger LOG = LoggerFactory.getLogger(QuickfixjComponent.class);
 
     private final Object engineInstancesLock = new Object();
@@ -80,26 +82,23 @@ public class QuickfixjComponent extends 
     @Override
     protected void doStart() throws Exception {
         super.doStart();
-        synchronized (engineInstancesLock) {
-            for (QuickfixjEngine engine : engines.values()) {
-                startQuickfixjEngine(engine);
-            }
-        }
-    }
-
-    private void startQuickfixjEngine(QuickfixjEngine engine) throws Exception {
-        LOG.info("Starting QuickFIX/J engine: uri=", engine.getUri());
-        engine.start();
+        // we defer starting quickfix engines till the onCamelContextStarted callback
     }
 
     @Override
     protected void doStop() throws Exception {
-        super.doStop();
+        // stop engines when stopping component
         synchronized (engineInstancesLock) {
             for (QuickfixjEngine engine : engines.values()) {
                 engine.stop();
             }
         }
+        super.doStop();
+    }
+
+    private void startQuickfixjEngine(QuickfixjEngine engine) throws Exception {
+        LOG.info("Starting QuickFIX/J engine: {}", engine.getUri());
+        engine.start();
     }
 
     // Test Support
@@ -131,4 +130,13 @@ public class QuickfixjComponent extends 
         this.configurations = configurations;
     }
 
+    @Override
+    public void onCamelContextStarted(CamelContext camelContext, boolean alreadyStarted)
throws Exception {
+        // only start quickfix engines when CamelContext have finished starting
+        synchronized (engineInstancesLock) {
+            for (QuickfixjEngine engine : engines.values()) {
+                startQuickfixjEngine(engine);
+            }
+        }
+    }
 }

Modified: camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java?rev=1377416&r1=1377415&r2=1377416&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
Sun Aug 26 10:49:41 2012
@@ -26,6 +26,7 @@ import java.util.concurrent.CopyOnWriteA
 
 import javax.management.JMException;
 
+import org.apache.camel.support.ServiceSupport;
 import org.apache.camel.util.ObjectHelper;
 import org.quickfixj.jmx.JmxExporter;
 import org.slf4j.Logger;
@@ -76,7 +77,7 @@ import quickfix.UnsupportedMessageType;
  * The wrapper will create an initiator or acceptor or both depending on the
  * roles of sessions described in the settings file.
  */
-public class QuickfixjEngine {
+public class QuickfixjEngine extends ServiceSupport {
     public static final String DEFAULT_START_TIME = "00:00:00";
     public static final String DEFAULT_END_TIME = "00:00:00";
     public static final long DEFAULT_HEARTBTINT = 30;
@@ -93,10 +94,7 @@ public class QuickfixjEngine {
     private final LogFactory sessionLogFactory;
     private final MessageFactory messageFactory;
     private final MessageCorrelator messageCorrelator = new MessageCorrelator();
-    
-    private boolean started;
     private List<QuickfixjEventListener> eventListeners = new CopyOnWriteArrayList<QuickfixjEventListener>();
-
     private final String uri;
 
     public enum ThreadModel {
@@ -191,7 +189,8 @@ public class QuickfixjEngine {
         return new SessionSettings(inputStream);
     }
 
-    public void start() throws Exception {
+    @Override
+    protected void doStart() throws Exception {
         if (acceptor != null) {
             acceptor.start();
             if (jmxExporter != null) {
@@ -204,28 +203,25 @@ public class QuickfixjEngine {
                 jmxExporter.register(initiator);
             }
         }
-        started = true;
-    }
-
-    public void stop() throws Exception {
-        stop(forcedShutdown);
     }
 
-    public void stop(boolean force) throws Exception {
+    @Override
+    protected void doStop() throws Exception {
         if (acceptor != null) {
             acceptor.stop();
         }
         if (initiator != null) {
             initiator.stop();
         }
-        started = false;
     }
 
-    public boolean isStarted() {
-        return started;
+    @Override
+    protected void doShutdown() throws Exception {
+        // also clear event listeners
+        eventListeners.clear();
     }
-    
-    private Initiator createInitiator(Application application, SessionSettings settings,

+
+    private Initiator createInitiator(Application application, SessionSettings settings,
             MessageStoreFactory messageStoreFactory, LogFactory sessionLogFactory, 
             MessageFactory messageFactory, ThreadModel threadModel) throws ConfigError {
         

Modified: camel/branches/camel-2.9.x/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjComponentTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjComponentTest.java?rev=1377416&r1=1377415&r2=1377416&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjComponentTest.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjComponentTest.java
Sun Aug 26 10:49:41 2012
@@ -26,6 +26,7 @@ import java.net.URLClassLoader;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.camel.CamelContext;
 import org.apache.camel.Consumer;
 import org.apache.camel.Endpoint;
 import org.apache.camel.Exchange;
@@ -76,6 +77,7 @@ public class QuickfixjComponentTest {
     private SessionID sessionID;
     private SessionSettings settings;
     private QuickfixjComponent component;
+    private CamelContext camelContext;
     private MessageFactory engineMessageFactory;
     private MessageStoreFactory engineMessageStoreFactory;
     private LogFactory engineLogFactory;
@@ -118,9 +120,10 @@ public class QuickfixjComponentTest {
     }
     
     private void setUpComponent(boolean injectQfjPlugins) throws IOException, MalformedURLException,
NoSuchMethodException {
-        DefaultCamelContext camelContext = new DefaultCamelContext();
+        camelContext = new DefaultCamelContext();
         component = new QuickfixjComponent();
         component.setCamelContext(camelContext);
+        camelContext.addComponent("quickfix", component);
         
         if (injectQfjPlugins) {
             engineMessageFactory = new DefaultMessageFactory();
@@ -144,6 +147,9 @@ public class QuickfixjComponentTest {
         if (component != null) {
             component.stop();
         }
+        if (camelContext != null) {
+            camelContext.stop();
+        }
     }
 
     @Test
@@ -168,8 +174,9 @@ public class QuickfixjComponentTest {
         assertThat(component.getEngines().get(settingsFile.getName()), is(notNullValue()));
         assertThat(component.getEngines().get(settingsFile.getName()).isStarted(), is(false));
         assertThat(((QuickfixjEndpoint)e2).getSessionID(), is(sessionID));
-        
-        component.start();
+
+        // will start the component
+        camelContext.start();
         assertThat(component.getEngines().get(settingsFile.getName()).isStarted(), is(true));
         
         // Move these too an endpoint testcase if one exists
@@ -186,7 +193,8 @@ public class QuickfixjComponentTest {
 
         writeSettings();
 
-        component.start();
+        // will start the component
+        camelContext.start();
 
         Endpoint e1 = component.createEndpoint(getEndpointUri(settingsFile.getName(), null));
         assertThat(component.getEngines().size(), is(1));
@@ -229,9 +237,10 @@ public class QuickfixjComponentTest {
 
         // Endpoint automatically starts the consumer
         assertThat(((StatefulService)consumer).isStarted(), is(true));
-        
-        component.start();
-        
+
+        // will start the component
+        camelContext.start();
+
         assertTrue("Session not created", latch.await(5000, TimeUnit.MILLISECONDS));
         
         component.stop();
@@ -283,8 +292,9 @@ public class QuickfixjComponentTest {
         });
         ServiceHelper.startService(consumer);
 
-        component.start();
-        
+        // will start the component
+        camelContext.start();
+
         assertTrue("Session not created", logonLatch.await(5000, TimeUnit.MILLISECONDS));
        
         Endpoint producerEndpoint = component.createEndpoint(getEndpointUri(settingsFile.getName(),
acceptorSessionID));



Mime
View raw message