# FIXIncomingMessageHandler.java

    /*
    *  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.synapse.transport.fix;

    import org.apache.axis2.AxisFault;
    import org.apache.axis2.Constants;
    import org.apache.axis2.wsdl.WSDLConstants;
    import org.apache.axis2.context.ConfigurationContext;
    import org.apache.axis2.context.MessageContext;
    import org.apache.axis2.description.AxisOperation;
    import org.apache.axis2.description.AxisService;
    import org.apache.axis2.description.Parameter;
    import org.apache.axis2.transport.base.AbstractTransportListener;
    import org.apache.axis2.transport.base.AbstractTransportSender;
    import org.apache.axis2.transport.base.BaseConstants;
    import org.apache.axis2.transport.base.BaseUtils;
    import org.apache.axis2.transport.base.threads.WorkerPool;
    import org.apache.synapse.transport.fix.FIXConstants;
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import quickfix.*;
    import quickfix.field.MsgSeqNum;
    import quickfix.field.MsgType;
    import quickfix.field.SenderCompID;
    import quickfix.field.TargetCompID;

    import javax.xml.namespace.QName;
    import java.util.Map;
    import java.util.Queue;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.atomic.AtomicInteger;

    /**
     * FIXIncomingMessageHandler is responsible for handling all incoming FIX messages. This is where the
     * Quickfix/J engine meets Synapse core. Admin level FIX messages are handled by Quickfix/J itself.
     * All the application level messages are handed over to the Synapse core.
     */
    public class FIXIncomingMessageHandler implements Application {

        private ConfigurationContext cfgCtx;
        /**
         * A thread pool used to process incoming FIX messages
         */
        private WorkerPool workerPool;
        /**
         * AxisService to which this FIX application is bound to
         */
        private AxisService service;
        private Log log;
        /**
         * A boolean value indicating the type of the FIX application
         */
        private boolean acceptor;
        /**
         * A Map of counters with one counter per session
         */
        private Map<SessionID, AtomicInteger> countersMap;
        private Queue<MessageContext> outgoingMessages;
        private boolean allNewApproach = true;
        private boolean dropExtraResponses = false;
        private Semaphore semaphore;
        private SessionEventHandler eventHandler;

        public FIXIncomingMessageHandler(ConfigurationContext cfgCtx, WorkerPool workerPool,
                                         AxisService service, boolean acceptor) {
            this.cfgCtx = cfgCtx;
            this.workerPool = workerPool;
            this.service = service;
            this.log = LogFactory.getLog(this.getClass());
            this.acceptor = acceptor;
            countersMap = new ConcurrentHashMap<SessionID, AtomicInteger>();
            outgoingMessages = new LinkedBlockingQueue<MessageContext>();
            semaphore = new Semaphore(0);
            getResponseHandlingApproach();

            Parameter eventHandlerParam;
            if (acceptor) {
                eventHandlerParam = service.getParameter(FIXConstants.FIX_ACCEPTOR_EVENT_HANDLER);
            } else {
                eventHandlerParam = service.getParameter(FIXConstants.FIX_INITIATOR_EVENT_HANDLER);
            }

            if (eventHandlerParam != null && eventHandlerParam.getValue() != null &&
                !"".equals(eventHandlerParam.getValue())) {
                try {
                    Class clazz = getClass().getClassLoader().loadClass(
                            (String) eventHandlerParam.getValue());
                    eventHandler = (SessionEventHandler) clazz.newInstance();
                } catch (ClassNotFoundException e) {
                    log.error("Unable to find the session event handler class: " +
                              eventHandlerParam.getValue(), e);
                } catch (Exception e) {
                    log.error("Error while initializing the session event handler class: " +
                              eventHandlerParam.getValue(), e);
                }
            }
        }

        private void getResponseHandlingApproach() {
            Parameter param = service.getParameter(FIXConstants.FIX_RESPONSE_HANDLER_APPROACH);
            if (param != null && "false".equals(param.getValue().toString())) {
                allNewApproach = false;
            }

            Parameter dropResponsesParam = service.getParameter(FIXConstants.FIX_DROP_EXTRA_RESPONSES);
            if (dropResponsesParam != null && "true".equals(dropResponsesParam.getValue().toString())) {
                dropExtraResponses = true;
            }
        }

        public void setOutgoingMessageContext(MessageContext msgCtx) {
            if (!allNewApproach) {
                outgoingMessages.offer(msgCtx);
            }
        }

        public void acquire() throws InterruptedException {
            semaphore.acquire();
        }

        private void handleException(String msg, Exception e) {
            log.error(msg, e);
            throw new AxisFIXException(msg, e);
        }

        /**
         * This method is called when quickfix creates a new session. A session
         * comes into and remains in existence for the life of the application.
         * Sessions exist whether or not a counter party is connected to it. As soon
         * as a session is created, the application can begin sending messages to it. If no one
         * is logged on, the messages will be sent at the time a connection is
         * established with the counter party.
         *
         * @param sessionID QuickFIX session ID
         */
        public void onCreate(SessionID sessionID) {
            log.info("New FIX session created: " + sessionID.toString());
            if (eventHandler != null) {
                eventHandler.onCreate(this, sessionID);
            }
        }

        /**
         * This callback notifies when a valid logon has been established with a
         * counter party. This is called when a connection has been established and
         * the FIX logon process has completed with both parties exchanging valid
         * logon messages.
         *
         * @param sessionID QuickFIX session ID
         */
        public void onLogon(SessionID sessionID) {
            if (!countersMap.containsKey(sessionID)) {
                countersMap.put(sessionID, new AtomicInteger(0));
            }
            log.info("FIX session logged on: " + sessionID.toString());
            semaphore.release();

            if (eventHandler != null) {
                eventHandler.onLogon(this, sessionID);
            }
        }

        /**
         * This callback notifies when a FIX session is no longer online. This
         * could happen during a normal logout exchange or because of a forced
         * termination or a loss of network connection.
         *
         * @param sessionID QuickFIX session ID
         */
        public void onLogout(SessionID sessionID) {
            FIXTransportSender trpSender = (FIXTransportSender) cfgCtx.getAxisConfiguration().
                    getTransportOut(FIXConstants.TRANSPORT_NAME).getSender();
            trpSender.logOutIncomingSession(sessionID);
            countersMap.remove(sessionID);
            log.info("FIX session logged out: " + sessionID.toString());

            if (eventHandler != null) {
                eventHandler.onLogout(this, sessionID);
            }
        }

        /**
         * This callback provides Synapse with a peek at the administrative messages
         * that are being sent from your FIX engine to the counter party. This is
         * normally not useful for an application however it is provided for any
         * logging one may wish to do.
         *
         * @param message   QuickFIX message
         * @param sessionID QuickFIX session ID
         */
        public void toAdmin(Message message, SessionID sessionID) {
            if (log.isDebugEnabled()) {
                StringBuffer sb = new StringBuffer();
                try {
                    sb.append("Sending admin level FIX message to ").append(message.getHeader().getField(new TargetCompID()).getValue());
                    sb.append("\nMessage Type: ").append(message.getHeader().getField(new MsgType()).getValue());
                    sb.append("\nMessage Sequence Number: ").append(message.getHeader().getField(new MsgSeqNum()).getValue());
                    sb.append("\nSender ID: ").append(message.getHeader().getField(new SenderCompID()).getValue());
                } catch (FieldNotFound e) {
                    sb.append("Sending admin level FIX message...");
                    log.warn("One or more required fields are not found in the response message", e);
                }
                log.debug(sb.toString());
                if (log.isTraceEnabled()) {
                    log.trace("Message: " + message.toString());
                }
            }
            try {
                // Before FIX server sends (35=A) admin login message get username/password values from proxy configuration
                if (message.getHeader().getField(new StringField(FIXConstants.FIX_MESSAGE_TYPE)).getValue().equals(FIXConstants.LOGON)) {

                    Parameter userName = service.getParameter(FIXConstants.FIX_USERNAME);
                    Parameter passWord = service.getParameter(FIXConstants.FIX_PASSWORD);

                    if (userName != null && passWord != null) {
                        message.setString(FIXConstants.USERNAME_TAG, userName.getValue().toString());
                        message.setString(FIXConstants.PASSWORD_TAG, passWord.getValue().toString());
                    }
                    if (log.isDebugEnabled()) {
                        log.debug("Username:" + userName + " or password:" + passWord + "from proxy configuration..");
                    }
                }
            } catch (FieldNotFound e) {
                log.error("One or more required fields are not found in the response message", e);
            }

            if (eventHandler != null) {
                eventHandler.toAdmin(this, message, sessionID);
            }
        }

        /**
         * This callback notifies when an administrative message is sent from a
         * counterparty to the FIX engine.
         *
         * @param message   QuickFIX message
         * @param sessionID QuickFIX session ID
         * @throws FieldNotFound
         * @throws IncorrectDataFormat
         * @throws IncorrectTagValue
         * @throws RejectLogon         causes a logon reject
         */
        public void fromAdmin(Message message, SessionID sessionID) throws FieldNotFound,
                                                                           IncorrectDataFormat,
                                                                           IncorrectTagValue,
                                                                           RejectLogon {

            if (log.isDebugEnabled()) {
                StringBuffer sb = new StringBuffer();
                sb.append("Received admin level FIX message from ").append(message.getHeader().getField(new SenderCompID()).getValue());
                sb.append("\nMessage Type: ").append(message.getHeader().getField(new MsgType()).getValue());
                sb.append("\nMessage Sequence Number: ").append(message.getHeader().getField(new MsgSeqNum()).getValue());
                sb.append("\nReceiver ID: ").append(message.getHeader().getField(new TargetCompID()).getValue());
                log.debug(sb.toString());
                if (log.isTraceEnabled()) {
                    log.trace("Message: " + message.toString());
                }
            }

            if (eventHandler != null) {
                eventHandler.fromAdmin(this, message, sessionID);
            }
        }

        /**
         * This is a callback for application messages that are being sent to a
         * counter party.
         *
         * @param message   QuickFIX message
         * @param sessionID QuickFIX session ID
         * @throws DoNotSend This exception aborts message transmission
         */
        public void toApp(Message message, SessionID sessionID) throws DoNotSend {
            if (log.isDebugEnabled()) {
                StringBuffer sb = new StringBuffer();
                try {
                    sb.append("Sending application level FIX message to ").append(message.getHeader().getField(new TargetCompID()).getValue());
                    sb.append("\nMessage Type: ").append(message.getHeader().getField(new MsgType()).getValue());
                    sb.append("\nMessage Sequence Number: ").append(message.getHeader().getField(new MsgSeqNum()).getValue());
                    sb.append("\nSender ID: ").append(message.getHeader().getField(new SenderCompID()).getValue());
                } catch (FieldNotFound e) {
                    sb.append("Sending application level FIX message...");
                    log.warn("One or more required fields are not found in the response message", e);
                }
                log.debug(sb.toString());
                if (log.isTraceEnabled()) {
                    log.trace("Message: " + message.toString());
                }
            }

            if (eventHandler != null) {
                eventHandler.toApp(this, message, sessionID);
            }
        }

        /**
         * This callback receives messages for the application. This is one of the
         * core entry points for the FIX application. Every application level
         * request will come through here. A new thread will be spawned from the
         * thread pool for each incoming message.
         *
         * @param message   QuickFIX message
         * @param sessionID QuickFIX session ID
         * @throws FieldNotFound
         * @throws IncorrectDataFormat
         * @throws IncorrectTagValue
         * @throws UnsupportedMessageType
         */
        public void fromApp(Message message, SessionID sessionID)
                throws FieldNotFound, IncorrectDataFormat,
                       IncorrectTagValue, UnsupportedMessageType {
            if (log.isDebugEnabled()) {
                StringBuffer sb = new StringBuffer();
                sb.append("Received FIX message from ").append(message.getHeader().getField(new SenderCompID()).getValue());
                sb.append("\nMessage Sequence Number: ").append(message.getHeader().getField(new MsgSeqNum()).getValue());
                sb.append("\nReceiver ID: ").append(message.getHeader().getField(new TargetCompID()).getValue());
                log.debug(sb.toString());
                if (log.isTraceEnabled()) {
                    log.trace("Message: " + message.toString());
                }
            }

            AtomicInteger atomicCounter = countersMap.get(sessionID);
            int counter = atomicCounter.incrementAndGet();
            boolean rolled = atomicCounter.compareAndSet(FIXConstants.DEFAULT_COUNTER_UPPER_LIMIT, 0);
            if (rolled && log.isDebugEnabled()) {
                log.debug("Incoming request counter rolled over for the session: " + sessionID);
            }
            workerPool.execute(new FIXWorkerThread(message, sessionID, counter));
        }

        /**
         * This Runnable class can be used when it is required to process each incoming message
         * using separate threads.
         */
        class FIXWorkerThread implements Runnable {

            private Message message;
            private SessionID sessionID;
            private int counter;

            public FIXWorkerThread(Message message, SessionID sessionID, int counter) {
                this.message = message;
                this.sessionID = sessionID;
                this.counter = counter;
            }

            private void handleIncomingRequest() {
                if (log.isDebugEnabled()) {
                    log.debug("Source session: " + sessionID + " - Received message with sequence " +
                              "number " + counter);
                }

                //Create message context for the incoming message
                AbstractTransportListener trpListener = (AbstractTransportListener) cfgCtx.getAxisConfiguration().
                        getTransportIn(FIXConstants.TRANSPORT_NAME).getReceiver();

                MessageContext msgCtx = trpListener.createMessageContext();
                msgCtx.setProperty(Constants.OUT_TRANSPORT_INFO, new FIXOutTransportInfo(sessionID));

                if (service != null) {
                    // Set the service for which the message is intended to
                    msgCtx.setAxisService(service);
                    // find the operation for the message, or default to one
                    Parameter operationParam = service.getParameter(BaseConstants.OPERATION_PARAM);
                    QName operationQName = (
                            operationParam != null ?
                            BaseUtils.getQNameFromString(operationParam.getValue()) :
                            BaseConstants.DEFAULT_OPERATION);

                    AxisOperation operation = service.getOperation(operationQName);
                    if (operation != null) {
                        msgCtx.setAxisOperation(operation);
                        msgCtx.setAxisMessage(operation.getMessage(WSDLConstants.MESSAGE_LABEL_IN_VALUE));
                        msgCtx.setSoapAction("urn:" + operation.getName().getLocalPart());
                    }
                }

                String fixApplication = FIXConstants.FIX_INITIATOR;
                if (acceptor) {
                    fixApplication = FIXConstants.FIX_ACCEPTOR;
                } else {
                    msgCtx.setProperty("synapse.isresponse", true);
                }

                try {
                    //Put the FIX message in a SOAPEnvelope
                    FIXUtils.getInstance().setSOAPEnvelope(message, counter, sessionID.toString(), msgCtx);
                    trpListener.handleIncomingMessage(
                            msgCtx,
                            FIXUtils.getTransportHeaders(service.getName(), fixApplication),
                            null,
                            FIXConstants.FIX_DEFAULT_CONTENT_TYPE
                    );
                } catch (AxisFault e) {
                    handleException("Error while processing FIX message", e);
                }
            }

            private void handleIncomingResponse(MessageContext outMsgCtx) {
                AbstractTransportSender trpSender = (AbstractTransportSender) cfgCtx.getAxisConfiguration().
                        getTransportOut(FIXConstants.TRANSPORT_NAME).getSender();

                MessageContext msgCtx = trpSender.createResponseMessageContext(outMsgCtx);

                try {
                    //Put the FIX message in a SOAPEnvelope
                    FIXUtils.getInstance().setSOAPEnvelope(message, counter, sessionID.toString(), msgCtx);
                    msgCtx.setServerSide(true);
                    trpSender.handleIncomingMessage(
                            msgCtx,
                            FIXUtils.getTransportHeaders(service.getName(), FIXConstants.FIX_INITIATOR),
                            null,
                            FIXConstants.FIX_DEFAULT_CONTENT_TYPE
                    );
                } catch (AxisFault e) {
                    handleException("Error while processing response FIX message", e);
                }
            }

            public void run() {

                if (allNewApproach) {
                    //treat all messages (including responses) as new messages
                    handleIncomingRequest();
                } else {
                    if (acceptor) {
                        //treat messages coming from an acceptor as new request messages
                        handleIncomingRequest();
                    } else {
                        MessageContext outMsgCtx = outgoingMessages.poll();
                        if (outMsgCtx != null) {
                            //handle as a response to an outgoing message
                            handleIncomingResponse(outMsgCtx);
                        } else if (!dropExtraResponses) {
                            //handle as a new request message
                            handleIncomingRequest();
                        } else {
                            log.debug("Dropping additional FIX response");
                        }
                    }
                }
            }

        }

    }