path: root/sandbox/dougsleite/guardian-model/src/main/java/org/apache/tuscany/sca/guardian/GuardianGroupImpl.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.tuscany.sca.guardian;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.LinkedList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamReader;
import org.apache.axis2.jaxws.message.util.ResettableReader;
import org.osoa.sca.annotations.Property;
import org.osoa.sca.annotations.Scope;
import org.osoa.sca.annotations.Service;

@Service(GuardianGroup.class)
@Scope("COMPOSITE")
public class GuardianGroupImpl implements GuardianGroup {

    private List<GuardianMember> guardianList;
    private ResettableReader reader;
    private InnerGuardianGroupThread innerThread;
    private List<GlobalExceptionInterface> concurrentExList;

    public GuardianGroupImpl() {
        guardianList = new LinkedList<GuardianMember>();
        concurrentExList = new LinkedList<GlobalExceptionInterface>();
        innerThread = new InnerGuardianGroupThread();
    }

    @Property(name = "recovery_rules", required = false)
    public void setRecoveryRules(String recoveryRules) {
        try {
            FileInputStream fileInputStream = new FileInputStream(recoveryRules);
            XMLStreamReader xmlReader = XMLInputFactory.newInstance().createXMLStreamReader(fileInputStream);
            reader = new ResettableReader(xmlReader);
        } catch (XMLStreamException ex) {
            Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex);
        } catch (FileNotFoundException ex) {
            Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public void addGuardianMember(GuardianMember guardianMember) {
        guardianList.add(guardianMember);
        guardianMember.setUniqueParticipantID(guardianList.size() - 1);
    }

    public void enableContext(Context context) {
        System.out.println("Enable Context.. nothing to do!");
        //throw new UnsupportedOperationException("Not supported yet.");
    }

    public void removeContext() {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public synchronized void gthrow(GlobalExceptionInterface ex, List<String> participantList) {

        //1)Ivoked by a GuardianMember instance through the gthrow method
        //2)Notify all participants about the exception (FIFO atomic broadcast model) - it will cause the suspension of all participants
        //  2.1)Invoke the gthrow of the guardian members associated with the participants on participantList

        concurrentExList.add(ex);

        //Sends a message representing the exception to the other guardian members
        SuspendException suspendEx = new SuspendException();
        if (participantList == null) {
            for (GuardianMember g : guardianList) {
                g.gthrow(suspendEx, null);
            }
        } else {
            for (GuardianMember g : guardianList) {
                if (participantList.contains(g.getParticipantIdentifier())) {
                    g.gthrow(suspendEx, null);
                    //g.gthrow(suspendEx,null); -> g is the signler
                }
            }
        }

//        //3)Once ALL required participants are SUSPENDED (suspended point), invoke the defined Recovery Rules
//        for (GuardianMember g : guardianList) {
//            g.addException(new SuspendException());
//        }
        //  3.1) recovery_rules < signaled exceptions + context information of all participant > target context + exception to raise

        //Simpleste Recovery Rule: broadcast the exception for all participants - FIXME: hardcoded
//        for (GuardianMemberImpl g : guardianMembers) {
//            ex.setTargetContext(Context.CURRENT_CONTEXT);
//            g.addException(ex);
//        }

        if (!innerThread.isRunning) {
            innerThread.setGlobalException(ex);
            new Thread(innerThread).start();
        }
    }

    public boolean propagate(GlobalExceptionInterface ex) {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public void checkExceptionStatus() {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public boolean removeGuardianMember(GuardianMember guardianMember) {
        return this.guardianList.remove(guardianMember);
    }

    private class InnerGuardianGroupThread implements Runnable {

        private boolean isRunning;
        private GlobalExceptionInterface ex = null;

        public InnerGuardianGroupThread() {
            isRunning = false;
        }

        public void setGlobalException(GlobalExceptionInterface ex) {
            this.ex = ex;
        }

        public GlobalExceptionInterface getGlobalException() {
            return ex;
        }

        public void run() {
            if (ex != null) {
                isRunning = true;
                applyRecoveryRules(ex);

                for (GuardianMember gm : guardianList) {
                    if (gm.getParticipantState() == SUSPENDED_PARTICIPANT_STATE) {
                        gm.setParticipantState(NORMAL_PARTICIPANT_STATE);
                    }
                }
            }
            isRunning = false;
        }

        public boolean isRunning() {
            return isRunning;
        }

        private void applyRecoveryRules(GlobalExceptionInterface ex) {
            reader.reset();
            ruleTag(ex);
        }

        private void ruleTag(GlobalExceptionInterface ex) {
            try {
                while (reader.hasNext()) {
                    reader.next();
                    //<rule name="" signaled_exception="">
                    if (reader.isStartElement() && reader.getLocalName().equals("rule")) {
                        for (int i = 0; i < reader.getAttributeCount(); i++) {
                            //ex == signaled_exception
                            if (reader.getAttributeLocalName(i).equals("signaled_exception") && ex.getClass().getName().equals(reader.getAttributeValue(i))) {
                                participantExceptionTag(ex);
                                break;
                            }
                        }
                    }
                }
            } catch (XMLStreamException exc) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex);
            }
        }

        private void participantExceptionTag(GlobalExceptionInterface ex) {
            List<GuardianMember> gmList;
            try {
                while (reader.hasNext() && !(reader.isEndElement() && reader.getLocalName().equals("rule"))) {
                    reader.next();
                    //<participant match="<REG_EXP> | SIGNALER">
                    if (reader.isStartElement() && reader.getLocalName().equals("participant")) {
                        String participantMatch = reader.getAttributeValue(0).trim();

                        gmList = getMatchingParticipants(participantMatch, ex);

                        if (!gmList.isEmpty()) {
                            throwExceptionTag(gmList, ex);
                        }
                    }
                }
            } catch (XMLStreamException exc) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, exc);
            }
        }

        private void throwExceptionTag(List<GuardianMember> gmList, GlobalExceptionInterface ex) {

            String exceptionClassName;
            String targetContextName;
            Integer min_participant_joined;
            Integer max_participant_joined;

            try {

                while (reader.hasNext() && !(reader.isEndElement() && reader.getLocalName().equals("participant"))) {
                    reader.next();

                    //<throw_exception class="<Exception>" target_context="<Context>"/>
                    if (reader.isStartElement() && reader.getLocalName().equals("throw_exception")) {

                        exceptionClassName = null;
                        targetContextName = null;
                        min_participant_joined = null;
                        max_participant_joined = null;
                        for (int j = 0; j < reader.getAttributeCount(); j++) {
                            if (reader.getAttributeLocalName(j).equals("class")) {
                                //class value
                                exceptionClassName = reader.getAttributeValue(j);
                            } else if (reader.getAttributeLocalName(j).equals("target_context")) {
                                //target_context value
                                targetContextName = reader.getAttributeValue(j);
                            } else if (reader.getAttributeLocalName(j).equals("min_participant_joined")) {
                                //min_participant_joined value
                                min_participant_joined = Integer.parseInt(reader.getAttributeValue(j));
                            } else {
                                //max_participant_joined value
                                max_participant_joined = Integer.parseInt(reader.getAttributeValue(j));
                            }
                        }

                        //Test the min and max joined participants condition
                        if (min_participant_joined != null && max_participant_joined != null) {
                            if (!(guardianList.size() >= min_participant_joined && guardianList.size() < max_participant_joined)) {
                                break;
                            }
                        } else if (min_participant_joined != null) {
                            if (!(guardianList.size() >= min_participant_joined)) {
                                break;
                            }
                        } else if (max_participant_joined != null) {
                            if (!(guardianList.size() >= min_participant_joined)) {
                                break;
                            }
                        }

                        //<affected_participants>
                        String affectedParticipants = affectedParticipantsTag();
                        int index = -1;

                        //Verify if the parameter is an index
                        try {
                            index = Integer.parseInt(affectedParticipants);
                        } catch (NumberFormatException nexc) {
                            index = -1;
                        }

                        Class exceptionClass = Class.forName(exceptionClassName);
                        Context targetContext;
                        if (targetContextName.toUpperCase().equals(Context.CURRENT_CONTEXT.getName().toUpperCase())) {
                            targetContext = Context.CURRENT_CONTEXT;
                        } else if (targetContextName.toUpperCase().equals(Context.INIT_CONTEXT.getName().toUpperCase())) {
                            targetContext = Context.INIT_CONTEXT;
                        } else {
                            targetContext = new Context(targetContextName);
                        }
                        GlobalException newException = (GlobalException) exceptionClass.newInstance();

                        newException.setTargetContext(targetContext);
                        newException.setSignalingContext(ex.getSignalingContext());
                        newException.setSignalingParticipant(ex.getSignalingParticipant());

//                        //Time window of 10 seconds
//                        try {
//                            Thread.sleep(10000);
//                        } catch (InterruptedException ex1) {
//                            Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
//                        }
//                        //Check concurrent exception existence
//                        if (concurrentExList.size() != 1) {
//                            applyConcurrentRecoveryRules(concurrentExList);
//                        }

                        //Add the exception to the participants matched
                        if (index != -1) {
                            gmList.get(index).addException(newException);
                        } else if (affectedParticipants != null && affectedParticipants.length() != 0) {
                            if (affectedParticipants.toUpperCase().equals("FIRST")) {
                                gmList.get(0).addException(newException);
                            } else if (affectedParticipants.toUpperCase().equals("LAST")) {
                                gmList.get(gmList.size() - 1).addException(newException);
                            } else if (affectedParticipants.toUpperCase().equals("ALL")) {
                                for (GuardianMember gm : gmList) {
                                    gm.addException(newException);
                                }
                            }
                        } else {
                            for (GuardianMember gm : gmList) {
                                gm.addException(newException);
                            }
                        }
                    }
                }
            } catch (XMLStreamException ex1) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
            } catch (InstantiationException ex1) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
            } catch (IllegalAccessException ex1) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
            } catch (ClassNotFoundException ex1) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
            }
        }

        private String affectedParticipantsTag() {
            String affectedParticipants = null;
            try {
                while (reader.hasNext() && !(reader.isEndElement() && reader.getLocalName().equals("throw_exception"))) {
                    reader.next();
                    //<affected_participants>
                    if (reader.isStartElement() && reader.getLocalName().equals("affected_participants")) {
                        affectedParticipants = reader.getElementText();
                    }
                }
            } catch (XMLStreamException ex) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex);
            }

            return affectedParticipants;
        }

        private void applyConcurrentRecoveryRules(List<GlobalExceptionInterface> exceptionList) {
            System.out.println("CONCURRENT RECOVERY RULES!");
            System.out.println("exceptions: "+exceptionList);
            exceptionList.clear();
        }

        private List<GuardianMember> getMatchingParticipants(String regularExpression, GlobalExceptionInterface signaledException) {
            List<GuardianMember> matchingParticipants = new LinkedList();

            if (regularExpression.toUpperCase().equals("SIGNALER")) {
                for (GuardianMember gm : guardianList) {
                    if (gm.getParticipantIdentifier().equals(signaledException.getSignalingParticipant())) {
                        matchingParticipants.add(gm);
                        break;
                    }
                }
            } else if (regularExpression.toUpperCase().equals("!SIGNALER")) {
                for (GuardianMember gm : guardianList) {
                    if (!gm.getParticipantIdentifier().equals(signaledException.getSignalingParticipant())) {
                        matchingParticipants.add(gm);
                    }
                }

            } else {
                //Create an java regular expression
                String re = createJavaRegularExpression(regularExpression);

                for (GuardianMember gm : guardianList) {
                    if (gm.getParticipantIdentifier().matches(re)) {
                        matchingParticipants.add(gm);
                    }
                }
            }

            return matchingParticipants;
        }

        /* Valid expressions: *, <Context>.*, <Context>, *.<Context>, *.<Context>.*,
         *                    *.<Context>.*.<Context>.*, <REG_EXP> || <REG_EXP>
         *
         * Invalid expressions: *.*, **,
         *
         * Not supported yet: !<Context>, !<Context> || <Context>, !(<Context> || <Context>)
         */
        private String createJavaRegularExpression(String regularExpression) throws InvalidRegularExpression {
            StringBuffer re = new StringBuffer();

            String[] splitedByBar = regularExpression.split("\\|\\|");
            String[] splitedByPeriod;

            for (int i = 0; i < splitedByBar.length; i++) {

                splitedByPeriod = splitedByBar[i].split("\\.");

                if (i > 0) {
                    re.append("|");
                }

                re.append("^");
                for (int j = 0; j < splitedByPeriod.length; j++) {

                    //*
                    if (splitedByPeriod[j].equals("*")) {

                        //Validate the regular expression
                        if (j + 1 != splitedByPeriod.length && splitedByPeriod[j + 1].equals("*")) {
                            throw new InvalidRegularExpression();
                        }

                        //*
                        if (splitedByPeriod.length == 1) {
                            re.append("(\\w+)");
                        } //*.<CONTEXT>

                        if (j == 0 && splitedByPeriod.length != 1) {
                            re.append("(\\w+\\");
                            re.append(".)*");
                        } //<CONTEXT>.*
                        else {
                            re.append("(\\");
                            re.append(".\\w+)*");
                        }
                    } //<CONTEXT>
                    else {

//                    //Validate the regular expression
//                    if (splitedByPeriod[j].matches("^(\\*)*$")) {
//                        throw new RuntimeException("Invalid name for a context");
//                    }

                        //<CONTEXT> || <CONTEXT>.<CONTEXT>.<CONTEXT> || *.<CONTEXT>
                        if (splitedByPeriod.length == 1) {
                            re.append("(\\w+\\");
                            re.append(".)*");
                        }

                        if (j == 0 || j - 1 == 0) {
                            re.append("(" + splitedByPeriod[j] + ")");
                        } else {
                            re.append("(\\." + splitedByPeriod[j] + ")");
                        }
                    }
                }
                re.append("$");
            }
            return re.toString();
        }
    }
}