/* * 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 guardianList; private ResettableReader reader; public GuardianGroupImpl() { guardianList = new LinkedList(); } @Property(name = "recovery_rules", required = true) public void setRecoveryRules(String recoveryRules) { try { //FileInputStream fileInputStream = new FileInputStream("src/main/resources/recoveryrules.xml"); FileInputStream fileInputStream = new FileInputStream("src/main/resources/recoveryrules.xml"); 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) { throw new UnsupportedOperationException("Not supported yet."); } public void removeContext() { throw new UnsupportedOperationException("Not supported yet."); } public void gthrow(GlobalException ex, List 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 flags = new ArrayList(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(GlobalException ex) { String exceptionClassName; String targetContextName; Integer min_participant_joined; Integer max_participant_joined; List gmList; try { reader.reset(); while (reader.hasNext()) { reader.next(); // 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(); // 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(); // 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); 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 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 getMatchingParticipants(String regularExpression, GlobalException signaledException) { List 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: *, .*, , *., *..*, * *..*..*, || * * Invalid expressions: *.*, **, * * Not supported yet: !, ! || , !( || ) */ 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+)"); } //*. if (j == 0 && splitedByPeriod.length != 1) { re.append("(\\w+\\"); re.append(".)*"); } //.* else { re.append("(\\"); re.append(".\\w+)*"); } } // else { // //Validate the regular expression // if (splitedByPeriod[j].matches("^(\\*)*$")) { // throw new RuntimeException("Invalid name for a 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(GlobalException 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); } }