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.ArrayList;
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;

    public GuardianGroupImpl() {
        guardianList = new LinkedList<GuardianMember>();
    }

    @Property(name = "recovery_rules", required = true)
    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 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

        //Sends a message representing the exception to the other guardian members
        if (participantList == null) {
            for (GuardianMember g : guardianList) {
                if (!g.getParticipantIdentifier().equals(ex.getSignalingParticipant())) {
                    //g.gthrow(ex, participantList);
                    g.gthrow(null, null);
                }
            }
        } else {
            for (GuardianMember g : guardianList) {
                if (participantList.contains(g.getCurrentContext())) {
                    //g.gthrow(ex, participantList);
                    g.gthrow(null, null);
                }
            }
        }

        //Check if the participants are blocked
        List<Integer> flags = new ArrayList<Integer>(guardianList.size());
        for (int i = 0; i < guardianList.size(); i++) {
            if (guardianList.get(i).getService().isBlocked()) {
                flags.add(i, 1);
            } else {
                flags.add(i, 0);
            }
        }

        //Wait until all participants are blocked
        while (flags.contains(0)) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ex1) {
                Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
            }
        }

        //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);
//        }

        applyRecoveryRules(ex);

//        //HARDCODED - SERVER-BACKUP EXAMPLE
//        //Rule 1
//        if (ex instanceof JoinException) {
//            if (guardianMembers.size() > 1) {
//
//                PrimaryExistsException primaryExists = new PrimaryExistsException();
//                primaryExists.setTargetContext(new Context("MAIN"));
//
//                BackupJoinedException backupJoined = new BackupJoinedException();
//                backupJoined.setTargetContext(new Context("PRIMARY"));
//
//                for (GuardianMember g : guardianMembers) {
//                    //let p = JoinException.signaler
//                    if (g.getParticipantIdentifier().equals(ex.getSignalingParticipant())) {
//                        g.addException(primaryExists);
//                        System.out.println("adding PrimaryExistsException to " + g.getParticipantIdentifier());
//                    } else {
//                        g.addException(backupJoined);
//                        System.out.println("adding BackupJoinedException to " + g.getParticipantIdentifier());
//                    }
//                }
//            }
//        } //Rule 2
//        else if (ex instanceof PrimaryFailedException) {
//
//            PrimaryFailedException primaryFailedInit = new PrimaryFailedException();
//            primaryFailedInit.setTargetContext(Context.INIT_CONTEXT);
//
//            PrimaryFailedException primaryFailedMain = new PrimaryFailedException();
//            primaryFailedMain.setTargetContext(new Context("MAIN"));
//
//            for (GuardianMember g : guardianMembers) {
//                if (g.getCurrentContext().getName().equals("PRIMARY")) {
//                    System.out.println("adding PrimaryFailedException to " + g.getParticipantIdentifier());
//                    g.addException(primaryFailedInit);
//                } else if (g.getCurrentContext().getName().equals("BACKUP")) {
//                    System.out.println("adding PrimaryFailedException to " + g.getParticipantIdentifier());
//                    g.addException(primaryFailedMain);
//                }
//            }
//        } //Rule 3
//        else if (ex instanceof BackupFailedException) {
//
//            BackupFailedException backupFailedPrimary = new BackupFailedException();
//            backupFailedPrimary.setTargetContext(new Context("PRIMARY"));
//
//            BackupFailedException backupFailedInit = new BackupFailedException();
//            backupFailedInit.setTargetContext(Context.INIT_CONTEXT);
//
//            for (GuardianMember g : guardianMembers) {
//                if (g.getCurrentContext().getName().equals("PRIMARY")) {
//                    System.out.println("adding BackupFailedException to " + g.getParticipantIdentifier());
//                    g.addException(backupFailedPrimary);
//                } else if (g.getCurrentContext().getName().equals("BACKUP")) {
//                    System.out.println("adding BackupFailedException to " + g.getParticipantIdentifier());
//                    g.addException(backupFailedInit);
//                }
//            }
//
//        }

    }

    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());

                    //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 applyRecoveryRulesOriginal(GlobalExceptionInterface ex) {

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

        List<GuardianMember> gmList;

        try {

            reader.reset();

            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))) {

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

                                //<participant match="<REG_EXP> | SIGNALER">
                                if (reader.isStartElement() && reader.getLocalName().equals("participant")) {

                                    String participant_match = reader.getAttributeValue(0).trim();

                                    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;
                                                }
                                            }

                                            //match value
                                            gmList = getMatchingParticipants(participant_match, ex);

                                            String affectedParticipants = null;
                                            int index = -1;
                                            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();

                                                    //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);

                                            //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);
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                            break;
                        }
                    }
                }
            }
        } catch (XMLStreamException ex1) {
            Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
        } catch (Exception ex1) {
            Logger.getLogger(GuardianGroupImpl.class.getName()).log(Level.SEVERE, null, ex1);
        }
    }

    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();
    }

    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);
    }
}