camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hadr...@apache.org
Subject svn commit: r1130604 - in /camel/trunk/components/camel-quickfix: ./ src/main/java/org/apache/camel/component/quickfixj/ src/test/java/org/apache/camel/component/quickfixj/ src/test/java/org/apache/camel/component/quickfixj/examples/ src/test/java/org/...
Date Thu, 02 Jun 2011 15:26:13 GMT
Author: hadrian
Date: Thu Jun  2 15:26:12 2011
New Revision: 1130604

URL: http://svn.apache.org/viewvc?rev=1130604&view=rev
Log:
CAMEL-4038. Original contributed patch applied with thanks to Steve. Checkstyle fixes to follow

Added:
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java
Modified:
    camel/trunk/components/camel-quickfix/pom.xml
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
    camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConvertersTest.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/TestSupport.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/trading/TradeExecutorExample.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java
    camel/trunk/components/camel-quickfix/src/test/resources/examples/inprocess.cfg

Modified: camel/trunk/components/camel-quickfix/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/pom.xml?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/pom.xml (original)
+++ camel/trunk/components/camel-quickfix/pom.xml Thu Jun  2 15:26:12 2011
@@ -67,12 +67,16 @@
             <artifactId>slf4j-api</artifactId>
         </dependency>
 
+        <dependency>
+			<groupId>org.apache.camel</groupId>
+			<artifactId>camel-jetty</artifactId>
+            <scope>test</scope>
+        </dependency>
 		<dependency>
 			<groupId>org.slf4j</groupId>
 			<artifactId>slf4j-log4j12</artifactId>
 			<scope>test</scope>
 		</dependency>
-
 	    <dependency>
 	      <groupId>org.springframework</groupId>
 	      <artifactId>spring-test</artifactId>

Added: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java?rev=1130604&view=auto
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java (added)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java Thu Jun  2 15:26:12 2011
@@ -0,0 +1,109 @@
+/**
+ * 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.quickfixj;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangeTimedOutException;
+
+import quickfix.Message;
+import quickfix.SessionID;
+
+public class MessageCorrelator implements QuickfixjEventListener {
+	public static final long DEFAULT_CORRELATION_TIMEOUT = 1000L;
+	private final CopyOnWriteArrayList<MessageCorrelationRule> rules = new CopyOnWriteArrayList<MessageCorrelationRule>();
+
+	public Callable<Message> getReply(SessionID sessionID, Exchange exchange)
+			throws InterruptedException, ExchangeTimedOutException {
+
+		MessagePredicate messageCriteria = (MessagePredicate) exchange
+				.getProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY);
+		final MessageCorrelationRule correlationRule = new MessageCorrelationRule(
+				exchange, sessionID, messageCriteria);
+		
+		final long timeout = exchange.getProperty(
+				QuickfixjProducer.CORRELATION_TIMEOUT_KEY,
+				DEFAULT_CORRELATION_TIMEOUT, Long.class);
+
+		rules.add(correlationRule);
+
+		return new Callable<Message>() {
+			@Override
+			public Message call() throws Exception {
+				if (!correlationRule.getLatch().await(timeout,
+						TimeUnit.MILLISECONDS)) {
+					throw new ExchangeTimedOutException(
+							correlationRule.getExchange(), timeout);
+				}
+				return correlationRule.getReplyMessage();
+			}
+		};
+	}
+
+	@Override
+	public void onEvent(QuickfixjEventCategory eventCategory,
+			SessionID sessionID, Message message) throws Exception {
+		if (message != null) {
+			for (MessageCorrelationRule rule : rules) {
+				if (rule.getMessageCriteria().evaluate(message)) {
+					rule.setReplyMessage(message);
+					rules.remove(rule);
+					rule.getLatch().countDown();
+				}
+			}
+		}
+	}
+
+	private class MessageCorrelationRule {
+		private final Exchange exchange;
+		private final CountDownLatch latch = new CountDownLatch(1);
+		private final MessagePredicate messageCriteria;
+
+		private Message replyMessage;
+
+		public MessageCorrelationRule(Exchange exchange, SessionID sessionID,
+				MessagePredicate messageCriteria) {
+			this.exchange = exchange;
+			this.messageCriteria = messageCriteria;
+		}
+
+		public void setReplyMessage(Message message) {
+			this.replyMessage = message;
+		}
+
+		public Message getReplyMessage() {
+			return replyMessage;
+		}
+
+		public CountDownLatch getLatch() {
+			return latch;
+		}
+
+		public Exchange getExchange() {
+			return exchange;
+		}
+
+		public MessagePredicate getMessageCriteria() {
+			return messageCriteria;
+		}
+	}
+
+}

Added: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java?rev=1130604&view=auto
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java (added)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java Thu Jun  2 15:26:12 2011
@@ -0,0 +1,68 @@
+package org.apache.camel.component.quickfixj;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import quickfix.Field;
+import quickfix.FieldMap;
+import quickfix.FieldNotFound;
+import quickfix.Message;
+import quickfix.SessionID;
+import quickfix.field.MsgType;
+import quickfix.field.SenderCompID;
+import quickfix.field.TargetCompID;
+
+public class MessagePredicate {
+	private final List<Field<String>> headerCriteria = new ArrayList<Field<String>>();
+	private final List<Field<String>> bodyCriteria = new ArrayList<Field<String>>();
+	
+	public MessagePredicate(SessionID requestingSessionID, String msgType) {
+		// Reverse session ID for reply
+		// TODO may need to optionally include subID and locationID
+		addHeaderFieldIfPresent(SenderCompID.FIELD, requestingSessionID.getTargetCompID());
+		addHeaderFieldIfPresent(TargetCompID.FIELD, requestingSessionID.getSenderCompID());
+		withMessageType(msgType);
+	}
+	
+	private void addHeaderFieldIfPresent(int tag, String value) {
+		if (value != null && !"".equals(value)) {
+			withHeaderField(tag, value);
+		}
+	}
+
+	public boolean evaluate(Message message) {
+		return evaluate(message, bodyCriteria) && evaluate(message.getHeader(), headerCriteria);
+	}
+	
+	private boolean evaluate(FieldMap fieldMap, List<Field<String>> criteria) {
+		for (Field<String> c : criteria) {
+			String value = null;
+			try {
+				if (fieldMap.isSetField(c.getField())) {
+					value = fieldMap.getString(c.getField());
+				}
+			} catch (FieldNotFound e) {
+				// ignored, shouldn't happen
+			}
+			if (!c.getObject().equals(value)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	public MessagePredicate withField(int tag, String value) {
+		bodyCriteria.add(new Field<String>(tag, value));
+		return this;
+	}
+
+	public MessagePredicate withHeaderField(int tag, String value) {
+		headerCriteria.add(new Field<String>(tag, value));
+		return this;
+	}
+
+	private MessagePredicate withMessageType(String msgType) {
+		headerCriteria.add(new Field<String>(MsgType.FIELD, msgType));
+		return this;
+	}
+}

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjComponent.java Thu Jun  2 15:26:12 2011
@@ -22,6 +22,7 @@ import java.net.URISyntaxException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+
 import javax.management.JMException;
 
 import org.apache.camel.Endpoint;
@@ -29,6 +30,7 @@ import org.apache.camel.impl.DefaultComp
 import org.apache.camel.util.UnsafeUriCharactersEncoder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import quickfix.ConfigError;
 import quickfix.FieldConvertError;
 import quickfix.LogFactory;
@@ -72,7 +74,7 @@ public class QuickfixjComponent extends 
                     }
                 }
 
-                endpoint = new QuickfixjEndpoint(uri, getCamelContext());
+                endpoint = new QuickfixjEndpoint(engine, uri, getCamelContext());
                 engine.addEventListener(endpoint);
                 endpoints.put(uri, endpoint);
             }

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java Thu Jun  2 15:26:12 2011
@@ -18,12 +18,17 @@ package org.apache.camel.component.quick
 
 import org.apache.camel.Endpoint;
 import org.apache.camel.Exchange;
+import org.apache.camel.Message;
 import org.apache.camel.Processor;
 import org.apache.camel.impl.DefaultConsumer;
 
+import quickfix.MessageUtils;
+import quickfix.Session;
+import quickfix.SessionID;
+
 public class QuickfixjConsumer extends DefaultConsumer {
 
-    public QuickfixjConsumer(Endpoint endpoint, Processor processor) {
+	public QuickfixjConsumer(Endpoint endpoint, Processor processor) {
         super(endpoint, processor);
     }
 
@@ -31,9 +36,38 @@ public class QuickfixjConsumer extends D
         if (isStarted()) {
             try {
                 getProcessor().process(exchange);
+                if (exchange.getPattern().isOutCapable() && exchange.hasOut()) {
+                	sendOutMessage(exchange);
+                }
             } catch (Exception e) {
                 exchange.setException(e);
             }
         }
     }
+
+	private void sendOutMessage(Exchange exchange) {
+		try {
+			Message camelMessage = exchange.getOut();
+			quickfix.Message quickfixjMessage = camelMessage.getBody(quickfix.Message.class);
+	 
+			if (log.isDebugEnabled()) {
+			    log.debug("Sending FIX message reply: " + quickfixjMessage.toString());
+			}
+			
+			SessionID messageSessionID = MessageUtils.getReverseSessionID(exchange.getIn().getBody(quickfix.Message.class));
+			
+			Session session = getSession(messageSessionID);
+			if (session == null) {
+			   throw new IllegalStateException("Unknown session: " + messageSessionID);
+			}
+			
+			session.send(quickfixjMessage);
+		} catch (Exception e) {
+			exchange.setException(e);
+		}
+	}
+
+	Session getSession(SessionID messageSessionID) {
+		return Session.lookupSession(messageSessionID);
+	}
 }

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEndpoint.java Thu Jun  2 15:26:12 2011
@@ -25,73 +25,124 @@ import org.apache.camel.Exchange;
 import org.apache.camel.MultipleConsumersSupport;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
+import org.apache.camel.ResolveEndpointFailedException;
 import org.apache.camel.component.quickfixj.converter.QuickfixjConverters;
 import org.apache.camel.impl.DefaultEndpoint;
-import org.apache.camel.util.ServiceHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import quickfix.Message;
 import quickfix.SessionID;
 
-public class QuickfixjEndpoint extends DefaultEndpoint implements QuickfixjEventListener, MultipleConsumersSupport {
-    public static final String EVENT_CATEGORY_KEY = "EventCategory";
-    public static final String SESSION_ID_KEY = "SessionID";
-    public static final String MESSAGE_TYPE_KEY = "MessageType";
-    public static final String DATA_DICTIONARY_KEY = "DataDictionary";
-    
-    private static final Logger LOG = LoggerFactory.getLogger(QuickfixjEndpoint.class);
-    
-    private SessionID sessionID;
-    private final List<QuickfixjConsumer> consumers = new CopyOnWriteArrayList<QuickfixjConsumer>();
-    
-    public QuickfixjEndpoint(String uri, CamelContext context) {
-        super(uri, context);
-    }
-
-    protected SessionID getSessionID() {
-        return sessionID;
-    }
-
-    public void setSessionID(SessionID sessionID) {
-        this.sessionID = sessionID;
-    }
-    
-    public Consumer createConsumer(Processor processor) throws Exception {
-        LOG.info("Creating QuickFIX/J consumer: " + (sessionID != null ? sessionID : "No Session"));
-        QuickfixjConsumer consumer = new QuickfixjConsumer(this, processor);
-        consumers.add(consumer);
-        return consumer;
-    }
-
-    public Producer createProducer() throws Exception {
-        LOG.info("Creating QuickFIX/J producer: " + (sessionID != null ? sessionID : "No Session"));
-        return new QuickfixjProducer(this);
-    }
-
-    public boolean isSingleton() {
-        return true;
-    }
-
-    public void onEvent(QuickfixjEventCategory eventCategory, SessionID sessionID, Message message) throws Exception {
-        if (this.sessionID == null || this.sessionID.equals(sessionID)) {
-            for (QuickfixjConsumer consumer : consumers) {
-                Exchange exchange = QuickfixjConverters.toExchange(this, sessionID, message, eventCategory);
-                consumer.onExchange(exchange);
-                if (exchange.getException() != null) {
-                    throw exchange.getException();
-                }
-            }
-        }
-    }
-    
-    public boolean isMultipleConsumersSupported() {
-        return true;
-    }
-
-    @Override
-    protected void doStop() throws Exception {
-        // clear list of consumers
-        consumers.clear();
-    }
+public class QuickfixjEndpoint extends DefaultEndpoint implements
+		QuickfixjEventListener, MultipleConsumersSupport {
+	public static final String EVENT_CATEGORY_KEY = "EventCategory";
+	public static final String SESSION_ID_KEY = "SessionID";
+	public static final String MESSAGE_TYPE_KEY = "MessageType";
+	public static final String DATA_DICTIONARY_KEY = "DataDictionary";
+
+	private static final Logger LOG = LoggerFactory
+			.getLogger(QuickfixjEndpoint.class);
+
+	private SessionID sessionID;
+	private final List<QuickfixjConsumer> consumers = new CopyOnWriteArrayList<QuickfixjConsumer>();
+	private final QuickfixjEngine engine;
+
+	public QuickfixjEndpoint(QuickfixjEngine engine, String uri, CamelContext context) {
+		super(uri, context);
+		this.engine = engine;
+	}
+
+	public SessionID getSessionID() {
+		return sessionID;
+	}
+
+	public void setSessionID(SessionID sessionID) {
+		this.sessionID = sessionID;
+	}
+
+	public Consumer createConsumer(Processor processor) throws Exception {
+		LOG.info("Creating QuickFIX/J consumer: "
+				+ (sessionID != null ? sessionID : "No Session")
+				+ ", ExchangePattern=" + getExchangePattern());
+		QuickfixjConsumer consumer = new QuickfixjConsumer(this, processor);
+		consumers.add(consumer);
+		return consumer;
+	}
+
+	public Producer createProducer() throws Exception {
+		LOG.info("Creating QuickFIX/J producer: "
+				+ (sessionID != null ? sessionID : "No Session"));
+		if (isWildcarded()) {
+			throw new ResolveEndpointFailedException(
+					"Cannot create consumer on wildcarded session identifier: "
+							+ sessionID);
+		}
+		return new QuickfixjProducer(this);
+	}
+
+	public boolean isSingleton() {
+		return true;
+	}
+
+	public void onEvent(QuickfixjEventCategory eventCategory,
+			SessionID sessionID, Message message) throws Exception {
+		if (this.sessionID == null || isMatching(sessionID)) {
+			for (QuickfixjConsumer consumer : consumers) {
+				Exchange exchange = QuickfixjConverters.toExchange(this,
+						sessionID, message, eventCategory);
+				consumer.onExchange(exchange);
+				if (exchange.getException() != null) {
+					throw exchange.getException();
+				}
+			}
+		}
+	}
+
+	private boolean isMatching(SessionID sessionID) {
+		return this.sessionID.equals(sessionID)
+				|| (isMatching(this.sessionID.getBeginString(),
+						sessionID.getBeginString())
+						&& isMatching(this.sessionID.getSenderCompID(),
+								sessionID.getSenderCompID())
+						&& isMatching(this.sessionID.getSenderSubID(),
+								sessionID.getSenderSubID())
+						&& isMatching(this.sessionID.getSenderLocationID(),
+								sessionID.getSenderLocationID())
+						&& isMatching(this.sessionID.getTargetCompID(),
+								sessionID.getTargetCompID())
+						&& isMatching(this.sessionID.getTargetSubID(),
+								sessionID.getTargetSubID()) && isMatching(
+						this.sessionID.getTargetLocationID(),
+						sessionID.getTargetLocationID()));
+	}
+
+	private boolean isMatching(String s1, String s2) {
+		return s1.equals("") || s1.equals("*") || s1.equals(s2);
+	}
+
+	private boolean isWildcarded() {
+		return sessionID != null
+				&& (sessionID.getBeginString().equals("*")
+						|| sessionID.getSenderCompID().equals("*")
+						|| sessionID.getSenderSubID().equals("*")
+						|| sessionID.getSenderLocationID().equals("*")
+						|| sessionID.getTargetCompID().equals("*")
+						|| sessionID.getTargetSubID().equals("*") || sessionID
+						.getTargetLocationID().equals("*"));
+	}
+
+	public boolean isMultipleConsumersSupported() {
+		return true;
+	}
+
+	public QuickfixjEngine getEngine() {
+		return engine;
+	}
+	
+	@Override
+	protected void doStop() throws Exception {
+		// clear list of consumers
+		consumers.clear();
+	}
 }

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEngine.java Thu Jun  2 15:26:12 2011
@@ -92,7 +92,8 @@ public class QuickfixjEngine {
     private final MessageStoreFactory messageStoreFactory;
     private final LogFactory sessionLogFactory;
     private final MessageFactory messageFactory;
-
+    private final MessageCorrelator messageCorrelator = new MessageCorrelator();
+    
     private boolean started;
     private List<QuickfixjEventListener> eventListeners = new CopyOnWriteArrayList<QuickfixjEventListener>();
 
@@ -119,6 +120,8 @@ public class QuickfixjEngine {
             MessageStoreFactory messageStoreFactoryOverride, LogFactory sessionLogFactoryOverride,
             MessageFactory messageFactoryOverride) throws ConfigError, FieldConvertError, IOException, JMException {
 
+    	addEventListener(messageCorrelator);
+    	
         this.uri = uri;
         this.forcedShutdown = forcedShutdown;
         
@@ -463,6 +466,10 @@ public class QuickfixjEngine {
         return uri;
     }
 
+    public MessageCorrelator getMessageCorrelator() {
+		return messageCorrelator;
+	}
+
     // For Testing
     Initiator getInitiator() {
         return initiator;

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjProducer.java Thu Jun  2 15:26:12 2011
@@ -16,6 +16,8 @@
  */
 package org.apache.camel.component.quickfixj;
 
+import java.util.concurrent.Callable;
+
 import org.apache.camel.Endpoint;
 import org.apache.camel.Exchange;
 import org.apache.camel.impl.DefaultProducer;
@@ -26,31 +28,61 @@ import quickfix.Session;
 import quickfix.SessionID;
 
 public class QuickfixjProducer extends DefaultProducer {
-    private final SessionID sessionID;
-
+    public static final String CORRELATION_TIMEOUT_KEY = "CorrelationTimeout";
+    public static final String CORRELATION_CRITERIA_KEY = "CorrelationCriteria";
+	
+	private final SessionID sessionID;
+    
     public QuickfixjProducer(Endpoint endpoint) {
         super(endpoint);
         sessionID = ((QuickfixjEndpoint) getEndpoint()).getSessionID();
     }
 
     public void process(Exchange exchange) throws Exception {
-        Message message = exchange.getIn().getBody(Message.class);
-        if (log.isDebugEnabled()) {
-            log.debug("Sending FIX message: " + message.toString());
-        }
-        
-        SessionID messageSessionID = sessionID;
-        if (messageSessionID == null) {
-            messageSessionID = MessageUtils.getSessionID(message);
-        }
-        
-        Session session = Session.lookupSession(messageSessionID);
-        if (session == null) {
-            exchange.setException(new IllegalStateException("Unknown session: " + messageSessionID));
-            return;
-        }
-        
-        session.send(message);
+        sendMessage(exchange, exchange.getIn());
     }
 
+	void sendMessage(Exchange exchange, org.apache.camel.Message camelMessage)
+			throws InterruptedException {
+		try {
+			Message message = camelMessage.getBody(Message.class);
+			if (log.isDebugEnabled()) {
+				log.debug("Sending FIX message: " + message.toString());
+			}
+
+			SessionID messageSessionID = sessionID;
+			if (messageSessionID == null) {
+				messageSessionID = MessageUtils.getSessionID(message);
+			}
+
+			Session session = getSession(messageSessionID);
+			if (session == null) {
+				throw new IllegalStateException("Unknown session: "
+						+ messageSessionID);
+			}
+
+			Callable<Message> callable = null;
+
+			if (exchange.getPattern().isOutCapable()) {
+				QuickfixjEndpoint endpoint = (QuickfixjEndpoint) getEndpoint();
+				MessageCorrelator messageCorrelator = endpoint.getEngine()
+						.getMessageCorrelator();
+				callable = messageCorrelator.getReply(
+						endpoint.getSessionID(), exchange);
+			}
+
+			session.send(message);
+
+			if (callable != null) {
+				Message reply = callable.call();
+				exchange.getOut().setBody(reply);
+			}
+		} catch (Exception e) {
+			exchange.setException(e);
+		}
+	}
+
+	Session getSession(SessionID messageSessionID) {
+		return Session.lookupSession(messageSessionID);
+	}
 }

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConsumerTest.java Thu Jun  2 15:26:12 2011
@@ -28,21 +28,41 @@ import org.junit.Test;
 import org.mockito.Matchers;
 import org.mockito.Mockito;
 
+import quickfix.FixVersions;
+import quickfix.Message;
+import quickfix.MessageUtils;
+import quickfix.Session;
+import quickfix.SessionID;
+import quickfix.field.BeginString;
+import quickfix.field.SenderCompID;
+import quickfix.field.TargetCompID;
+
 public class QuickfixjConsumerTest {
     private Exchange mockExchange;
     private Processor mockProcessor;
     private Endpoint mockEndpoint;
-
+	private SessionID sessionID;
+	private Message inboundFixMessage;
+    
     @Before
     public void setUp() {
+
         mockExchange = Mockito.mock(Exchange.class);
         org.apache.camel.Message mockCamelMessage = Mockito.mock(org.apache.camel.Message.class);
         Mockito.when(mockExchange.getIn()).thenReturn(mockCamelMessage);
         
+        inboundFixMessage = new Message();
+        inboundFixMessage.getHeader().setString(BeginString.FIELD, FixVersions.BEGINSTRING_FIX44);
+        inboundFixMessage.getHeader().setString(SenderCompID.FIELD, "SENDER");
+        inboundFixMessage.getHeader().setString(TargetCompID.FIELD, "TARGET");
+        sessionID = MessageUtils.getSessionID(inboundFixMessage);
+
+		Mockito.when(mockCamelMessage.getBody(quickfix.Message.class)).thenReturn(inboundFixMessage);
+        
         mockProcessor = Mockito.mock(Processor.class);
         
         mockEndpoint = Mockito.mock(Endpoint.class);
-        Mockito.when(mockEndpoint.createExchange(ExchangePattern.InOnly)).thenReturn(mockExchange);        
+        Mockito.when(mockEndpoint.createExchange(ExchangePattern.InOnly)).thenReturn(mockExchange);  
     }
     
     @Test
@@ -67,8 +87,7 @@ public class QuickfixjConsumerTest {
     }
     
     @Test
-    public void setExceptionOnExchange() throws Exception {
-              
+    public void setExceptionOnExchange() throws Exception {            
         QuickfixjConsumer consumer = new QuickfixjConsumer(mockEndpoint, mockProcessor);
         consumer.start();
         
@@ -79,5 +98,45 @@ public class QuickfixjConsumerTest {
         consumer.onExchange(mockExchange);
         
         Mockito.verify(mockExchange).setException(exception);
-    }    
+    }
+    
+    @Test
+    public void setExceptionOnInOutExchange() throws Exception {            
+        org.apache.camel.Message mockCamelOutMessage = Mockito.mock(org.apache.camel.Message.class);
+        Mockito.when(mockExchange.getPattern()).thenReturn(ExchangePattern.InOut);
+        Mockito.when(mockExchange.hasOut()).thenReturn(true);
+        Mockito.when(mockExchange.getOut()).thenReturn(mockCamelOutMessage);
+        Mockito.when(mockCamelOutMessage.getBody(Message.class)).thenReturn(new Message());
+        
+        QuickfixjConsumer consumer = new QuickfixjConsumer(mockEndpoint, mockProcessor);
+        consumer.start();
+        
+        // Simulate a message from the FIX engine
+        consumer.onExchange(mockExchange);
+        
+        Mockito.verify(mockExchange).setException(Mockito.isA(IllegalStateException.class));
+    }
+
+    @Test
+    public void processInOutExchange() throws Exception {
+        org.apache.camel.Message mockCamelOutMessage = Mockito.mock(org.apache.camel.Message.class);
+        Mockito.when(mockExchange.hasOut()).thenReturn(true);
+        Mockito.when(mockExchange.getOut()).thenReturn(mockCamelOutMessage);
+        Message outboundFixMessage = new Message();
+		Mockito.when(mockCamelOutMessage.getBody(Message.class)).thenReturn(outboundFixMessage);
+        
+        QuickfixjConsumer consumer = Mockito.spy(new QuickfixjConsumer(mockEndpoint, mockProcessor));
+        Session mockSession = Mockito.spy(TestSupport.createSession(sessionID));
+		Mockito.doReturn(mockSession).when(consumer).getSession(MessageUtils.getReverseSessionID(inboundFixMessage));
+		Mockito.doReturn(true).when(mockSession).send(Mockito.isA(Message.class));
+        
+        consumer.start();
+        
+        Mockito.when(mockExchange.getPattern()).thenReturn(ExchangePattern.InOut);
+        
+        consumer.onExchange(mockExchange);
+        
+        Mockito.verify(mockExchange, Mockito.never()).setException(Mockito.isA(Exception.class));
+        Mockito.verify(mockSession).send(outboundFixMessage);
+    }
 }

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConvertersTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConvertersTest.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConvertersTest.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjConvertersTest.java Thu Jun  2 15:26:12 2011
@@ -98,7 +98,7 @@ public class QuickfixjConvertersTest {
     @Test
     public void convertToExchange() {
         SessionID sessionID = new SessionID("FIX.4.0", "FOO", "BAR");
-        QuickfixjEndpoint endpoint = new QuickfixjEndpoint("", camelContext);
+        QuickfixjEndpoint endpoint = new QuickfixjEndpoint(null, "", camelContext);
         
         Message message = new Message();     
         message.getHeader().setString(MsgType.FIELD, MsgType.ORDER_SINGLE);
@@ -116,7 +116,7 @@ public class QuickfixjConvertersTest {
     @Test
     public void convertToExchangeWithNullMessage() {
         SessionID sessionID = new SessionID("FIX.4.0", "FOO", "BAR");
-        QuickfixjEndpoint endpoint = new QuickfixjEndpoint("", camelContext);
+        QuickfixjEndpoint endpoint = new QuickfixjEndpoint(null, "", camelContext);
         
         Exchange exchange = QuickfixjConverters.toExchange(endpoint, sessionID, null, QuickfixjEventCategory.AppMessageSent);
         

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java Thu Jun  2 15:26:12 2011
@@ -16,33 +16,133 @@
  */
 package org.apache.camel.component.quickfixj;
 
+import java.io.IOException;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import javax.management.JMException;
+
 import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Matchers;
 import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
+import quickfix.ConfigError;
+import quickfix.FieldConvertError;
 import quickfix.FixVersions;
 import quickfix.Message;
+import quickfix.MessageUtils;
+import quickfix.Session;
 import quickfix.SessionID;
+import quickfix.field.BeginString;
+import quickfix.field.MsgType;
+import quickfix.field.SenderCompID;
+import quickfix.field.TargetCompID;
+import quickfix.fix42.Email;
 
 public class QuickfixjProducerTest {
-
+	private Exchange mockExchange;
+	private QuickfixjEndpoint mockEndpoint;
+	private org.apache.camel.Message mockCamelMessage;
+	private QuickfixjProducer producer;
+	private SessionID sessionID;
+	private Message inboundFixMessage;
+	private QuickfixjEngine quickfixjEngine;
+
+	@Before
+	public void setUp() throws ConfigError, FieldConvertError, IOException, JMException {
+        mockExchange = Mockito.mock(Exchange.class);
+        mockEndpoint = Mockito.mock(QuickfixjEndpoint.class);
+        mockCamelMessage = Mockito.mock(org.apache.camel.Message.class);
+        Mockito.when(mockExchange.getIn()).thenReturn(mockCamelMessage);
+        Mockito.when(mockExchange.getPattern()).thenReturn(ExchangePattern.InOnly);
+        
+        quickfixjEngine = TestSupport.createEngine();
+        Mockito.when(mockEndpoint.getEngine()).thenReturn(quickfixjEngine);
+        
+        inboundFixMessage = new Message();
+        inboundFixMessage.getHeader().setString(BeginString.FIELD, FixVersions.BEGINSTRING_FIX44);
+        inboundFixMessage.getHeader().setString(SenderCompID.FIELD, "SENDER");
+        inboundFixMessage.getHeader().setString(TargetCompID.FIELD, "TARGET");
+        sessionID = MessageUtils.getSessionID(inboundFixMessage);
+   
+        Mockito.when(mockCamelMessage.getBody(Message.class)).thenReturn(inboundFixMessage);
+
+        Mockito.when(mockEndpoint.getSessionID()).thenReturn(sessionID);     
+
+        producer = Mockito.spy(new QuickfixjProducer(mockEndpoint));
+	}
+	
+	@SuppressWarnings("serial")
+	public class TestException extends RuntimeException {
+		
+	}
+	
     @Test
     public void setExceptionOnExchange() throws Exception {
-        Exchange mockExchange = Mockito.mock(Exchange.class);
+        Session mockSession = Mockito.spy(TestSupport.createSession(sessionID));
+		Mockito.doReturn(mockSession).when(producer).getSession(MessageUtils.getSessionID(inboundFixMessage));
+		Mockito.doThrow(new TestException()).when(mockSession).send(Mockito.isA(Message.class));
 
-        QuickfixjEndpoint mockEndpoint = Mockito.mock(QuickfixjEndpoint.class);
-        org.apache.camel.Message mockCamelMessage = Mockito.mock(org.apache.camel.Message.class);
-        Mockito.when(mockExchange.getIn()).thenReturn(mockCamelMessage);
-        Mockito.when(mockCamelMessage.getBody(Message.class)).thenReturn(new Message());
+		producer.process(mockExchange);    
+        Mockito.verify(mockExchange).setException(Matchers.isA(TestException.class));
+    }
+    
+    @Test
+    public void processInOnlyExchange() throws Exception {        
+        Session mockSession = Mockito.spy(TestSupport.createSession(sessionID));
+		Mockito.doReturn(mockSession).when(producer).getSession(MessageUtils.getSessionID(inboundFixMessage));
+		Mockito.doReturn(true).when(mockSession).send(Mockito.isA(Message.class));
+        
+        producer.process(mockExchange);
+        
+        Mockito.verify(mockExchange, Mockito.never()).setException(Matchers.isA(IllegalStateException.class));
+        Mockito.verify(mockSession).send(inboundFixMessage);
+    }
 
-        SessionID sessionID = new SessionID(FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET");       
-        Mockito.when(mockEndpoint.getSessionID()).thenReturn(sessionID);
+    @Test
+    public void processInOutExchange() throws Exception {
+        Mockito.when(mockExchange.getPattern()).thenReturn(ExchangePattern.InOut);
+        Mockito.when(mockExchange.getProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY)).
+        	thenReturn(new MessagePredicate(sessionID, MsgType.EMAIL));
+        Mockito.when(mockExchange.getProperty(
+				QuickfixjProducer.CORRELATION_TIMEOUT_KEY,
+				1000L, Long.class)).thenReturn(5000L);
+				
+        org.apache.camel.Message mockOutboundCamelMessage = Mockito.mock(org.apache.camel.Message.class);
+		Mockito.when(mockExchange.getOut()).thenReturn(mockOutboundCamelMessage);
         
-        QuickfixjProducer producer = new QuickfixjProducer(mockEndpoint);
+        final Message outboundFixMessage = new Email();
+        outboundFixMessage.getHeader().setString(SenderCompID.FIELD, "TARGET");
+        outboundFixMessage.getHeader().setString(TargetCompID.FIELD, "SENDER");
         
+        Session mockSession = Mockito.spy(TestSupport.createSession(sessionID));
+		Mockito.doReturn(mockSession).when(producer).getSession(MessageUtils.getSessionID(inboundFixMessage));
+		Mockito.doAnswer(new Answer<Boolean>() {
+			@Override
+			public Boolean answer(InvocationOnMock invocation) throws Throwable {
+				new Timer().schedule(new TimerTask() {				
+					@Override
+					public void run() {
+						try {
+							quickfixjEngine.getMessageCorrelator().onEvent(QuickfixjEventCategory.AppMessageReceived, sessionID, outboundFixMessage);
+						} catch (Exception e) {
+							e.printStackTrace();
+						}
+					}
+				}, 10);
+				return true;
+			}			
+		}).when(mockSession).send(Mockito.isA(Message.class));
+
         producer.process(mockExchange);
         
-        Mockito.verify(mockExchange).setException(Matchers.isA(IllegalStateException.class));
+        Mockito.verify(mockExchange, Mockito.never()).setException(Matchers.isA(IllegalStateException.class));
+        Mockito.verify(mockSession).send(inboundFixMessage);
+        Mockito.verify(mockOutboundCamelMessage).setBody(outboundFixMessage);
     }
 }

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/TestSupport.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/TestSupport.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/TestSupport.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/TestSupport.java Thu Jun  2 15:26:12 2011
@@ -19,7 +19,23 @@ package org.apache.camel.component.quick
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.util.Date;
 
+import javax.management.JMException;
+
+import org.mockito.Mockito;
+
+import quickfix.Acceptor;
+import quickfix.Application;
+import quickfix.ConfigError;
+import quickfix.DefaultSessionFactory;
+import quickfix.FieldConvertError;
+import quickfix.LogFactory;
+import quickfix.MessageFactory;
+import quickfix.MessageStore;
+import quickfix.MessageStoreFactory;
+import quickfix.Session;
+import quickfix.SessionFactory;
 import quickfix.SessionID;
 import quickfix.SessionSettings;
 import quickfix.field.EmailThreadID;
@@ -55,4 +71,51 @@ public final class TestSupport {
         email.addGroup(text);
         return email;
     }
+    
+    public static Session createSession(SessionID sessionID) throws ConfigError, IOException {
+    	MessageStoreFactory mockMessageStoreFactory = Mockito.mock(MessageStoreFactory.class);
+    	MessageStore mockMessageStore = Mockito.mock(MessageStore.class);
+    	Mockito.when(mockMessageStore.getCreationTime()).thenReturn(new Date());
+    	
+		Mockito.when(mockMessageStoreFactory.create(sessionID)).thenReturn(mockMessageStore);
+    	
+		DefaultSessionFactory factory = new DefaultSessionFactory(
+    			Mockito.mock(Application.class),
+    			mockMessageStoreFactory,
+    			Mockito.mock(LogFactory.class));
+    	
+    	SessionSettings settings = new SessionSettings();
+    	settings.setLong(Session.SETTING_HEARTBTINT, 10);
+    	settings.setString(Session.SETTING_START_TIME, "00:00:00");
+    	settings.setString(Session.SETTING_END_TIME, "00:00:00");
+    	settings.setString(SessionFactory.SETTING_CONNECTION_TYPE, SessionFactory.ACCEPTOR_CONNECTION_TYPE);
+    	settings.setBool(Session.SETTING_USE_DATA_DICTIONARY, false);
+    	
+		return factory.create(sessionID, settings);
+    			
+    }
+    
+    public static QuickfixjEngine createEngine() throws ConfigError, FieldConvertError, IOException, JMException {   	
+    	SessionID sessionID = new SessionID("FIX.4.4:SENDER->TARGET");
+
+    	MessageStoreFactory mockMessageStoreFactory = Mockito.mock(MessageStoreFactory.class);
+    	MessageStore mockMessageStore = Mockito.mock(MessageStore.class);
+    	Mockito.when(mockMessageStore.getCreationTime()).thenReturn(new Date());
+		Mockito.when(mockMessageStoreFactory.create(sessionID)).thenReturn(mockMessageStore);
+    	
+    	SessionSettings settings = new SessionSettings();
+    	
+    	settings.setLong(sessionID, Session.SETTING_HEARTBTINT, 10);
+    	settings.setString(sessionID, Session.SETTING_START_TIME, "00:00:00");
+    	settings.setString(sessionID, Session.SETTING_END_TIME, "00:00:00");
+    	settings.setString(sessionID, SessionFactory.SETTING_CONNECTION_TYPE, SessionFactory.ACCEPTOR_CONNECTION_TYPE);
+    	settings.setLong(sessionID, Acceptor.SETTING_SOCKET_ACCEPT_PORT, 8000);
+    	settings.setBool(sessionID, Session.SETTING_USE_DATA_DICTIONARY, false);
+    	
+		return new QuickfixjEngine("", settings, false, 
+				mockMessageStoreFactory, 
+				Mockito.mock(LogFactory.class), 
+				Mockito.mock(MessageFactory.class));
+    			
+    }
 }

Added: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java?rev=1130604&view=auto
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java (added)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java Thu Jun  2 15:26:12 2011
@@ -0,0 +1,172 @@
+/**
+ * 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.quickfixj.examples;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Body;
+import org.apache.camel.Exchange;
+import org.apache.camel.Header;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.quickfixj.MessagePredicate;
+import org.apache.camel.component.quickfixj.QuickfixjComponent;
+import org.apache.camel.component.quickfixj.QuickfixjEndpoint;
+import org.apache.camel.component.quickfixj.QuickfixjEventCategory;
+import org.apache.camel.component.quickfixj.QuickfixjProducer;
+import org.apache.camel.component.quickfixj.examples.transform.QuickfixjMessageJsonTransformer;
+import org.apache.camel.component.quickfixj.examples.util.CountDownLatchDecrementer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.jetty.util.log.Log;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import quickfix.FieldNotFound;
+import quickfix.SessionID;
+import quickfix.field.AvgPx;
+import quickfix.field.ClOrdID;
+import quickfix.field.CumQty;
+import quickfix.field.ExecID;
+import quickfix.field.ExecTransType;
+import quickfix.field.ExecType;
+import quickfix.field.LeavesQty;
+import quickfix.field.MsgType;
+import quickfix.field.OrdStatus;
+import quickfix.field.OrderID;
+import quickfix.field.Side;
+import quickfix.field.Symbol;
+import quickfix.fix42.ExecutionReport;
+import quickfix.fix42.OrderStatusRequest;
+
+public class RequestReplyExample {
+    private static final Logger LOG = LoggerFactory.getLogger(QuickfixjComponent.class);
+
+    public static void main(String[] args) throws Exception {
+        new RequestReplyExample().run();
+    }
+    
+    public void run() throws Exception {        
+        DefaultCamelContext context = new DefaultCamelContext();
+        final CountDownLatch logonLatch = new CountDownLatch(1);
+        final String orderStatusServiceUrl = "http://localhost:9123/order/status";
+        
+        RouteBuilder routes = new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+            	// Synchronize the logon so we don't start sending status requests too early
+                from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET").
+                    filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)).
+                    bean(new CountDownLatchDecrementer("logon", logonLatch));
+
+                // Incoming status requests are converted to InOut exchange pattern and passed to the
+                // order status service. The response is sent back to the session making the request.
+                from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:MARKET->TRADER&exchangePattern=InOut").
+                	filter(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.ORDER_STATUS_REQUEST)).
+                    bean(new MarketOrderStatusService());
+                
+                from ("jetty:" + orderStatusServiceUrl).
+	            	bean(new OrderStatusRequestTransformer()).
+                	routingSlip(bean(FixSessionRouter.class, "route")).
+	            	bean(new QuickfixjMessageJsonTransformer());
+            }
+        };
+        
+        context.addRoutes(routes);
+        
+        LOG.info("Starting Camel context");
+        context.start();
+        
+        if (!logonLatch.await(5L, TimeUnit.SECONDS)) {
+            throw new IllegalStateException("Logon did not succeed");
+        }
+        
+        // Send a request to the order status web service.
+        // Verify that the response is a JSON response.
+        
+        URL orderStatusUrl = new URL(orderStatusServiceUrl + "?sessionID=FIX.4.2:TRADER->MARKET&orderID=abc");
+		HttpURLConnection connection = (HttpURLConnection) orderStatusUrl.openConnection();
+        BufferedReader orderStatusReply = new BufferedReader(new InputStreamReader(connection.getInputStream()));
+        String line = orderStatusReply.readLine();
+        if (!line.equals("\"message\": {")) {
+        	throw new Exception("Don't appear to be a JSON response");
+        }
+        else {
+        	StringBuilder sb = new StringBuilder();
+        	while (line != null) {
+        		sb.append(line);
+        		sb.append('\n');
+        		line = orderStatusReply.readLine();
+        	}
+        	Log.info("Web request response:\n" + sb);
+        	
+        }
+        orderStatusReply.close();
+        
+        LOG.info("Shutting down Camel context");
+        context.stop();
+        
+        LOG.info("Example complete");
+    }
+        
+    public static class OrderStatusRequestTransformer {
+    	public void transform(Exchange exchange) throws FieldNotFound {
+    		String sessionID = (String) exchange.getIn().getHeader("sessionID");
+    		String orderID = (String) exchange.getIn().getHeader("orderID");
+    		
+            OrderStatusRequest request = new OrderStatusRequest(new ClOrdID("XYZ"), new Symbol("GOOG"), new Side(Side.BUY));
+            request.set(new OrderID(orderID));
+             
+            // Look for a reply execution report back to the requester session
+            // and having the requested OrderID. This is a loose correlation but the best
+            // we can do with FIX 4.2. Newer versions of FIX have an optional explicit correlation field.
+            exchange.setProperty(QuickfixjProducer.CORRELATION_CRITERIA_KEY,
+            		new MessagePredicate(new SessionID(sessionID), MsgType.EXECUTION_REPORT)
+            			.withField(OrderID.FIELD, request.getString(OrderID.FIELD)));
+            
+            exchange.getIn().setBody(request);
+    	}
+    }
+    
+    public static class MarketOrderStatusService {
+        private static final Logger LOG = LoggerFactory.getLogger(QuickfixjComponent.class);
+        
+        public ExecutionReport getOrderStatus(OrderStatusRequest request) throws FieldNotFound {
+        	LOG.info("Received order status request for orderId=" + request.getOrderID().getValue());
+    		return new ExecutionReport(
+    				request.getOrderID(), new ExecID(UUID.randomUUID().toString()),
+    						new ExecTransType(ExecTransType.STATUS), 
+    						new ExecType(ExecType.REJECTED),
+    						new OrdStatus(OrdStatus.REJECTED),
+    						new Symbol("GOOG"),
+    						new Side(Side.BUY),
+    						new LeavesQty(100),
+    						new CumQty(0),
+    						new AvgPx(0));
+    	}
+    }
+    
+    public static class FixSessionRouter {
+    	public String route(@Header("sessionID") String sessionID, @Body Object body) {
+    		return String.format("quickfix:examples/inprocess.cfg?sessionID=%s", sessionID);
+    	}
+    }
+}

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/trading/TradeExecutorExample.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/trading/TradeExecutorExample.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/trading/TradeExecutorExample.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/trading/TradeExecutorExample.java Thu Jun  2 15:26:12 2011
@@ -63,7 +63,6 @@ public class TradeExecutorExample {
             @Override
             public void configure() throws Exception {
                 // Release latch when session logon events are received
-                // We expect four logon events (four sessions)
                 from("quickfix:examples/inprocess.cfg").
                     filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)).
                     bean(new CountDownLatchDecrementer("logon", logonLatch));

Modified: camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java Thu Jun  2 15:26:12 2011
@@ -18,16 +18,32 @@ package org.apache.camel.component.quick
 
 import java.util.Iterator;
 
+import quickfix.ConfigError;
 import quickfix.DataDictionary;
 import quickfix.Field;
 import quickfix.FieldMap;
+import quickfix.FieldNotFound;
 import quickfix.FieldType;
 import quickfix.Group;
 import quickfix.Message;
-
+import quickfix.MessageUtils;
+import quickfix.Session;
+import quickfix.SessionID;
 
 public class QuickfixjMessageJsonTransformer {
-    
+   
+	public String transform(Message message) throws FieldNotFound, ConfigError {
+		SessionID sessionID = MessageUtils.getSessionID(message);
+        Session session = Session.lookupSession(sessionID);
+        DataDictionary dataDictionary = session.getDataDictionary();
+        
+        if (dataDictionary == null) {
+            throw new IllegalStateException("No Data Dictionary. Exchange must reference an existing session");
+        }
+        
+		return transform(message, dataDictionary);
+	}
+	
     public String transform(Message message, DataDictionary dataDictionary) {
         return transform(message, "", dataDictionary);
     }

Modified: camel/trunk/components/camel-quickfix/src/test/resources/examples/inprocess.cfg
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/resources/examples/inprocess.cfg?rev=1130604&r1=1130603&r2=1130604&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/resources/examples/inprocess.cfg (original)
+++ camel/trunk/components/camel-quickfix/src/test/resources/examples/inprocess.cfg Thu Jun  2 15:26:12 2011
@@ -23,6 +23,7 @@
 UseJmx=Y
 SocketAcceptProtocol=VM_PIPE
 SocketConnectProtocol=VM_PIPE
+HeartBtInt=120
 
 #
 # Initiator for simulating a trader



Mime
View raw message