camel-commits mailing list archives

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

URL: http://svn.apache.org/viewvc?rev=1130661&view=rev
Log:
CAMEL-4038. Checkstyle fixes.

Modified:
    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/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/QuickfixjEventListener.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/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/RequestReplyExample.java
    camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/transform/QuickfixjMessageJsonTransformer.java

Modified: 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=1130661&r1=1130660&r2=1130661&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessageCorrelator.java Thu Jun  2 17:15:30 2011
@@ -28,82 +28,74 @@ 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 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;
-		}
-	}
+    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);
+        rules.add(correlationRule);
+
+        final long timeout = exchange.getProperty(
+            QuickfixjProducer.CORRELATION_TIMEOUT_KEY,
+            DEFAULT_CORRELATION_TIMEOUT, Long.class);
+
+        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;
+        }
+    }
 }

Modified: 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=1130661&r1=1130660&r2=1130661&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/MessagePredicate.java Thu Jun  2 17:15:30 2011
@@ -1,3 +1,19 @@
+/**
+ * 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.ArrayList;
@@ -13,56 +29,56 @@ 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;
-	}
+    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/QuickfixjConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjConsumer.java?rev=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -28,7 +28,7 @@ import quickfix.SessionID;
 
 public class QuickfixjConsumer extends DefaultConsumer {
 
-	public QuickfixjConsumer(Endpoint endpoint, Processor processor) {
+    public QuickfixjConsumer(Endpoint endpoint, Processor processor) {
         super(endpoint, processor);
     }
 
@@ -37,7 +37,7 @@ public class QuickfixjConsumer extends D
             try {
                 getProcessor().process(exchange);
                 if (exchange.getPattern().isOutCapable() && exchange.hasOut()) {
-                	sendOutMessage(exchange);
+                    sendOutMessage(exchange);
                 }
             } catch (Exception e) {
                 exchange.setException(e);
@@ -45,29 +45,29 @@ public class QuickfixjConsumer extends D
         }
     }
 
-	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);
-		}
-	}
+    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);
-	}
+    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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -34,115 +34,106 @@ 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>();
-	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();
-	}
+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) {
+        if (this.sessionID.equals(sessionID)) {
+            return true;
+        }
+        return 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() {
+        if (sessionID == null) {
+            return false;
+        }
+        return 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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -120,17 +120,14 @@ public class QuickfixjEngine {
             MessageStoreFactory messageStoreFactoryOverride, LogFactory sessionLogFactoryOverride,
             MessageFactory messageFactoryOverride) throws ConfigError, FieldConvertError, IOException, JMException {
 
-    	addEventListener(messageCorrelator);
-    	
+        addEventListener(messageCorrelator);
+
         this.uri = uri;
         this.forcedShutdown = forcedShutdown;
         
         messageFactory = messageFactoryOverride != null ? messageFactoryOverride : new DefaultMessageFactory();
-
         sessionLogFactory = sessionLogFactoryOverride != null ? sessionLogFactoryOverride : inferLogFactory(settings);
-
-        messageStoreFactory = messageStoreFactoryOverride != null ? messageStoreFactoryOverride
-                : inferMessageStoreFactory(settings);
+        messageStoreFactory = messageStoreFactoryOverride != null ? messageStoreFactoryOverride : inferMessageStoreFactory(settings);
 
         // Set default session schedule if not specified in configuration
         if (!settings.isSetting(Session.SETTING_START_TIME)) {
@@ -165,15 +162,15 @@ public class QuickfixjEngine {
             Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
 
             if (isConnectorRole(settings, SessionFactory.ACCEPTOR_CONNECTION_TYPE)) {
-                acceptor = createAcceptor(new Dispatcher(), settings, messageStoreFactory, sessionLogFactory,
-                        messageFactory, threadModel);
+                acceptor = createAcceptor(new Dispatcher(), settings, messageStoreFactory, 
+                    sessionLogFactory, messageFactory, threadModel);
             } else {
                 acceptor = null;
             }
 
             if (isConnectorRole(settings, SessionFactory.INITIATOR_CONNECTION_TYPE)) {
-                initiator = createInitiator(new Dispatcher(), settings, messageStoreFactory, sessionLogFactory,
-                        messageFactory, threadModel);
+                initiator = createInitiator(new Dispatcher(), settings, messageStoreFactory, 
+                    sessionLogFactory, messageFactory, threadModel);
             } else {
                 initiator = null;
             }
@@ -467,8 +464,8 @@ public class QuickfixjEngine {
     }
 
     public MessageCorrelator getMessageCorrelator() {
-		return messageCorrelator;
-	}
+        return messageCorrelator;
+    }
 
     // For Testing
     Initiator getInitiator() {

Modified: camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEventListener.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEventListener.java?rev=1130661&r1=1130660&r2=1130661&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEventListener.java (original)
+++ camel/trunk/components/camel-quickfix/src/main/java/org/apache/camel/component/quickfixj/QuickfixjEventListener.java Thu Jun  2 17:15:30 2011
@@ -24,7 +24,5 @@ import quickfix.SessionID;
  * @see QuickfixjEngine
  */
 public interface QuickfixjEventListener {
-    
     void onEvent(QuickfixjEventCategory eventCategory, SessionID sessionID, Message message) throws Exception;
-    
 }

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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -30,8 +30,8 @@ import quickfix.SessionID;
 public class QuickfixjProducer extends DefaultProducer {
     public static final String CORRELATION_TIMEOUT_KEY = "CorrelationTimeout";
     public static final String CORRELATION_CRITERIA_KEY = "CorrelationCriteria";
-	
-	private final SessionID sessionID;
+    
+    private final SessionID sessionID;
     
     public QuickfixjProducer(Endpoint endpoint) {
         super(endpoint);
@@ -42,47 +42,43 @@ public class QuickfixjProducer extends D
         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);
-	}
+    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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -41,8 +41,8 @@ public class QuickfixjConsumerTest {
     private Exchange mockExchange;
     private Processor mockProcessor;
     private Endpoint mockEndpoint;
-	private SessionID sessionID;
-	private Message inboundFixMessage;
+    private SessionID sessionID;
+    private Message inboundFixMessage;
     
     @Before
     public void setUp() {
@@ -56,11 +56,9 @@ public class QuickfixjConsumerTest {
         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);
+        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);  
     }
@@ -123,19 +121,17 @@ public class QuickfixjConsumerTest {
         Mockito.when(mockExchange.hasOut()).thenReturn(true);
         Mockito.when(mockExchange.getOut()).thenReturn(mockCamelOutMessage);
         Message outboundFixMessage = new Message();
-		Mockito.when(mockCamelOutMessage.getBody(Message.class)).thenReturn(outboundFixMessage);
+        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));
+        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/QuickfixjProducerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/QuickfixjProducerTest.java?rev=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -45,16 +45,16 @@ 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;
+    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 {
+    @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);
@@ -75,28 +75,28 @@ public class QuickfixjProducerTest {
         Mockito.when(mockEndpoint.getSessionID()).thenReturn(sessionID);     
 
         producer = Mockito.spy(new QuickfixjProducer(mockEndpoint));
-	}
-	
-	@SuppressWarnings("serial")
-	public class TestException extends RuntimeException {
-		
-	}
-	
+    }
+    
+    @SuppressWarnings("serial")
+    public class TestException extends RuntimeException {
+        
+    }
+    
     @Test
     public void setExceptionOnExchange() throws Exception {
         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));
+        Mockito.doReturn(mockSession).when(producer).getSession(MessageUtils.getSessionID(inboundFixMessage));
+        Mockito.doThrow(new TestException()).when(mockSession).send(Mockito.isA(Message.class));
 
-		producer.process(mockExchange);    
+        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));
+        Mockito.doReturn(mockSession).when(producer).getSession(MessageUtils.getSessionID(inboundFixMessage));
+        Mockito.doReturn(true).when(mockSession).send(Mockito.isA(Message.class));
         
         producer.process(mockExchange);
         
@@ -107,37 +107,38 @@ public class QuickfixjProducerTest {
     @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_CRITERIA_KEY)).thenReturn(
+            new MessagePredicate(sessionID, MsgType.EMAIL));
         Mockito.when(mockExchange.getProperty(
-				QuickfixjProducer.CORRELATION_TIMEOUT_KEY,
-				1000L, Long.class)).thenReturn(5000L);
-				
+            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);
+        Mockito.when(mockExchange.getOut()).thenReturn(mockOutboundCamelMessage);
         
         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));
+        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) {
+                            // TODO: probably not the best thing... use a LOG
+                            e.printStackTrace();
+                        }
+                    }
+                }, 10);
+                return true;
+            }            
+        }).when(mockSession).send(Mockito.isA(Message.class));
 
         producer.process(mockExchange);
         

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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -73,49 +73,47 @@ public final class TestSupport {
     }
     
     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);
-    			
+        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");
+    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));
-    			
+        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));
     }
 }

Modified: 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=1130661&r1=1130660&r2=1130661&view=diff
==============================================================================
--- camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java (original)
+++ camel/trunk/components/camel-quickfix/src/test/java/org/apache/camel/component/quickfixj/examples/RequestReplyExample.java Thu Jun  2 17:15:30 2011
@@ -36,7 +36,6 @@ import org.apache.camel.component.quickf
 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;
 
@@ -72,21 +71,21 @@ public class RequestReplyExample {
         RouteBuilder routes = new RouteBuilder() {
             @Override
             public void configure() throws Exception {
-            	// Synchronize the logon so we don't start sending status requests too early
+                // 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("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());
+                from("jetty:" + orderStatusServiceUrl)
+                    .bean(new OrderStatusRequestTransformer())
+                    .routingSlip(bean(FixSessionRouter.class, "route"))
+                    .bean(new QuickfixjMessageJsonTransformer());
             }
         };
         
@@ -103,21 +102,19 @@ public class RequestReplyExample {
         // 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();
+        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);
-        	
+            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();
         
@@ -128,45 +125,44 @@ public class RequestReplyExample {
     }
         
     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");
-    		
+        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.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));
-    	}
+            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);
-    	}
+        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/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=1130661&r1=1130660&r2=1130661&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 17:15:30 2011
@@ -32,8 +32,8 @@ import quickfix.SessionID;
 
 public class QuickfixjMessageJsonTransformer {
    
-	public String transform(Message message) throws FieldNotFound, ConfigError {
-		SessionID sessionID = MessageUtils.getSessionID(message);
+    public String transform(Message message) throws FieldNotFound, ConfigError {
+        SessionID sessionID = MessageUtils.getSessionID(message);
         Session session = Session.lookupSession(sessionID);
         DataDictionary dataDictionary = session.getDataDictionary();
         
@@ -41,9 +41,9 @@ public class QuickfixjMessageJsonTransfo
             throw new IllegalStateException("No Data Dictionary. Exchange must reference an existing session");
         }
         
-		return transform(message, dataDictionary);
-	}
-	
+        return transform(message, dataDictionary);
+    }
+
     public String transform(Message message, DataDictionary dataDictionary) {
         return transform(message, "", dataDictionary);
     }



Mime
View raw message