/*
 *  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.eventing;

import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.axis2.AxisFault;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.AxisOperation;
import org.apache.axis2.description.AxisService;
import org.apache.axis2.description.InOutAxisOperation;
import org.apache.axis2.engine.AxisConfiguration;
import org.apache.axis2.engine.AxisEngine;
import org.apache.axis2.util.MessageContextBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.synapse.SynapseConstants;
import org.apache.synapse.SynapseException;
import org.apache.synapse.endpoints.Endpoint;
import org.apache.synapse.endpoints.AddressEndpoint;
import org.apache.synapse.endpoints.EndpointDefinition;
import org.apache.synapse.config.SynapseConfiguration;
import org.apache.synapse.core.SynapseEnvironment;
import org.apache.synapse.core.axis2.Axis2MessageContext;
import org.apache.synapse.core.axis2.SynapseMessageReceiver;
import org.apache.synapse.eventing.builders.ResponseMessageBuilder;
import org.apache.synapse.eventing.builders.SubscriptionMessageBuilder;
import org.apache.synapse.util.MessageHelper;
import org.wso2.eventing.EventingConstants;
import org.wso2.eventing.Subscription;
import org.wso2.eventing.Event;
import org.wso2.eventing.SubscriptionManager;
import org.wso2.eventing.exceptions.EventException;

import javax.xml.namespace.QName;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Eventsource that accepts the event requests using a message receiver.
 * Eventsource is responsible on two tasks accepting the subscriptions and subscription related
 * reqests and dispatching events.
 * Subscriptions contains operations listed in the WS-Eventing specification. {SubscribeOP,
 * UnsubscribeOP, RenewOP, GetstatusOP, SubscriptionEndOP}
 * based on the action in the request eventsource identify the operation and send it for processing.
 * Eventsource link with a subscription manager to store the subscriptions.
 */
public class SynapseEventSource extends SynapseMessageReceiver {

    private String name;
    private SubscriptionManager subscriptionManager;
    private static final Log log = LogFactory.getLog(SynapseEventSource.class);
    private String fileName;
    /* Contains properties used in the configuration and possess confidential information such as
     encrypted passwords  */
    private Map<String, String> configurationProperties = new HashMap<String, String>();

    public SynapseEventSource(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public SubscriptionManager getSubscriptionManager() {
        return subscriptionManager;
    }

    public void setSubscriptionManager(SubscriptionManager subscriptionManager) {
        this.subscriptionManager = subscriptionManager;
    }

    public String getFileName() {
        return fileName;
    }

    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    public void buildService(AxisConfiguration axisCfg) throws AxisFault {
        AxisService eventSourceService = new AxisService();
        eventSourceService.setName(this.name);
        // Add wse operations 
        addOperations(eventSourceService);
        axisCfg.addService(eventSourceService);
        //Set the service parameters
        eventSourceService
                .addParameter(EventingConstants.SUBSCRIPTION_MANAGER, subscriptionManager);
        eventSourceService.addParameter(SynapseEventingConstants.SERVICE_TYPE,
                SynapseEventingConstants.EVENTING_ST);
    }

    /**
     * Override the Message receiver method to accept subscriptions and events
     *
     * @param mc message context
     * @throws AxisFault
     */
    @Override
    public void receive(MessageContext mc) throws AxisFault {
        // Create synapse message context from the axis2 message context
        SynapseConfiguration synCfg = (SynapseConfiguration) mc.getConfigurationContext()
                .getAxisConfiguration().getParameter(SynapseConstants.SYNAPSE_CONFIG).getValue();
        SynapseEnvironment synEnv = (SynapseEnvironment) mc.getConfigurationContext()
                .getAxisConfiguration().getParameter(SynapseConstants.SYNAPSE_ENV).getValue();
        org.apache.synapse.MessageContext smc = new Axis2MessageContext(mc, synCfg, synEnv);
        // initialize the response message builder using the message context
        ResponseMessageBuilder messageBuilder = new ResponseMessageBuilder(mc);
        try {
            if (EventingConstants.WSE_SUBSCRIBE.equals(mc.getWSAAction())) {
                // add new subscription to the SynapseSubscription store through subscription manager
                processSubscriptionRequest(mc, messageBuilder);
            } else if (EventingConstants.WSE_UNSUBSCRIBE.equals(mc.getWSAAction())) {
                // Unsubscribe the matching subscription
                processUnSubscribeRequest(mc, messageBuilder);
            } else if (EventingConstants.WSE_GET_STATUS.equals(mc.getWSAAction())) {
                // Response with the status of the subscription
                processGetStatusRequest(mc, messageBuilder);
            } else if (EventingConstants.WSE_RENEW.equals(mc.getWSAAction())) {
                // Renew subscription
                processReNewRequest(mc, messageBuilder);
            } else {
                // Treat as an Event
                if (log.isDebugEnabled()) {
                    log.debug("Event received");
                }
                dispatchEvents(smc);
            }
        } catch (EventException e) {
            handleException("Subscription manager processing error", e);
        }
    }

    /**
     * Dispatch the message to the target endpoint
     *
     * @param soapEnvelope   Soap Enevlop with message
     * @param responseAction WSE action for the response
     * @param mc             Message Context
     * @param faultMessage   Fault message
     * @throws AxisFault     AxisFault
     */
    private void dispatchResponse(SOAPEnvelope soapEnvelope,
                                  String responseAction,
                                  MessageContext mc,
                                  boolean faultMessage) throws AxisFault {
        MessageContext rmc = MessageContextBuilder.createOutMessageContext(mc);
        rmc.getOperationContext().addMessageContext(rmc);
        rmc.setEnvelope(soapEnvelope);
        rmc.setWSAAction(responseAction);
        rmc.setSoapAction(responseAction);
        rmc.setProperty(SynapseConstants.ISRESPONSE_PROPERTY, Boolean.TRUE);
        if (faultMessage) {
            AxisEngine.sendFault(rmc);
        } else {
            AxisEngine.send(rmc);
        }
    }

    /**
     * Public method for event dispatching, used by the eventPublisher mediator and eventSource
     *
     * @param msgCtx message context
     */
    public void dispatchEvents(org.apache.synapse.MessageContext msgCtx) {

        List<Subscription> subscribers = null;

        // Call event dispatcher
        msgCtx.getEnvironment().getExecutorService()
                .execute(new EventDispatcher(msgCtx));
    }

    /**
     * Dispatching events async on a different thread
     */
    class EventDispatcher implements Runnable {
        private org.apache.synapse.MessageContext synCtx;
        private List<Subscription> subscriptions;

        EventDispatcher(org.apache.synapse.MessageContext synCtx) {
            this.synCtx = synCtx;
        }

        @Override
        public void run() {
            try {
                MessageContext msgCtx = ((Axis2MessageContext) synCtx).getAxis2MessageContext();
                Event<MessageContext> event = new Event(msgCtx);
                subscriptions = subscriptionManager.getMatchingSubscriptions(event);
            } catch (EventException e) {
                handleException("Matching subscriptions fetching error", e);
            }

            for (Subscription subscription : subscriptions) {
                synCtx.setProperty(SynapseConstants.OUT_ONLY,
                        "true");    // Set one way message for events
                try {
                    getEndpointFromURL(subscription.getEndpointUrl(), synCtx.getEnvironment())
                            .send(MessageHelper.cloneMessageContext(synCtx));
                } catch (AxisFault axisFault) {
                    log.error("Event sending failure " + axisFault.toString());
                }
                if (log.isDebugEnabled()) {
                    log.debug("Event push to  : " + subscription.getEndpointUrl());
                }
            }
        }
    }

    /**
     * Process the subscription message request
     *
     * @param mc             axis2 message context
     * @param messageBuilder respose message builder
     * @throws AxisFault     axis fault
     * @throws EventException eventing exception
     */
    private void processSubscriptionRequest(MessageContext mc,
                                            ResponseMessageBuilder messageBuilder)
            throws AxisFault, EventException {
        SynapseSubscription subscription = SubscriptionMessageBuilder.createSubscription(mc);
        if (log.isDebugEnabled()) {
            log.debug("SynapseSubscription request recived  : " + subscription.getId());
        }
        if (subscription.getId() != null) {
            String subID = subscriptionManager.subscribe(subscription);
            if (subID != null) {
                // Send the subscription responce
                if (log.isDebugEnabled()) {
                    log.debug("Sending subscription response for SynapseSubscription ID : " +
                            subscription.getId());
                }
                SOAPEnvelope soapEnvelope =
                        messageBuilder.genSubscriptionResponse(subscription);
                dispatchResponse(soapEnvelope, EventingConstants.WSE_SUbSCRIBE_RESPONSE,
                        mc, false);
            } else {
                // Send the Fault responce
                if (log.isDebugEnabled()) {
                    log.debug("SynapseSubscription Failed, sending fault response");
                }
                SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                        EventingConstants.WSE_FAULT_CODE_RECEIVER, "EventSourceUnableToProcess",
                        "Unable to subscribe ", "");
                dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc,
                        true);
            }
        } else {
            // Send the Fault responce
            if (log.isDebugEnabled()) {
                log.debug("SynapseSubscription Failed, sending fault response");
            }
            SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                                                                        subscription.getErrorCode(),
                                                                        subscription.getErrorSubCode(),
                                                                        subscription.getErrorReason(), "");
            dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc,
                    true);
        }
    }

    /**
     * Process the UnSubscribe message request
     *
     * @param mc             axis2 message context
     * @param messageBuilder respose message builder
     * @throws AxisFault     axis fault
     * @throws EventException eventing exception
     */
    private void processUnSubscribeRequest(MessageContext mc,
                                           ResponseMessageBuilder messageBuilder)
            throws AxisFault, EventException {
        SynapseSubscription subscription =
                SubscriptionMessageBuilder.createUnSubscribeMessage(mc);
        if (log.isDebugEnabled()) {
            log.debug("UnSubscribe response recived for SynapseSubscription ID : " +
                    subscription.getId());
        }
        if (subscriptionManager.unsubscribe(subscription.getId())) {
            //send the response
            if (log.isDebugEnabled()) {
                log.debug("Sending UnSubscribe responce for SynapseSubscription ID : " +
                        subscription.getId());
            }
            SOAPEnvelope soapEnvelope = messageBuilder.genUnSubscribeResponse(subscription);
            dispatchResponse(soapEnvelope, EventingConstants.WSE_UNSUBSCRIBE_RESPONSE,
                    mc, false);
        } else {
            // Send the Fault responce
            if (log.isDebugEnabled()) {
                log.debug("UnSubscription failed, sending fault repsponse");
            }
            SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                    EventingConstants.WSE_FAULT_CODE_RECEIVER, "EventSourceUnableToProcess",
                    "Unable to Unsubscribe", "");
            dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc,
                    true);
        }
    }

    /**
     * Process the GetStatus message request
     *
     * @param mc             axis2 message context
     * @param messageBuilder respose message builder
     * @throws AxisFault     axis fault
     * @throws EventException event
     */
    private void processGetStatusRequest(MessageContext mc,
                                         ResponseMessageBuilder messageBuilder)
            throws AxisFault, EventException {
        Subscription subscription =
                SubscriptionMessageBuilder.createGetStatusMessage(mc);
        if (log.isDebugEnabled()) {
            log.debug("GetStatus request recived for SynapseSubscription ID : " +
                    subscription.getId());
        }
        subscription = subscriptionManager.getSubscription(subscription.getId());
        if (subscription != null) {
            if (log.isDebugEnabled()) {
                log.debug("Sending GetStatus responce for SynapseSubscription ID : " +
                        subscription.getId());
            }
            //send the responce
            SOAPEnvelope soapEnvelope = messageBuilder.genGetStatusResponse(subscription);
            dispatchResponse(soapEnvelope, EventingConstants.WSE_GET_STATUS_RESPONSE,
                    mc, false);
        } else {
            // Send the Fault responce
            if (log.isDebugEnabled()) {
                log.debug("GetStatus failed, sending fault response");
            }
            SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                    EventingConstants.WSE_FAULT_CODE_RECEIVER, "EventSourceUnableToProcess",
                    "Subscription Not Found", "");
            dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc,
                    true);
        }
    }

    /**
     * Process the ReNew message request
     *
     * @param mc             axis2 message context
     * @param messageBuilder respose message builder
     * @throws AxisFault axis fault
     * @throws EventException event exception
     */
    private void processReNewRequest(MessageContext mc,
                                     ResponseMessageBuilder messageBuilder)
            throws AxisFault, EventException {
        SynapseSubscription subscription =
                SubscriptionMessageBuilder.createRenewSubscribeMessage(mc);
        if (log.isDebugEnabled()) {
            log.debug("ReNew request recived for SynapseSubscription ID : " +
                    subscription.getId());
        }
        String subID = subscription.getId();
        if (subID != null) {
            if (subscriptionManager.renew(subscription)) {
                //send the response
                if (log.isDebugEnabled()) {
                    log.debug("Sending ReNew response for SynapseSubscription ID : " +
                            subscription.getId());
                }
                SOAPEnvelope soapEnvelope =
                        messageBuilder.genRenewSubscriptionResponse(subscription);
                dispatchResponse(soapEnvelope, EventingConstants.WSE_RENEW_RESPONSE,
                        mc, false);
            } else {
                // Send the Fault responce
                if (log.isDebugEnabled()) {
                    log.debug("ReNew failed, sending fault response");
                }
                SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                        EventingConstants.WSE_FAULT_CODE_RECEIVER, "UnableToRenew",
                        "Subscription Not Found", "");
                dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc,
                        true);
            }
        } else {
            SOAPEnvelope soapEnvelope = messageBuilder.genFaultResponse(mc,
                                                                        subscription.getErrorCode(),
                                                                        subscription.getErrorSubCode(),
                                                                        subscription.getErrorReason(), "");
            dispatchResponse(soapEnvelope, EventingConstants.WSA_FAULT, mc, true);
        }
    }

    /**
     * Create a Endpoint for a given URL
     *
     * @param endpointUrl      URL
     * @param se    synapse environment
     * @return AddressEndpoint address endpoint
     */
    private Endpoint getEndpointFromURL(String endpointUrl, SynapseEnvironment se) {
        AddressEndpoint endpoint = new AddressEndpoint();
        EndpointDefinition def = new EndpointDefinition();
        def.setAddress(endpointUrl.trim());
        endpoint.setDefinition(def);

        endpoint.init(se);

        return endpoint;
    }

    /**
     * Set the operations avilable for EventSource service
     *
     * @param eventSourceService service
     * @throws AxisFault         axis fault
     */
    private void addOperations(AxisService eventSourceService) throws AxisFault {
        // Create operations
        AxisOperation mediateOperation =
                new InOutAxisOperation(SynapseConstants.SYNAPSE_OPERATION_NAME);
        AxisOperation subscribeOperation =
                new InOutAxisOperation(new QName(EventingConstants.WSE_SUBSCRIBE_OP));
        AxisOperation unsubscribeOperation =
                new InOutAxisOperation(new QName(EventingConstants.WSE_UNSUBSCRIBE_OP));
        AxisOperation renewOperation =
                new InOutAxisOperation(new QName(EventingConstants.WSE_RENEW_OP));
        AxisOperation getStatusOperation =
                new InOutAxisOperation(new QName(EventingConstants.WSE_GET_STATUS_OP));
        AxisOperation subscriptionEndOperation =
                new InOutAxisOperation(new QName(EventingConstants.WSE_SUBSCRIPTIONEND_OP));
        // Assign the message reciver
        mediateOperation.setMessageReceiver(this);
        subscribeOperation.setMessageReceiver(this);
        unsubscribeOperation.setMessageReceiver(this);
        renewOperation.setMessageReceiver(this);
        getStatusOperation.setMessageReceiver(this);
        subscriptionEndOperation.setMessageReceiver(this);        
        // Set Soap Action
        subscribeOperation.setSoapAction(EventingConstants.WSE_SUBSCRIBE);
        unsubscribeOperation.setSoapAction(EventingConstants.WSE_UNSUBSCRIBE);
        renewOperation.setSoapAction(EventingConstants.WSE_RENEW);
        getStatusOperation.setSoapAction(EventingConstants.WSE_GET_STATUS);
        // Add operations to the Service
        eventSourceService.addOperation(mediateOperation);
        eventSourceService.addOperation(subscribeOperation);
        eventSourceService.addOperation(unsubscribeOperation);
        eventSourceService.addOperation(renewOperation);
        eventSourceService.addOperation(getStatusOperation);
        eventSourceService.addOperation(subscriptionEndOperation);
        
    }

    // Methods for accessing configuration properties - self-explainable

    public void putConfigurationProperty(String name, String value) {
        configurationProperties.put(name, value);
    }

    public String getConfigurationProperty(String name) {
        return configurationProperties.get(name);
    }

    public boolean isContainsConfigurationProperty(String name) {
        return configurationProperties.containsKey(name);
    }

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