# FIXTransportSender.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.context.ConfigurationContext;
    import org.apache.axis2.context.MessageContext;
    import org.apache.axis2.description.AxisService;
    import org.apache.axis2.description.Parameter;
    import org.apache.axis2.description.TransportOutDescription;
    import org.apache.axis2.transport.OutTransportInfo;
    import org.apache.axis2.transport.base.AbstractTransportSender;
    import org.apache.axis2.transport.base.BaseUtils;
    import org.apache.axis2.transport.base.threads.WorkerPool;
    import org.apache.axis2.transport.base.threads.WorkerPoolFactory;
    import org.apache.commons.logging.LogFactory;
    import quickfix.*;
    import quickfix.field.*;

    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.Hashtable;
    import java.util.Map;

    /**
     * The FIX transport sender implementation. This implementation looks at the SOAPBody of the message
     * context to identify how the message was first received by Axis2 engine and also looks at some FIX
     * header fields to make the optimum routing decision.
     * <p/>
     * This transport sender implementation does not support forwarding FIX messages to sessions with
     * different BeginString values.When it performs a message forwarding it makes sure the forwarding
     * takes place according to the conditions specified in the 'Third Party Routing' section in the
     * FIX protocol specification.
     */
    public class FIXTransportSender extends AbstractTransportSender {

        private FIXSessionFactory sessionFactory;
        private FIXOutgoingMessageHandler messageSender;
        private WorkerPool workerPool;

        public FIXTransportSender() {
            this.log = LogFactory.getLog(this.getClass());
        }

        /**
         * @param cfgCtx       the axis2 configuration context
         * @param transportOut the Out Transport description
         * @throws AxisFault on error
         */
        public void init(ConfigurationContext cfgCtx, TransportOutDescription transportOut) throws AxisFault {
            super.init(cfgCtx, transportOut);
            this.sessionFactory = FIXSessionFactory.getInstance(new FIXApplicationFactory(cfgCtx));
            this.workerPool = WorkerPoolFactory.getWorkerPool(
                                10, 20, 5, -1, "FIX Sender Worker thread group", "FIX-Worker");
            this.sessionFactory.setSenderThreadPool(this.workerPool);
            messageSender = new FIXOutgoingMessageHandler();
            messageSender.setSessionFactory(this.sessionFactory);
            log.info("FIX transport sender initialized...");
        }

        public void stop() {
            try {
                this.workerPool.shutdown(10000);
            } catch (InterruptedException e) {
                log.warn("Thread interrupted while waiting for worker pool to shut down");
            }
            sessionFactory.disposeFIXInitiators();
            super.stop();
        }

        /**
         * Performs the actual sending of the message.
         *
         * @param msgCtx           the axis2 message context of the message to be sent
         * @param targetEPR        the EPR for which the message is to be sent
         * @param outTransportInfo the OutTransportInfo for the message
         * @throws AxisFault on error
         */
        public void sendMessage(MessageContext msgCtx, String targetEPR,
                                OutTransportInfo outTransportInfo) throws AxisFault {

            if (log.isDebugEnabled()) {
                log.debug("Attempting to send a FIX message, Message ID:" + msgCtx.getMessageID());
            }
            Message fixMessage = null;
            String serviceName = FIXUtils.getServiceName(msgCtx);
            String fixApplication = FIXUtils.getFixApplication(msgCtx);
            String sourceSession = FIXUtils.getSourceSession(msgCtx);
            int counter = FIXUtils.getSequenceNumber(msgCtx);

            try {
                fixMessage = FIXUtils.getInstance().createFIXMessage(msgCtx);
            } catch (IOException e) {
                handleException("Exception occurred while creating the FIX message from SOAP Envelope", e);
            }

            if (FIXConstants.FIX_ACCEPTOR.equals(fixApplication)) {
                //A message came in through an acceptor bound to a service
                if (targetEPR != null) {
                    //Forward the message to the given EPR
                    sendUsingEPR(targetEPR, serviceName, fixMessage, sourceSession, counter, msgCtx);
                } else if (outTransportInfo != null && outTransportInfo instanceof FIXOutTransportInfo) {
                    //Send the message back to the sender
                    sendUsingTrpOutInfo(outTransportInfo, serviceName, fixMessage,
                            sourceSession, counter, msgCtx);
                }
            } else if (FIXConstants.FIX_INITIATOR.equals(fixApplication)) {

                if (sendUsingAcceptorSession(serviceName, fixMessage, sourceSession, counter, msgCtx)) {
                    return;
                } else if (targetEPR != null) {
                    sendUsingEPR(targetEPR, serviceName, fixMessage, sourceSession, counter, msgCtx);
                    return;
                }
                handleException("Unable to find a session to send the message...");

            } else {
                //A message generated in Axis2 engine or a message arrived over a different transport
                if (targetEPR != null) {
                    sendUsingEPR(targetEPR, serviceName, fixMessage, sourceSession, counter, msgCtx);
                } else {
                    sendUsingAcceptorSession(serviceName, fixMessage, sourceSession, counter, msgCtx);
                }
            }
        }

        private boolean isTargetValid(Map<String, String> fieldValues, SessionID targetSession,
                                     boolean beginStrValidation) {

            String beginString = fieldValues.get(FIXConstants.BEGIN_STRING);
            String deliverToCompID = fieldValues.get(FIXConstants.DELIVER_TO_COMP_ID);
            String deliverToSubID = fieldValues.get(FIXConstants.DELIVER_TO_SUB_ID);
            String deliverToLocationID = fieldValues.get(FIXConstants.DELIVER_TO_LOCATION_ID);

            if (beginStrValidation && !targetSession.getBeginString().equals(beginString)) {
                return false;
            } else if (!targetSession.getTargetCompID().equals(deliverToCompID)) {
                return false;
            } else if (deliverToSubID != null && !deliverToSubID.equals(targetSession.getTargetSubID())) {
                return false;
            } else if (deliverToLocationID != null && !deliverToLocationID.equals(targetSession.getTargetLocationID())) {
                return false;
            }
            return true;
        }

        /**
         * Prepares the message to be forwarded according to the conditions specified in the FIX protocol
         * specification.
         *
         * @param message     the FIX message to be forwarded
         * @param fieldValues a Map of field values for quick access
         */
        private void prepareToForwardMessage(Message message, Map<String, String> fieldValues) {
            //set OnBehalfOf* fields
            message.getHeader().setField(new OnBehalfOfCompID(fieldValues.get(FIXConstants.SENDER_COMP_ID)));
            if (fieldValues.get(FIXConstants.SENDER_SUB_ID) != null) {
                message.getHeader().setField(new OnBehalfOfSubID(fieldValues.get(FIXConstants.SENDER_SUB_ID)));
            }

            if (fieldValues.get(FIXConstants.SENDER_LOCATION_ID) != null) {
                message.getHeader().setField(new OnBehalfOfLocationID(fieldValues.get(FIXConstants.SENDER_LOCATION_ID)));
            }

            //remove additional Sender* fields and DeliverTo* fields
            message.getHeader().removeField(SenderSubID.FIELD);
            message.getHeader().removeField(SenderLocationID.FIELD);
            message.getHeader().removeField(DeliverToCompID.FIELD);
            message.getHeader().removeField(DeliverToSubID.FIELD);
            message.getHeader().removeField(DeliverToLocationID.FIELD);
        }

        /**
         * Puts DeliverToX fields in the message to enable the message to be forwarded at the destination.
         *
         * @param message   the FIX message to be forwarded
         * @param targetEPR the EPR to which the message will be sent
         */
        private void setDeliverToXFields(Message message, String targetEPR) {
            Hashtable<String, String> properties = BaseUtils.getEPRProperties(targetEPR);
            String deliverTo = properties.get(FIXConstants.DELIVER_TO_COMP_ID);
            //If a DeliverToCompID field is given in EPR put the field in the message
            if (deliverTo != null) {
                message.getHeader().setField(new DeliverToCompID(deliverTo));
                deliverTo = properties.get(FIXConstants.DELIVER_TO_SUB_ID);
                if (deliverTo != null) {
                    message.getHeader().setField(new DeliverToSubID(deliverTo));
                }

                deliverTo = properties.get(FIXConstants.DELIVER_TO_LOCATION_ID);
                if (deliverTo != null) {
                    message.getHeader().setField(new DeliverToLocationID(deliverTo));
                }
            }
        }

        /**
         * Puts DeliverToX fields in the message to enable the message to be forwarded at the destination.
         *
         * @param message     the FIX message to be forwarded
         * @param fieldValues the Map of field values for quick access
         */
        private void setDeliverToXFields(Message message, Map<String, String> fieldValues) {
            //Use the fields of the message to set DeliverToX fields
            String onBehalf = fieldValues.get(FIXConstants.ON_BEHALF_OF_COMP_ID);
            if (onBehalf != null) {
                message.getHeader().setField(new DeliverToCompID(onBehalf));
                onBehalf = fieldValues.get(FIXConstants.ON_BEHALF_OF_SUB_ID);
                if (onBehalf != null) {
                    message.getHeader().setField(new DeliverToSubID(onBehalf));
                }

                onBehalf = fieldValues.get(FIXConstants.ON_BEHALF_OF_LOCATION_ID);
                if (onBehalf != null) {
                    message.getHeader().setField(new DeliverToLocationID(onBehalf));
                }

                message.getHeader().removeField(OnBehalfOfCompID.FIELD);
                message.getHeader().removeField(OnBehalfOfSubID.FIELD);
                message.getHeader().removeField(OnBehalfOfLocationID.FIELD);
            }
        }

        /**
         * Puts DeliverToX fields in the message to enable the message to be forwarded at the destination.
         * This method retrieves the parameters from the services.xml and put them in the message as
         * DeliverToX fields. Should be used when a response message has to forwarded at the destination.
         *
         * @param message the FIX message to be forwarded
         * @param service the AxisService of the message
         */
        private void setDeliverToXFields(Message message, AxisService service) {
            Parameter param = service.getParameter(FIXConstants.FIX_RESPONSE_DELIVER_TO_COMP_ID_PARAM);
            if (param != null) {
                message.getHeader().setField(new DeliverToCompID(param.getValue().toString()));
                param = service.getParameter(FIXConstants.FIX_RESPONSE_DELIVER_TO_SUB_ID_PARAM);
                if (param != null) {
                    message.getHeader().setField(new DeliverToSubID(param.getValue().toString()));
                }

                param = service.getParameter(FIXConstants.FIX_RESPONSE_DELIVER_TO_LOCATION_ID_PARAM);
                if (param != null) {
                    message.getHeader().setField(new DeliverToLocationID(param.getValue().toString()));
                }
            }
        }

        /**
         * Sends a FIX message to the given EPR
         *
         * @param targetEPR   the EPR to which the message is sent to
         * @param serviceName name of the service which processed the message
         * @param fixMessage  the FIX message
         * @param srcSession  String uniquely identifying the incoming session
         * @param counter     application level sequence number of the message
         * @param msgCtx      the Axis2 MessageContext for the message
         * @return boolean value indicating the result
         * @throws AxisFault on error
         */
        private boolean sendUsingEPR(String targetEPR, String serviceName, Message fixMessage,
                                     String srcSession, int counter, MessageContext msgCtx) throws AxisFault {

            FIXOutTransportInfo fixOut = new FIXOutTransportInfo(targetEPR);
            SessionID sessionID = fixOut.getSessionID();
            Map<String, String> fieldValues = FIXUtils.getMessageForwardingParameters(fixMessage);
            String beginString = fieldValues.get(FIXConstants.BEGIN_STRING);
            String deliverToCompID = fieldValues.get(FIXConstants.DELIVER_TO_COMP_ID);

            AxisService service = cfgCtx.getAxisConfiguration().getService(serviceName);

            //match BeginString values
            if (isValidationOn(service) && beginString != null && !beginString.equals(sessionID.getBeginString())) {
                handleException("BeginString validation is on. Cannot forward messages to a session" +
                        " with a different BeginString");
             }

            if (deliverToCompID != null) {
                //message needs to be delivered
                if (!deliverToCompID.equals(sessionID.getTargetCompID())) {
                    handleException("Cannot forward messages that do not have a valid DeliverToCompID field");
                } else {
                    prepareToForwardMessage(fixMessage, fieldValues);
                    setDeliverToXFields(fixMessage, targetEPR);
                }
            }

            //synchronize this block to prevent multiple threads trying to creating individual sessions
            synchronized (this) {
                if (!Session.doesSessionExist(sessionID)) {
                    //try to create initiator to send the message
                    sessionFactory.createFIXInitiator(targetEPR, service, sessionID);
                }
            }

            try {
                messageSender.sendMessage(fixMessage, sessionID, srcSession, counter, msgCtx, targetEPR);
                return true;
            } catch (SessionNotFound e) {
                log.error("Error while sending the FIX message. Session " + sessionID.toString() + " does" +
                        " not exist", e);
                return false;
            }
        }

        /**
         * Sends a FIX message using the SessionID in the OutTransportInfo
         *
         * @param trpOutInfo  the TransportOutInfo for the message
         * @param fixMessage  the FIX message to be sent
         * @param srcSession  String uniquely identifying the incoming session
         * @param counter     application level sequence number of the message
         * @param serviceName name of the AxisService for the message
         * @param msgCtx      Axis2 MessageContext
         * @return boolean value indicating the result
         * @throws AxisFault on error
         */
        private boolean sendUsingTrpOutInfo(OutTransportInfo trpOutInfo, String serviceName,
                                            Message fixMessage, String srcSession, int counter,
                                            MessageContext msgCtx) throws AxisFault {

            FIXOutTransportInfo fixOut = (FIXOutTransportInfo) trpOutInfo;
            SessionID sessionID = fixOut.getSessionID();
            Map<String, String> fieldValues = FIXUtils.getMessageForwardingParameters(fixMessage);
            String beginString = fieldValues.get(FIXConstants.BEGIN_STRING);
            String deliverToCompID = fieldValues.get(FIXConstants.DELIVER_TO_COMP_ID);

            AxisService service = cfgCtx.getAxisConfiguration().getService(serviceName);

            //match BeginString values
            if (isValidationOn(service) && beginString != null && !beginString.equals(sessionID.getBeginString())) {
                handleException("BeginString validation is on. Cannot forward messages to a session" +
                        " with a different BeginString");
             }

            if (deliverToCompID != null) {
                //message needs to be delivered to some other party
                if (!deliverToCompID.equals(sessionID.getTargetCompID())) {
                    handleException("Cannot forward messages that do not have a valid DeliverToCompID field");
                } else {
                    prepareToForwardMessage(fixMessage, fieldValues);
                    setDeliverToXFields(fixMessage, service);
                }
            } else {
                setDeliverToXFields(fixMessage, fieldValues);
            }

            try {
                messageSender.sendMessage(fixMessage, sessionID, srcSession, counter, msgCtx, null);
                return true;
            } catch (SessionNotFound e) {
                log.error("Error while sending the FIX message. Session " + sessionID.toString() + " does" +
                        " not exist", e);
                return false;
            }
        }

        /**
         * Send the message using a session in the acceptor side
         *
         * @param serviceName the service of the message
         * @param fixMessage  the FIX message to be sent
         * @param srcSession  String uniquely identifying the incoming session
         * @param counter     the application level sequence number of the message
         * @param msgCtx      Axi2 MessageContext
         * @return boolean value indicating the result
         * @throws AxisFault on error
         */
        private boolean sendUsingAcceptorSession(String serviceName, Message fixMessage, String srcSession,
                                                 int counter, MessageContext msgCtx) throws AxisFault {

            Map<String, String> fieldValues = FIXUtils.getMessageForwardingParameters(fixMessage);
            String deliverToCompID = fieldValues.get(FIXConstants.DELIVER_TO_COMP_ID);

            Acceptor acceptor = sessionFactory.getAcceptor(serviceName);
            SessionID sessionID = null;

            AxisService service = cfgCtx.getAxisConfiguration().getService(serviceName);

            if (acceptor != null) {
                ArrayList<SessionID> sessions = acceptor.getSessions();
                if (sessions.size() == 1) {
                    sessionID = sessions.get(0);
                    if (deliverToCompID != null && !isTargetValid(fieldValues, sessionID, isValidationOn(service))) {
                        sessionID = null;
                    }

                } else if (sessions.size() > 1 && deliverToCompID != null) {
                    for (SessionID session : sessions) {
                        sessionID = session;
                        if (isTargetValid(fieldValues, sessionID, isValidationOn(service))) {
                            break;
                        }
                    }
                }
            }

            if (sessionID != null) {
                //Found a valid session. Now forward the message...
                FIXOutTransportInfo fixOutInfo = new FIXOutTransportInfo(sessionID);
                return sendUsingTrpOutInfo(fixOutInfo, serviceName, fixMessage,
                        srcSession, counter, msgCtx);
            }
            return false;
        }

        /**
         * Checks whether BeginString validation is on for the specified
         * service.
         *
         * @param service the AxisService of the message
         * @return a boolean value indicating the validation state
         */
        private boolean isValidationOn(AxisService service) {
            Parameter validationParam = service.getParameter(FIXConstants.FIX_BEGIN_STRING_VALIDATION);
            if (validationParam != null) {
                if ("true".equals(validationParam.getValue().toString())) {
                    return true;
                }
            }
            return false;
        }

        public void logOutIncomingSession(SessionID sessionID) {
            messageSender.cleanUpMessages(sessionID.toString());
        }

    }