airavata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chath...@apache.org
Subject [05/13] airavata git commit: retiring ws-messenger and remove dependency of workflow tracking - AIRAVATA-1556, AIRAVATA-1557
Date Tue, 27 Jan 2015 15:26:59 GMT
http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionManager.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionManager.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionManager.java
deleted file mode 100644
index e42e4c1..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionManager.java
+++ /dev/null
@@ -1,440 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.subscription;
-
-import java.io.StringReader;
-import java.util.AbstractMap;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
-
-import javax.xml.stream.XMLInputFactory;
-import javax.xml.stream.XMLStreamException;
-import javax.xml.stream.XMLStreamReader;
-
-import org.apache.airavata.wsmg.broker.context.ContextParameters;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.broker.context.ProcessingContextBuilder;
-import org.apache.airavata.wsmg.broker.wseventing.WSEProcessingContextBuilder;
-import org.apache.airavata.wsmg.broker.wseventing.WSEProtocolSupport;
-import org.apache.airavata.wsmg.broker.wsnotification.WSNTProtocolSupport;
-import org.apache.airavata.wsmg.broker.wsnotification.WSNotificationProcessingContextBuilder;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.commons.storage.WsmgStorage;
-import org.apache.airavata.wsmg.config.WSMGParameter;
-import org.apache.airavata.wsmg.config.WsmgConfigurationContext;
-import org.apache.airavata.wsmg.matching.AbstractMessageMatcher;
-import org.apache.airavata.wsmg.messenger.OutGoingQueue;
-import org.apache.airavata.wsmg.util.RunTimeStatistics;
-import org.apache.axiom.om.OMElement;
-import org.apache.axiom.om.impl.builder.StAXOMBuilder;
-import org.apache.axis2.AxisFault;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Manages subscribers.
- * 
- */
-public class SubscriptionManager {
-
-    private static final Logger log = LoggerFactory.getLogger(SubscriptionManager.class);
-
-    private HashMap<String, SubscriptionState> subscriptions = new HashMap<String, SubscriptionState>();
-
-    private ReentrantReadWriteLock subscriptionLock = new ReentrantReadWriteLock();
-
-    private WSEProtocolSupport wseProtocalSupport = new WSEProtocolSupport();
-
-    private WSNTProtocolSupport wsntProtocolSupport = new WSNTProtocolSupport();
-
-    private WsmgStorage subscriptionDB;
-
-    private WsmgConfigurationContext wsmgConfig;
-
-    private OutGoingQueue outGoingQueue;
-
-    private int counter = 1;
-
-    public SubscriptionManager(WsmgConfigurationContext paramters, WsmgStorage storage) {
-        init(paramters, storage);
-    }
-
-    private void init(WsmgConfigurationContext parameters, WsmgStorage storage) {
-
-        this.wsmgConfig = parameters;
-
-        subscriptionDB = storage;
-        outGoingQueue = parameters.getOutgoingQueue();
-        if (WSMGParameter.enableAutoCleanSubscriptions) {
-            CleanUpThread cleanUpThread = new CleanUpThread(this);
-            Thread t = new Thread(cleanUpThread);
-            t.start();
-        }
-
-        try {
-            checkSubscriptionDB(storage);
-        } catch (AxisFault e) {
-            log.error("Subscription database has malformed" + " subscriptions. Ignoring them.", e);
-
-        }
-
-    }
-
-    /**
-     * @return Returns the subscriptions.
-     */
-    public AbstractMap<String, SubscriptionState> getShallowSubscriptionsCopy() {
-
-        AbstractMap<String, SubscriptionState> ret = null;
-        readLockUnlockSubscriptions(true);
-        try {
-            ret = new HashMap<String, SubscriptionState>(subscriptions);
-        } finally {
-            readLockUnlockSubscriptions(false);
-        }
-
-        return ret;
-
-    }
-
-    public void subscribe(ProcessingContext ctx) throws AxisFault {
-
-        String subId = createSubscription(null, ctx);
-        if (subId == null) {
-            log.error("ERROR: No subscription created");
-            return;
-        }
-
-        if (NameSpaceConstants.WSE_NS.equals(ctx.getContextParameter(ContextParameters.SUBSCRIBE_ELEMENT)
-                .getNamespace())) {
-            wseProtocalSupport.createSubscribeResponse(ctx, subId);
-
-        } else { // WSNT
-
-            wsntProtocolSupport.createSubscribeResponse(ctx, subId);
-        }
-    }
-
-    /**
-     * @param subscriptionId
-     *            this is the ID that is in the SOAP header
-     * @param ctx
-     *            contexts constructed with the body elements
-     * @return subscription id
-     * @throws AxisFault
-     */
-    private String createSubscription(String subscriptionId, ProcessingContext ctx) throws AxisFault {
-
-        SubscriptionState state = null;
-        String key = null;
-
-        // get the first element element inside the soap body element and check
-        // whether namespace is WSE
-        if (NameSpaceConstants.WSE_NS.equals(ctx.getContextParameter(ContextParameters.SUBSCRIBE_ELEMENT)
-                .getNamespace())) {
-            state = wseProtocalSupport.createSubscriptionState(ctx, outGoingQueue);
-        } else { // Handle WSNT
-
-            state = wsntProtocolSupport.createSubscriptionState(ctx, outGoingQueue);
-        }
-
-        if (subscriptionId == null) { // New subscription entry
-            key = checkSubscriptionExist(state);
-            if (key != null) { // just renew previous subscriptions
-                return key;
-            }
-            // new subscriptions
-
-            state.setCreationTime(System.currentTimeMillis());
-
-            key = generateSubscriptionId(state.getXpathString() != null && state.getXpathString().length() > 0);
-
-        } else { // Startup from previous subscription database
-            key = subscriptionId;
-        }
-
-        for (AbstractMessageMatcher m : wsmgConfig.getMessageMatchers()) {
-            m.handleSubscribe(state, key);
-        }
-
-        if (subscriptionId == null) { // New subscription entry,
-
-            RunTimeStatistics.totalSubscriptions++;
-            try {
-                String subscribeXml = ctx.getContextParameter(ContextParameters.SUBSCRIBE_ELEMENT)
-                        .toStringWithConsume();
-
-                state.setId(key);
-                state.setSubscribeXml(subscribeXml);
-                subscriptionDB.insert(state);
-
-            } catch (Exception ex) {
-                log.error("unable to insert subscription to database", ex);
-                throw new AxisFault("unable to insert subscription to database ", ex);
-            }
-        }
-
-        addToSubscriptionMap(key, state);
-        return key;
-    }
-
-    private void addToSubscriptionMap(String key, SubscriptionState state) {
-
-        writeLockUnlockSubscription(true);
-        try {
-            subscriptions.put(key, state);
-        } finally {
-            writeLockUnlockSubscription(false);
-        }
-
-    }
-
-    /**
-     * @param xpathString
-     * @return
-     */
-    private String generateSubscriptionId(boolean xPath) {
-        String key;
-        String subIdPrefix = null; // Used to indicate weather a subscription
-        // has an XPath subscription Or Topic
-        // only.
-        if (!xPath) {
-            subIdPrefix = "T";
-        } else {
-            subIdPrefix = "X";
-        }
-        key = subIdPrefix + "sub" + (counter++) + "@" + WsmgCommonConstants.PREFIX;
-        return key;
-    }
-
-    /**
-     * if find the subscription already exists, return the current subscriptionId else return null;
-     */
-
-    public String checkSubscriptionExist(SubscriptionState state) {
-
-        String key = null;
-
-        readLockUnlockSubscriptions(true);
-        try {
-
-            for (Iterator<String> keyIterator = subscriptions.keySet().iterator(); keyIterator.hasNext();) {
-
-                String currentKey = keyIterator.next();
-                SubscriptionState value = subscriptions.get(currentKey);
-
-                if (value.equals(state)) {
-                    value.setCreationTime(System.currentTimeMillis());
-                    log.info("Subscription Already exists." + " Using the current subscriptionId");
-                    key = currentKey;
-                    break;
-                }
-
-            }
-
-        } finally {
-            readLockUnlockSubscriptions(false);
-        }
-
-        return key;
-    }
-
-    public void checkSubscriptionDB(WsmgStorage storage) throws AxisFault {
-        OMElement subscribeXmlElement;
-        String subscriptionId;
-        // Read subscription Info from Subscription DB
-        List<SubscriptionEntry> subscriptionEntry = storage.getAllSubscription();
-        if (subscriptionEntry == null) {
-            return;
-        }
-
-        WSNotificationProcessingContextBuilder wsntBuilder = new WSNotificationProcessingContextBuilder();
-        WSEProcessingContextBuilder wseBuilder = new WSEProcessingContextBuilder();
-
-        // Create subscription for these entries from DB
-        for (int i = 0; i < subscriptionEntry.size(); i++) {
-
-            ProcessingContextBuilder processingCtxBuilder = null;
-
-            log.info("Subscription No. " + i + " is " + subscriptionEntry.get(i).getSubscriptionId());
-
-            StringReader sr = new StringReader(subscriptionEntry.get(i).getSubscribeXml());
-            XMLInputFactory inputFactory = XMLInputFactory.newInstance();
-            XMLStreamReader inflow;
-            try {
-                inflow = inputFactory.createXMLStreamReader(sr);
-
-                StAXOMBuilder builder = new StAXOMBuilder(inflow); // get the
-                // root
-                // element (in
-                // this case the
-                // envelope)
-                subscribeXmlElement = builder.getDocumentElement();
-
-                if (subscribeXmlElement.getNamespace().getNamespaceURI()
-                        .equals(NameSpaceConstants.WSNT_NS.getNamespaceURI())) {
-                    processingCtxBuilder = wsntBuilder;
-
-                } else {
-                    processingCtxBuilder = wseBuilder;
-                }
-
-                subscriptionId = subscriptionEntry.get(i).getSubscriptionId();
-
-                ProcessingContext context = processingCtxBuilder.build(subscribeXmlElement);
-                createSubscription(subscriptionId, context);
-
-            } catch (XMLStreamException e) {
-                log.error("error occured while checking subscription db", e);
-            }
-        }
-        RunTimeStatistics.totalSubscriptionsAtStartUp += subscriptionEntry.size();
-    }
-
-    // This is used for debug
-    public void showAllSubscription() {
-        String key = null;
-        SubscriptionState value = null;
-        Set<String> keySet = subscriptions.keySet();
-        log.info("List of all subscriptions:");
-        for (Iterator<String> iterator = keySet.iterator(); iterator.hasNext();) {
-            key = iterator.next();
-            value = subscriptions.get(key);
-            log.info("******" + key + "-->" + value.getConsumerIPAddressStr() + "##" + value.getLocalTopic());
-        }
-    }
-
-    public int unsubscribe(ProcessingContext ctx) throws AxisFault {
-
-        String subscriptionId = ctx.getContextParameter(ContextParameters.SUB_ID);
-        if (subscriptionId == null || subscriptionId.trim().length() == 0) {
-            throw new AxisFault("subscription identifier is not provided");
-        }
-
-        removeSubscription(subscriptionId);
-        RunTimeStatistics.totalUnSubscriptions++;
-        return 0;
-    }
-
-    int removeSubscription(String subId) throws AxisFault {
-
-        SubscriptionState subscription = null;
-
-        writeLockUnlockSubscription(true);
-        try {
-            subscription = subscriptions.remove(subId);
-        } finally {
-            writeLockUnlockSubscription(false);
-        }
-
-        if (subscription == null) {
-            throw AxisFault.makeFault(new RuntimeException("unknown subscription: " + subId));
-
-        }
-
-        subscriptionDB.delete(subId);
-
-        for (AbstractMessageMatcher mm : wsmgConfig.getMessageMatchers()) {
-            mm.handleUnsubscribe(subId);
-        }
-
-        return 0;
-    }
-
-    public void resumeSubscription(ProcessingContext ctx) throws AxisFault {
-
-        String subscriptionId = ctx.getContextParameter(ContextParameters.SUB_ID);
-
-        if (subscriptionId == null) {
-            throw AxisFault.makeFault(new RuntimeException("missing subscription id"));
-        }
-
-        writeLockUnlockSubscription(true);// lock
-        try {
-            SubscriptionState subscription = subscriptions.get(subscriptionId);
-
-            if (subscription == null) {
-
-                throw AxisFault.makeFault(new RuntimeException("no subscription found for id: " + subscriptionId));
-            }
-
-            subscription.resume();
-        } finally {
-            // this will execute even exception is thrown.
-            writeLockUnlockSubscription(false);
-        }
-    }
-
-    public void pauseSubscription(ProcessingContext ctx) throws AxisFault {
-
-        String subscriptionId = ctx.getContextParameter(ContextParameters.SUB_ID);
-
-        if (subscriptionId == null) {
-            throw AxisFault.makeFault(new RuntimeException("missing subscription id"));
-        }
-
-        writeLockUnlockSubscription(true);// read lock should be sufficient
-        // (since we are not modifying the
-        // map)
-        try {
-            SubscriptionState subscription = subscriptions.get(subscriptionId);
-
-            if (subscription == null) {
-
-                throw AxisFault.makeFault(new RuntimeException("no subscription found for id: " + subscriptionId));
-
-            }
-
-            subscription.pause();
-        } finally {
-            // this will execute even exception is thrown.
-            writeLockUnlockSubscription(false);
-        }
-    }
-
-    public void readLockUnlockSubscriptions(boolean lock) {
-        ReadLock readlock = subscriptionLock.readLock();
-        lockUnlock(readlock, lock);
-    }
-
-    public void writeLockUnlockSubscription(boolean lock) {
-        WriteLock writeLock = subscriptionLock.writeLock();
-        lockUnlock(writeLock, lock);
-    }
-
-    private void lockUnlock(Lock l, boolean lock) {
-
-        if (lock) {
-            l.lock();
-        } else {
-            l.unlock();
-        }
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionState.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionState.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionState.java
deleted file mode 100644
index 9ddadb3..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/subscription/SubscriptionState.java
+++ /dev/null
@@ -1,301 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.subscription;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import javax.xml.namespace.QName;
-
-import org.apache.airavata.wsmg.broker.ConsumerInfo;
-import org.apache.airavata.wsmg.commons.util.OMElementComparator;
-import org.apache.airavata.wsmg.messenger.OutGoingQueue;
-import org.apache.airavata.wsmg.util.BrokerUtil;
-import org.apache.axiom.om.OMElement;
-import org.apache.axis2.addressing.EndpointReference;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SubscriptionState {
-
-    private final Logger logger = LoggerFactory.getLogger(SubscriptionState.class);
-
-    private long creationTime = 0;
-    private long lastAvailableTime = 0;
-
-    private int unAvailableCounter = 0;
-    private boolean isNeverExpire = false;
-    private boolean isWsrmPolicy;
-
-    public String subId;
-    public String curNotif;
-    private String localTopicString;
-    private String xpathString;
-    private String subscribeXml;
-
-    ConsumerInfo consumerInfo = null;
-
-    EndpointReference consumerReference;
-
-    URI consumerURI = null;
-
-    private OutGoingQueue outGoingQueue;
-
-    /**
-     * @return Returns the creationTime.
-     */
-    public long getCreationTime() {
-        return creationTime;
-    }
-
-    /**
-     * @param creationTime
-     *            The creationTime to set.
-     */
-    public void setCreationTime(long creationTime) {
-        this.creationTime = creationTime;
-    }
-
-    public void setId(String id) {
-        subId = id;
-    }
-
-    public String getId() {
-        return subId;
-    }
-
-    public boolean isWsrmPolicy() {
-        return isWsrmPolicy;
-    }
-
-    public void setWsrmPolicy(boolean wsrmPolicy) {
-        this.isWsrmPolicy = wsrmPolicy;
-    }
-
-    // TODO: outGoingQueue is not belong to this class. Move it to elsewhere
-    // related to notification handler in wsntAdapter.
-    public SubscriptionState(EndpointReference consumerRef, boolean useNotify, boolean wsrmEnabled, String topic,
-            String xpath, String type, OutGoingQueue outGoingQueue) {
-        this.consumerReference = consumerRef;
-        try {
-            this.consumerURI = new URI(consumerRef.getAddress());
-        } catch (URISyntaxException e) {
-            // this should not happen
-            logger.error("invalid consumer URI returned by axis om", e);
-
-        }
-        this.outGoingQueue = outGoingQueue;
-        // if (topic == null) {
-        // throw new IllegalArgumentException();
-        // }
-        this.localTopicString = topic;
-        this.xpathString = xpath;
-        this.isWsrmPolicy = wsrmEnabled;
-        consumerInfo = new ConsumerInfo(consumerRef.getAddress(), type, useNotify, false);
-
-    }
-
-    public void resume() {
-        consumerInfo.setPaused(false);
-    }
-
-    public void pause() {
-        consumerInfo.setPaused(true);
-    }
-
-    public String getConsumerIPAddressStr() {
-        return consumerURI.toString();
-    }
-
-    public URI getConsumerAddressURI() {
-        return consumerURI;
-    }
-
-    public String getLocalTopic() {
-        // QName topicExpressionQName =
-        // xsul.util.XsulUtil.toQName(localTopicString, localTopicString
-        // .requiredTextContent());
-        // String topicLocalString = topicExpressionQName.getLocalPart();
-        return localTopicString;
-    }
-
-    /**
-     * @return Returns the consumeReference.
-     */
-    public EndpointReference getConsumerReference() {
-        return consumerReference;
-    }
-
-    /**
-     * @return Returns the curNotif.
-     */
-    public String getCurNotif() {
-        return curNotif;
-    }
-
-    /**
-     * @param curNotif
-     *            The curNotif to set.
-     */
-    public void setCurNotif(String curNotif) {
-        this.curNotif = curNotif;
-    }
-
-    /**
-     * @return Returns the outGoingQueue.
-     */
-    public OutGoingQueue getOutGoingQueue() {
-        return outGoingQueue;
-    }
-
-    /**
-     * @param outGoingQueue
-     *            The outGoingQueue to set.
-     */
-    public void setOutGoingQueue(OutGoingQueue outGoingQueue) {
-        this.outGoingQueue = outGoingQueue;
-    }
-
-    /**
-     * @return Returns the consumerInfo.
-     */
-    public ConsumerInfo getConsumerInfo() {
-        return consumerInfo;
-    }
-
-    public void resetUnAvailableCounter() {
-        unAvailableCounter = 0;
-    }
-
-    public int addUnAvailableCounter() {
-        unAvailableCounter++;
-        return unAvailableCounter;
-    }
-
-    /**
-     * @return Returns the unAvailableCounter.
-     */
-    public int getUnAvailableCounter() {
-        return unAvailableCounter;
-    }
-
-    public String getXpathString() {
-        return xpathString;
-    }
-
-    public void setXpathString(String xpathString) {
-        this.xpathString = xpathString;
-    }
-
-    /**
-     * @return Returns the isNeverExpire.
-     */
-    public boolean isNeverExpire() {
-        return isNeverExpire;
-    }
-
-    /**
-     * @param isNeverExpire
-     *            The isNeverExpire to set.
-     */
-    public void setNeverExpire(boolean neverExpire) {
-        this.isNeverExpire = neverExpire;
-    }
-
-    /**
-     * @return Returns the lastAvailableTime.
-     */
-    public long getLastAvailableTime() {
-        return lastAvailableTime;
-    }
-
-    /**
-     * @param lastAvailableTime
-     *            The lastAvailableTime to set.
-     */
-    public void setLastAvailableTime(long lastAvailableTime) {
-        this.lastAvailableTime = lastAvailableTime;
-
-    }
-
-    public void setSubscribeXml(String xml) {
-        subscribeXml = xml;
-    }
-
-    public String getSubscribeXml() {
-        return subscribeXml;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (o instanceof SubscriptionState) {
-            SubscriptionState subscription = (SubscriptionState) o;
-            return BrokerUtil.sameStringValue(subscription.getLocalTopic(), this.getLocalTopic())
-                    && BrokerUtil.sameStringValue(subscription.getXpathString(), this.getXpathString())
-                    && BrokerUtil.sameStringValue(subscription.getConsumerIPAddressStr(),
-                            this.getConsumerIPAddressStr()) && equalReferenceParameters(subscription);
-        }
-        return false;
-    }
-
-    private boolean equalReferenceParameters(SubscriptionState anotherSubscription) {
-
-        Map<QName, OMElement> otherRefProperties = anotherSubscription.getConsumerReference()
-                .getAllReferenceParameters();
-        Map<QName, OMElement> myRefProperties = getConsumerReference().getAllReferenceParameters();
-
-        /*
-         * Basic comparison
-         */
-        if (otherRefProperties == null && myRefProperties == null) {
-            return true;
-        } else if (otherRefProperties == null || myRefProperties == null) {
-            return false;
-        } else if (otherRefProperties.size() != myRefProperties.size()) {
-            return false;
-        }
-
-        /*
-         * This OMElementComparator supports ignore list, but we don't use it here.
-         */
-        Iterator<Entry<QName, OMElement>> iterator = otherRefProperties.entrySet().iterator();
-        while (iterator.hasNext()) {
-
-            Entry<QName, OMElement> entry = iterator.next();
-            if (!myRefProperties.containsKey(entry.getKey())) {
-                return false;
-            }
-
-            OMElement myElement = myRefProperties.get(entry.getKey());
-            OMElement otherElement = entry.getValue();
-
-            if (!OMElementComparator.compare(myElement, otherElement)) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProcessingContextBuilder.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProcessingContextBuilder.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProcessingContextBuilder.java
deleted file mode 100644
index fe276bf..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProcessingContextBuilder.java
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wseventing;
-
-import java.util.Iterator;
-
-import javax.xml.namespace.QName;
-
-import org.apache.airavata.wsmg.broker.context.ContextParameters;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.broker.context.ProcessingContextBuilder;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.util.BrokerUtil;
-import org.apache.airavata.wsmg.util.WsEventingOperations;
-import org.apache.axiom.om.OMElement;
-import org.apache.axiom.soap.SOAPBody;
-import org.apache.axiom.soap.SOAPEnvelope;
-import org.apache.axiom.soap.SOAPHeader;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.addressing.EndpointReferenceHelper;
-import org.apache.axis2.context.MessageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class WSEProcessingContextBuilder extends ProcessingContextBuilder {
-
-    private static final Logger logger = LoggerFactory.getLogger(WSEProcessingContextBuilder.class);
-
-    public ProcessingContext build(OMElement elem) {
-
-        ProcessingContext processingContext = new ProcessingContext();
-
-        if (elem != null && elem.getLocalName().equals("Subscribe")) {
-            logger.debug("found subscribe element");
-            onSubscription(processingContext, elem);
-
-        }
-
-        return processingContext;
-    }
-
-    public ProcessingContext build(SOAPEnvelope elem) {
-
-        ProcessingContext context = null;
-
-        SOAPBody soapBody = elem.getBody();
-        if (soapBody != null) {
-
-            context = build(soapBody.getFirstElement());
-
-        } else {
-            context = build((OMElement) null);
-        }
-
-        context.setEnvelope(elem);
-        extractInfoFromHeader(context, elem.getHeader());
-
-        return context;
-    }
-
-    public ProcessingContext build(MessageContext msgContext, WsEventingOperations operation) {
-
-        ProcessingContext processingContext = new ProcessingContext();
-
-        switch (operation) {
-        case SUBSCRIBE: {
-
-            Iterator<OMElement> iterator = msgContext.getEnvelope().getBody()
-                    .getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(), "Subscribe"));
-
-            if (!iterator.hasNext()) {
-                throw new RuntimeException("invalid subscription message - no subscribe element");
-            }
-
-            onSubscription(processingContext, iterator.next());
-        }
-            break;
-
-        }
-
-        processingContext.setMessageConext(msgContext);
-        processingContext.setEnvelope(msgContext.getEnvelope());
-        extractInfoFromHeader(processingContext, msgContext.getEnvelope().getHeader());
-        String topicFromUrl = BrokerUtil.getTopicFromRequestPath(msgContext.getTo().getAddress());
-
-        processingContext.setContextParameter(ContextParameters.TOPIC_FROM_URL, topicFromUrl);
-
-        return processingContext;
-    }
-
-    /**
-     * @param processingContext
-     * @param subscribeElement
-     */
-    private void onSubscription(ProcessingContext processingContext, OMElement subscribeElement) {
-
-        processingContext.setContextParameter(ContextParameters.SUBSCRIBE_ELEMENT, subscribeElement);
-
-        // -- check optional element - expires
-        Iterator iterator = subscribeElement.getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(),
-                "Expires"));
-
-        if (iterator.hasNext()) {
-
-            processingContext.setContextParameter(ContextParameters.SUBSCRIBER_EXPIRES,
-                    ((OMElement) iterator.next()).getText());
-
-        }
-
-        iterator = subscribeElement
-                .getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(), "Filter"));
-
-        if (!iterator.hasNext()) {
-
-            throw new RuntimeException("invalid subscription - unable to find filter dialet");
-
-        }
-
-        processingContext.setContextParameter(ContextParameters.FILTER_ELEMENT, iterator.next());
-
-        iterator = subscribeElement.getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(),
-                "Delivery"));
-
-        if (!iterator.hasNext()) {
-            throw new RuntimeException("invalid subscription - unable to find delivery tag");
-        }
-
-        OMElement delivery = (OMElement) iterator.next();
-
-        iterator = delivery.getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(), "NotifyTo"));
-
-        if (!iterator.hasNext()) {
-            throw new RuntimeException("invalid subscription - unable to find NotifyTo tag");
-        }
-
-        OMElement notifyToElement = (OMElement) iterator.next();
-
-        processingContext.setContextParameter(ContextParameters.NOTIFY_TO_ELEMENT, notifyToElement);
-
-        try {
-
-            processingContext.setContextParameter(ContextParameters.NOTIFY_TO_EPR,
-                    EndpointReferenceHelper.fromOM(notifyToElement));
-
-        } catch (AxisFault e) {
-            throw new RuntimeException("invalid subscription - unable to parse notify to end point reference", e);
-        }
-
-    }
-
-    private void extractInfoFromHeader(ProcessingContext context, SOAPHeader header) {
-
-        Iterator ite = header.getChildrenWithName(new QName(NameSpaceConstants.WSE_NS.getNamespaceURI(),
-                WsmgCommonConstants.SUBSCRIPTION_ID));
-
-        if (ite.hasNext()) {
-            OMElement identifier = (OMElement) ite.next();
-            logger.debug("extracted identifier " + identifier.getText());
-
-            context.setContextParameter(ContextParameters.SUB_ID, identifier.getText());
-
-        }
-
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProtocolSupport.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProtocolSupport.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProtocolSupport.java
deleted file mode 100644
index f17a980..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEProtocolSupport.java
+++ /dev/null
@@ -1,204 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wseventing;
-
-import java.util.Calendar;
-import java.util.Date;
-
-import javax.xml.namespace.QName;
-
-import org.apache.airavata.wsmg.broker.context.ContextParameters;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.broker.subscription.SubscriptionState;
-import org.apache.airavata.wsmg.commons.CommonRoutines;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.messenger.OutGoingQueue;
-import org.apache.airavata.wsmg.util.BrokerUtil;
-import org.apache.axiom.om.OMAbstractFactory;
-import org.apache.axiom.om.OMElement;
-import org.apache.axiom.om.OMFactory;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.addressing.EndpointReference;
-import org.apache.axis2.addressing.EndpointReferenceHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class WSEProtocolSupport {
-
-    private static final Logger log = LoggerFactory.getLogger(WSEProtocolSupport.class);
-
-    public SubscriptionState createSubscriptionState(ProcessingContext ctx, OutGoingQueue outgoingQueue)
-            throws AxisFault {
-
-        boolean neverExpire = false; // is true if expiration time is less than
-        String topicLocalString = "";
-        String xpathString = "";
-        EndpointReference consumerReference = ctx.getContextParameter(ContextParameters.NOTIFY_TO_EPR);
-
-        if (consumerReference == null) {
-            throw new AxisFault("Only Push delivery Mode (NotifyTo) is supported in WSE");
-        }
-
-        String expireTimeString = ctx.getContextParameter(ContextParameters.SUBSCRIBER_EXPIRES);
-
-        if (expireTimeString == null) {
-            neverExpire = true;
-
-        } else {
-
-            long expireTime = Long.valueOf(expireTimeString);
-            if (expireTime < 0) {
-                neverExpire = true;
-            }
-        }
-
-        OMElement filterEl = ctx.getContextParameter(ContextParameters.FILTER_ELEMENT);
-
-        if (filterEl == null) {
-
-            topicLocalString = ctx.getContextParameter(ContextParameters.TOPIC_FROM_URL);
-
-            if (topicLocalString == null) {
-                topicLocalString = WsmgCommonConstants.WILDCARD_TOPIC;
-            }
-
-            log.debug("got topicLocalString=" + topicLocalString);
-            // topicLocalString = "wseTopic";
-            // // a special topic, used in WSNT and JMS. Do not use a
-            // wildcard topic here since wildcard string varies by system
-        } else {
-
-            String filterDialectAttrib = filterEl.getAttributeValue(new QName(null, "Dialect"));
-
-            if (filterDialectAttrib.compareTo(WsmgCommonConstants.TOPIC_EXPRESSION_SIMPLE_DIALECT) == 0) {
-                topicLocalString = BrokerUtil.getTopicLocalString(filterEl.getText()); // get what ever inside this
-                                                                                       // element
-
-                if (topicLocalString == null) {
-                    throw new AxisFault("topic is not given in the subscription");
-                }
-
-            } else if (filterDialectAttrib.compareTo(WsmgCommonConstants.XPATH_DIALECT) == 0) {
-
-                // use topicFromUrl if
-                // was provided
-                topicLocalString = ctx.getContextParameter(ContextParameters.TOPIC_FROM_URL);
-
-                xpathString = filterEl.getText();
-
-                log.debug("got topicLocalString=" + topicLocalString + " xpathString=" + xpathString);
-
-                // TODO: Add XPath canonicalization here in the parsing. To
-                // generate a
-                // canonicalized XPath string
-                // Possibly use Query query =
-                // XPQuery.parseQuery(xpathExpression, index);
-                if (xpathString == null) {
-                    throw new AxisFault("xpath expression is not given");
-                }
-            } else if (filterDialectAttrib.compareTo(WsmgCommonConstants.TOPIC_AND_XPATH_DIALECT) == 0) {
-                OMElement topicEl = filterEl.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                        .getNamespaceURI(), "TopicExpression"));
-                if (topicEl != null) {
-                    topicLocalString = BrokerUtil.getTopicLocalString(topicEl.getText());
-                }
-                OMElement xpathEl = filterEl.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                        .getNamespaceURI(), "MessageContent"));
-                if (xpathEl != null) {
-                    xpathString = xpathEl.getText();
-                    if (xpathString == null && topicLocalString == null) {
-                        throw new AxisFault("Both topic string and " + "XPath String are null!");
-                    }
-                }
-            } else {
-                throw new AxisFault("Unkown dialect: ");
-                // topicLocalString = "wseTopic"; //a special topic, used in
-                // WSNT and JMS
-            }
-
-        }
-
-        if (topicLocalString == null || topicLocalString.length() == 0) {
-            topicLocalString = WsmgCommonConstants.WILDCARD_TOPIC;
-        }
-
-        // Create SubscriptionState Object
-        SubscriptionState state = new SubscriptionState(consumerReference, true, false, topicLocalString, xpathString,
-                "wse", outgoingQueue);
-
-        state.setNeverExpire(neverExpire); // default false
-
-        return state;
-    }
-
-    public void createSubscribeResponse(ProcessingContext ctx, String subId) throws AxisFault {
-
-        OMFactory factory = OMAbstractFactory.getOMFactory();
-
-        ctx.addResponseMsgNameSpaces(NameSpaceConstants.WSE_NS);
-
-        OMElement responseMessage = factory.createOMElement("SubscribeResponse", NameSpaceConstants.WSE_NS);
-
-        OMElement identifier = factory.createOMElement(WsmgCommonConstants.SUBSCRIPTION_ID,
-                responseMessage.getNamespace());
-        identifier.setText(subId);
-        EndpointReference serviceLocationEndpointReference = new EndpointReference(ctx.getMessageContext()
-                .getAxisService().getEndpointURL());
-
-        serviceLocationEndpointReference.addReferenceParameter(identifier);
-
-        OMElement expiresEl = factory.createOMElement("Expires", responseMessage.getNamespace(), responseMessage);
-
-        Date expiration = getFutureExpirationDate();
-        String dateString = CommonRoutines.getXsdDateTime(expiration);
-        expiresEl.setText(dateString);
-
-        OMElement subscriptionManagerEpr = null;
-        try {
-
-            subscriptionManagerEpr = EndpointReferenceHelper.toOM(factory, serviceLocationEndpointReference, new QName(
-                    NameSpaceConstants.WSE_NS.getNamespaceURI(), "SubscriptionManager"), NameSpaceConstants.WSA_NS
-                    .getNamespaceURI());
-
-            responseMessage.addChild(subscriptionManagerEpr);
-            subscriptionManagerEpr.setNamespace(responseMessage.getNamespace());
-        } catch (AxisFault e) {
-            log.error("unable to resolve EPR from OM", e);
-            throw e;
-        }
-
-        ctx.setRespMessage(responseMessage);
-    }
-
-    private Date getFutureExpirationDate() {
-        // Get a Calendar for current locale and time zone
-        Calendar cal = Calendar.getInstance();
-        // currentDate.setDate(currentDate.getDate()+1);
-        // Get a Date object that represents 30 days from now
-        Date currentDate = new Date(); // Current date
-        cal.setTime(currentDate); // Set it in the Calendar object
-        cal.add(Calendar.DATE, 30); // Add 30 days
-        Date expiration = cal.getTime(); // Retrieve the resulting date
-        return expiration;
-    }
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingMsgReceiver.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingMsgReceiver.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingMsgReceiver.java
deleted file mode 100644
index 6af3bc5..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingMsgReceiver.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wseventing;
-
-import org.apache.airavata.wsmg.broker.AbstractBrokerMsgReceiver;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.config.WsmgConfigurationContext;
-import org.apache.airavata.wsmg.util.WsEventingOperations;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.context.MessageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * BrokerServiceMessageReceiverInOut message receiver
- */
-
-public class WSEventingMsgReceiver extends AbstractBrokerMsgReceiver {
-
-    private static final Logger log = LoggerFactory.getLogger(WSEventingMsgReceiver.class);
-    WSEProcessingContextBuilder builder = new WSEProcessingContextBuilder();
-
-    public MessageContext process(MessageContext inMsg, String operationName) throws AxisFault {
-
-        WsEventingOperations msgType = WsEventingOperations.valueFrom(operationName);
-        ProcessingContext processingContext = builder.build(inMsg, msgType);
-        MessageContext outputMsg = null;
-
-        log.debug("WS-Eventing: " + msgType);
-
-        switch (msgType) {
-        case SUBSCRIBE: {
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getSubscriptionManager().subscribe(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case UNSUBSCRIBE: {
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getSubscriptionManager().unsubscribe(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case RENEW:
-        case GET_STATUS:
-        case SUBSCRIPTION_END:
-        default:
-            throw new AxisFault("unsupported operation" + msgType.toString());
-
-        }
-        return outputMsg;
-    }
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingPublishMsgReceiver.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingPublishMsgReceiver.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingPublishMsgReceiver.java
deleted file mode 100644
index 63c6bbc..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wseventing/WSEventingPublishMsgReceiver.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wseventing;
-
-import org.apache.airavata.wsmg.broker.AbstractBrokerMsgReceiver;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.config.WsmgConfigurationContext;
-import org.apache.airavata.wsmg.util.WsEventingOperations;
-import org.apache.axiom.om.OMElement;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.context.MessageContext;
-
-public class WSEventingPublishMsgReceiver extends AbstractBrokerMsgReceiver {
-
-    WSEProcessingContextBuilder builder = new WSEProcessingContextBuilder();
-
-    @Override
-    protected MessageContext process(MessageContext inMsgContext, String operationName) throws AxisFault {
-
-        ProcessingContext processingContext = builder.build(inMsgContext, WsEventingOperations.PUBLISH);
-
-        try {
-
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsgContext
-                    .getConfigurationContext().getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getNotificationProcessor().processMsg(processingContext, NameSpaceConstants.WSE_NS);
-        } catch (Exception e) {
-            throw new AxisFault("unable to process message", e);
-        }
-        return createOutputMessageContext(inMsgContext, processingContext);
-    }
-
-    @Override
-    protected MessageContext createOutputMessageContext(MessageContext inMsg, ProcessingContext processingContext)
-            throws AxisFault {
-
-        MessageContext outputContext = null;
-
-        OMElement responseMessage = processingContext.getRespMessage();
-        if (responseMessage != null) {
-
-            outputContext = super.createOutputMessageContext(inMsg, processingContext);
-
-            String responseAction = String.format("%s/%s", NameSpaceConstants.WSE_NS.getNamespaceURI(),
-                    responseMessage.getLocalName());
-
-            outputContext.setSoapAction(responseAction);
-        }
-
-        return outputContext;
-    }
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNTProtocolSupport.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNTProtocolSupport.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNTProtocolSupport.java
deleted file mode 100644
index c8d3e37..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNTProtocolSupport.java
+++ /dev/null
@@ -1,237 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wsnotification;
-
-import java.util.Map;
-
-import javax.xml.namespace.QName;
-
-import org.apache.airavata.wsmg.broker.context.ContextParameters;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.broker.subscription.SubscriptionState;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.messenger.OutGoingQueue;
-import org.apache.airavata.wsmg.util.BrokerUtil;
-import org.apache.axiom.om.OMAbstractFactory;
-import org.apache.axiom.om.OMElement;
-import org.apache.axiom.om.OMFactory;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.addressing.EndpointReference;
-import org.apache.axis2.addressing.EndpointReferenceHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class WSNTProtocolSupport {
-
-    private static final Logger log = LoggerFactory.getLogger(WSNTProtocolSupport.class);
-
-    public SubscriptionState createSubscriptionState(ProcessingContext ctx, OutGoingQueue outgoingQueue)
-            throws AxisFault {
-
-        EndpointReference consumerReference = ctx.getContextParameter(ContextParameters.NOTIFY_TO_EPR);
-
-        if (consumerReference == null) {
-            throw new AxisFault("Only Push delivery Mode (NotifyTo) is supported");
-        }
-
-        boolean neverExpire = false; // is true if expiration time is less than
-        boolean useNotify = true; // notify by event notifications
-        boolean wsrmEnabled = false;
-        String topicLocalString = "";
-        String xpathString = "";
-
-        String expireTimeString = ctx.getContextParameter(ContextParameters.SUBSCRIBER_EXPIRES);
-
-        if (expireTimeString == null) {
-            neverExpire = true;
-        } else {
-            long expireTime = Long.valueOf(expireTimeString);
-            if (expireTime < 0) {
-                neverExpire = true;
-            }
-        }
-
-        OMElement useNotifyEl = ctx.getContextParameter(ContextParameters.USE_NOTIFY_ELEMENT);
-        if (useNotifyEl != null) {
-            String s = useNotifyEl.getText();
-            useNotify = Boolean.valueOf(s);
-        }
-
-        // get policy if exist
-        OMElement element = ctx.getContextParameter(ContextParameters.SUB_POLICY);
-        if (element != null) {
-            wsrmEnabled = true;
-        }
-
-        OMElement topicExpressionEl = ctx.getContextParameter(ContextParameters.TOPIC_EXPRESSION_ELEMENT);
-
-        if (topicExpressionEl != null) {
-            topicLocalString = BrokerUtil.getTopicLocalString(topicExpressionEl.getText());
-        }
-
-        OMElement xpathEl = ctx.getContextParameter(ContextParameters.XPATH_ELEMENT);
-
-        if (xpathEl != null) {
-            xpathString = BrokerUtil.getXPathString(xpathEl);
-        }
-        if (xpathString == null && topicLocalString == null) {
-            throw new AxisFault("Both topic string and XPath String are null!");
-
-        }
-
-        if (topicLocalString == null || topicLocalString.length() == 0) {
-            topicLocalString = WsmgCommonConstants.WILDCARD_TOPIC;
-        }
-
-        // Create SubscriptionState Object
-        SubscriptionState state = new SubscriptionState(consumerReference, useNotify, wsrmEnabled, topicLocalString,
-                xpathString, "wsnt", outgoingQueue);
-
-        state.setNeverExpire(neverExpire); // default false
-
-        return state;
-    }
-
-    public void createSubscribeResponse(ProcessingContext ctx, String subId) throws AxisFault {
-
-        OMFactory factory = OMAbstractFactory.getOMFactory();
-
-        ctx.addResponseMsgNameSpaces(NameSpaceConstants.WSNT_NS);
-        OMElement responseMessage = factory.createOMElement("SubscribeResponse", NameSpaceConstants.WSNT_NS);
-
-        OMElement identifier = factory.createOMElement(WsmgCommonConstants.SUBSCRIPTION_ID,
-                responseMessage.getNamespace());
-        identifier.setText(subId);
-        EndpointReference serviceLocationEndpointReference = new EndpointReference(ctx.getMessageContext()
-                .getAxisService().getEndpointURL());
-        serviceLocationEndpointReference.addReferenceParameter(identifier);
-
-        OMElement subscriptionReference = null;
-        try {
-            subscriptionReference = EndpointReferenceHelper.toOM(factory, serviceLocationEndpointReference, new QName(
-                    "SubscriptionReference"), NameSpaceConstants.WSA_NS.getNamespaceURI());
-
-            responseMessage.addChild(subscriptionReference);
-            subscriptionReference.setNamespace(responseMessage.getNamespace());
-
-        } catch (AxisFault e) {
-            log.error("unable to resolve EPR from OM", e);
-            throw e;
-        }
-
-        ctx.setRespMessage(responseMessage);
-    }
-
-    public static class Client {
-
-        public static OMElement createSubscriptionMsg(EndpointReference eventSinkLocation, String topicExpression,
-                String xpathExpression) throws AxisFault {
-            OMFactory factory = OMAbstractFactory.getOMFactory();
-
-            OMElement message = factory.createOMElement("SubscribeRequest", NameSpaceConstants.WSNT_NS);
-
-            if (topicExpression != null) {
-                OMElement topicExpEl = factory.createOMElement("TopicExpression", NameSpaceConstants.WSNT_NS, message);
-
-                topicExpEl.addAttribute("Dialect", WsmgCommonConstants.TOPIC_EXPRESSION_SIMPLE_DIALECT,
-                        NameSpaceConstants.WSNT_NS);
-                topicExpEl.declareNamespace(NameSpaceConstants.WIDGET_NS);
-                topicExpEl.setText(NameSpaceConstants.WIDGET_NS.getPrefix() + ":" + topicExpression);
-            }
-
-            if (xpathExpression != null) {
-                OMElement xpathExpEl = factory.createOMElement("Selector", NameSpaceConstants.WSNT_NS, message);
-                xpathExpEl.addAttribute("Dialect", WsmgCommonConstants.XPATH_DIALECT, null);
-                xpathExpEl.setText(xpathExpression);
-            }
-
-            OMElement useNotifyEl = factory.createOMElement("UseNotify", message.getNamespace(), message);
-            useNotifyEl.setText("true");// check wether we still need this
-
-            OMElement eprCrEl = EndpointReferenceHelper.toOM(factory, eventSinkLocation,
-                    new QName("ConsumerReference"), NameSpaceConstants.WSA_NS.getNamespaceURI());
-
-            message.addChild(eprCrEl);
-            eprCrEl.setNamespace(message.getNamespace());
-
-            return message;
-        }
-
-        public static String decodeSubscriptionResponse(OMElement subscriptionReference) throws AxisFault {
-
-            String subscriptionId = null;
-
-            EndpointReference subscriptionReferenceEPR = EndpointReferenceHelper.fromOM(subscriptionReference);
-
-            Map<QName, OMElement> referenceParams = subscriptionReferenceEPR.getAllReferenceParameters();
-
-            if (referenceParams != null) {
-                QName identifierQName = new QName(NameSpaceConstants.WSNT_NS.getNamespaceURI(),
-                        WsmgCommonConstants.SUBSCRIPTION_ID);
-
-                OMElement identifierEl = referenceParams.get(identifierQName);
-                subscriptionId = (identifierEl != null) ? identifierEl.getText() : null;
-
-            }
-
-            return subscriptionId;
-        }
-
-        public static OMElement createUnsubscribeMsg() {
-            OMFactory factory = OMAbstractFactory.getOMFactory();
-            OMElement message = factory.createOMElement("UnsubsribeRequest", NameSpaceConstants.WSNT_NS);
-
-            return message;
-        }
-
-        public static OMElement encodeNotification(String topic, OMElement message, EndpointReference producerReference)
-                throws AxisFault {
-            OMFactory factory = OMAbstractFactory.getOMFactory();
-
-            OMElement topicExpEl = factory.createOMElement("Topic", NameSpaceConstants.WSNT_NS);
-            topicExpEl.addAttribute("Dialect", WsmgCommonConstants.TOPIC_EXPRESSION_SIMPLE_DIALECT, null);
-            topicExpEl.declareNamespace(NameSpaceConstants.WIDGET_NS);
-            topicExpEl.setText(NameSpaceConstants.WIDGET_NS.getPrefix() + ":" + topic);
-
-            OMElement messageToNotify = factory.createOMElement("Notify", NameSpaceConstants.WSNT_NS);
-            messageToNotify.declareNamespace(NameSpaceConstants.WSNT_NS);
-            messageToNotify.declareNamespace(NameSpaceConstants.WSA_NS);
-            OMElement notificationMesssageEl = factory.createOMElement("NotificationMessage",
-                    messageToNotify.getNamespace(), messageToNotify);
-
-            notificationMesssageEl.addChild(topicExpEl);
-
-            notificationMesssageEl.addChild(EndpointReferenceHelper.toOM(factory, producerReference, new QName(
-                    notificationMesssageEl.getNamespace().getNamespaceURI(), "ProducerReference",
-                    notificationMesssageEl.getNamespace().getPrefix()), NameSpaceConstants.WSA_NS.getNamespaceURI()));
-
-            OMElement messageEl = factory.createOMElement("Message", notificationMesssageEl.getNamespace(),
-                    notificationMesssageEl);
-
-            messageEl.addChild(message);
-            return messageToNotify;
-        }
-
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationMsgReceiver.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationMsgReceiver.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationMsgReceiver.java
deleted file mode 100644
index 9d32ee9..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationMsgReceiver.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wsnotification;
-
-import org.apache.airavata.wsmg.broker.AbstractBrokerMsgReceiver;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.config.WsmgConfigurationContext;
-import org.apache.airavata.wsmg.util.WsNotificationOperations;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.context.MessageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * BrokerServiceMessageReceiverInOut message receiver
- */
-
-public class WSNotificationMsgReceiver extends AbstractBrokerMsgReceiver {
-
-    private static final Logger log = LoggerFactory.getLogger(WSNotificationMsgReceiver.class);
-
-    WSNotificationProcessingContextBuilder builder = new WSNotificationProcessingContextBuilder();
-
-    public MessageContext process(MessageContext inMsg, String operationName) throws AxisFault {
-
-        WsNotificationOperations msgType = WsNotificationOperations.valueFrom(operationName);
-
-        ProcessingContext processingContext = builder.build(inMsg, msgType);
-
-        MessageContext outputMsg = null;
-
-        switch (msgType) {
-        case NOTIFY: {
-            try {
-
-                WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg
-                        .getConfigurationContext().getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-                brokerConfigContext.getNotificationProcessor()
-                        .processMsg(processingContext, NameSpaceConstants.WSNT_NS);
-            } catch (Exception e) {
-                throw new AxisFault("unable to process message", e);
-            }
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case SUBSCRIBE: {
-
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-            brokerConfigContext.getSubscriptionManager().subscribe(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case UNSUBSCRIBE: {
-
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getSubscriptionManager().unsubscribe(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case RESUME_SUBSCRIPTION: {
-
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getSubscriptionManager().resumeSubscription(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case PAUSE_SUBSCRIPTION: {
-            WsmgConfigurationContext brokerConfigContext = (WsmgConfigurationContext) inMsg.getConfigurationContext()
-                    .getProperty(WsmgCommonConstants.BROKER_WSMGCONFIG);
-
-            brokerConfigContext.getSubscriptionManager().pauseSubscription(processingContext);
-            outputMsg = createOutputMessageContext(inMsg, processingContext);
-            break;
-        }
-        case GET_CURRENT_MSG:
-        default:
-            throw new AxisFault("not implemented yet");
-        }
-
-        return outputMsg;
-    }
-
-}// end of class

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationProcessingContextBuilder.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationProcessingContextBuilder.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationProcessingContextBuilder.java
deleted file mode 100644
index b517234..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/broker/wsnotification/WSNotificationProcessingContextBuilder.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.broker.wsnotification;
-
-import java.util.Iterator;
-
-import javax.xml.namespace.QName;
-
-import org.apache.airavata.wsmg.broker.context.ContextParameters;
-import org.apache.airavata.wsmg.broker.context.ProcessingContext;
-import org.apache.airavata.wsmg.broker.context.ProcessingContextBuilder;
-import org.apache.airavata.wsmg.commons.WsmgCommonConstants;
-import org.apache.airavata.wsmg.commons.NameSpaceConstants;
-import org.apache.airavata.wsmg.util.BrokerUtil;
-import org.apache.airavata.wsmg.util.WsNotificationOperations;
-import org.apache.axiom.om.OMElement;
-import org.apache.axiom.soap.SOAPBody;
-import org.apache.axiom.soap.SOAPEnvelope;
-import org.apache.axiom.soap.SOAPHeader;
-import org.apache.axis2.AxisFault;
-import org.apache.axis2.addressing.EndpointReference;
-import org.apache.axis2.addressing.EndpointReferenceHelper;
-import org.apache.axis2.context.MessageContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class WSNotificationProcessingContextBuilder extends ProcessingContextBuilder {
-
-    private static final Logger logger = LoggerFactory.getLogger(WSNotificationProcessingContextBuilder.class);
-
-    public ProcessingContext build(OMElement elem) {
-
-        ProcessingContext processingContext = new ProcessingContext();
-
-        if (elem == null
-                || (!elem.getNamespace().getNamespaceURI().equals(NameSpaceConstants.WSNT_NS.getNamespaceURI()))) {
-
-            logger.warn("invalid message payload recieved: " + elem);
-
-            return processingContext;
-        }
-
-        String localName = elem.getLocalName();
-
-        if (localName.equals("SubscribeRequest")) {
-            onSubscription(processingContext, elem);
-        }
-        return processingContext;
-    }
-
-    public void onSubscription(ProcessingContext context, OMElement subscribeElement) {
-        context.setContextParameter(ContextParameters.SUBSCRIBE_ELEMENT, subscribeElement);
-
-        OMElement consumerReference = subscribeElement.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                .getNamespaceURI(), "ConsumerReference"));
-        if (consumerReference == null) {
-            logger.warn("unable to find consumer reference" + " in subscribe message: " + subscribeElement);
-            return;
-        }
-        context.setContextParameter(ContextParameters.NOTIFY_TO_ELEMENT, consumerReference);
-
-        try {
-            EndpointReference consumerEpr = EndpointReferenceHelper.fromOM(consumerReference);
-            context.setContextParameter(ContextParameters.NOTIFY_TO_EPR, consumerEpr);
-
-        } catch (AxisFault e) {
-            logger.warn("invalid epr", e);
-            return;
-        }
-
-        OMElement topicExpression = subscribeElement.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                .getNamespaceURI(), "TopicExpression"));
-
-        if (topicExpression != null) { // topic can be null
-
-            context.setContextParameter(ContextParameters.TOPIC_EXPRESSION_ELEMENT, topicExpression);
-
-        }
-
-        OMElement useNotify = subscribeElement.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                .getNamespaceURI(), "UseNotify"));
-
-        if (useNotify != null) {
-            context.setContextParameter(ContextParameters.USE_NOTIFY_ELEMENT, useNotify);
-        }
-
-        OMElement selector = subscribeElement.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                .getNamespaceURI(), "Selector"));
-
-        if (selector != null) {
-            context.setContextParameter(ContextParameters.XPATH_ELEMENT, selector);
-        }
-
-        OMElement subscriptionPolicy = subscribeElement.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS
-                .getNamespaceURI(), WsmgCommonConstants.SUBSCRIPTION_POLICY));
-
-        if (subscriptionPolicy != null) {
-            context.setContextParameter(ContextParameters.SUB_POLICY, subscriptionPolicy);
-        }
-
-    }
-
-    public ProcessingContext build(MessageContext msgContext, WsNotificationOperations operation) {
-        ProcessingContext context = new ProcessingContext();
-
-        SOAPEnvelope soapEnvelope = msgContext.getEnvelope();
-
-        if (soapEnvelope == null) {
-            throw new RuntimeException("invalid message context - envelope is not found");
-        }
-
-        SOAPBody soapBody = soapEnvelope.getBody();
-
-        if (soapBody == null) {
-            throw new RuntimeException("invalid message context - soap envelope is not found");
-        }
-
-        SOAPHeader soapHeader = soapEnvelope.getHeader();
-
-        if (soapHeader == null) {
-            throw new RuntimeException("invalid message context - soap header is not found");
-        }
-
-        switch (operation) {
-        case SUBSCRIBE: {
-
-            Iterator<OMElement> iterator = soapBody.getChildrenWithName(new QName(NameSpaceConstants.WSNT_NS
-                    .getNamespaceURI(), "SubscribeRequest"));
-            if (!iterator.hasNext()) {
-                throw new RuntimeException("invalid message context - unable to find Subscribe information");
-            }
-
-            onSubscription(context, iterator.next());
-        }
-            break;
-        }
-
-        context.setEnvelope(soapEnvelope);
-        extractInfoFromHeader(context, soapHeader);
-        context.setMessageConext(msgContext);
-        String topicFromUrl = BrokerUtil.getTopicFromRequestPath(msgContext.getTo().getAddress());
-        context.setContextParameter(ContextParameters.TOPIC_FROM_URL, topicFromUrl);
-
-        return context;
-    }
-
-    private void extractInfoFromHeader(ProcessingContext context, SOAPHeader header) {
-
-        Iterator ite = header.getChildrenWithName(new QName(NameSpaceConstants.WSNT_NS.getNamespaceURI(),
-                WsmgCommonConstants.SUBSCRIPTION_ID));
-        if (ite.hasNext()) {
-            OMElement identifier = (OMElement) ite.next();
-            logger.debug("extracted identifier " + identifier.getText());
-
-            context.setContextParameter(ContextParameters.SUB_ID, identifier.getText());
-
-        }
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/OutGoingMessage.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/OutGoingMessage.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/OutGoingMessage.java
deleted file mode 100644
index 73eba17..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/OutGoingMessage.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.commons;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.airavata.wsmg.broker.AdditionalMessageContent;
-import org.apache.airavata.wsmg.broker.ConsumerInfo;
-
-public class OutGoingMessage implements Serializable {
-    /**
-	 * 
-	 */
-    private static final long serialVersionUID = -6999667921413261492L;
-
-    String textMessage;
-
-    AdditionalMessageContent additionalMessageContent;
-
-    List<ConsumerInfo> consumerInfoList = null;
-
-    // ConsumerInfo consumerInfo=null;
-
-    /**
-	 * 
-	 */
-    public OutGoingMessage() {
-        // super();
-        // TODO Auto-generated constructor stub
-        // consumerInfo=new ConsumerInfo();
-    }
-
-    /**
-     * @param textMessage
-     * @param additionalMessageContent
-     * @param consumerInfoList
-     */
-    public OutGoingMessage(String textMessage, AdditionalMessageContent additionalMessageContent,
-            List<ConsumerInfo> consumerInfoList) {
-        super();
-        // TODO Auto-generated constructor stub
-        this.textMessage = textMessage;
-        this.additionalMessageContent = additionalMessageContent;
-        this.consumerInfoList = consumerInfoList;
-    }
-
-    /**
-     * @param consumerInfo
-     *            The consumerInfo to set.
-     */
-    public void addConsumerInfo(ConsumerInfo consumerInfo) {
-        // this.consumerInfo = consumerInfo;
-        if (consumerInfoList == null) {
-            consumerInfoList = new ArrayList<ConsumerInfo>();
-        }
-        consumerInfoList.add(consumerInfo);
-    }
-
-    /**
-     * @return Returns the textMessage.
-     */
-    public String getTextMessage() {
-        return textMessage;
-    }
-
-    /**
-     * @param textMessage
-     *            The textMessage to set.
-     */
-    public void setTextMessage(String textMessage) {
-        this.textMessage = textMessage;
-    }
-
-    /**
-     * @return Returns the consumerInfoList.
-     */
-    public List<ConsumerInfo> getConsumerInfoList() {
-        return consumerInfoList;
-    }
-
-    /**
-     * @param consumerInfoList
-     *            The consumerInfoList to set.
-     */
-    public void setConsumerInfoList(List<ConsumerInfo> consumerInfoList) {
-        this.consumerInfoList = consumerInfoList;
-    }
-
-    /**
-     * @return Returns the soapHeader.
-     */
-    public AdditionalMessageContent getAdditionalMessageContent() {
-        return additionalMessageContent;
-    }
-
-    /**
-     * @param soapHeader
-     *            The soapHeader to set.
-     */
-    public void setAdditionalMessageContent(AdditionalMessageContent soapHeader) {
-        this.additionalMessageContent = soapHeader;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/storage/WsmgInMemoryStorage.java
----------------------------------------------------------------------
diff --git a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/storage/WsmgInMemoryStorage.java b/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/storage/WsmgInMemoryStorage.java
deleted file mode 100644
index 6b051b1..0000000
--- a/modules/ws-messenger/messagebroker/src/main/java/org/apache/airavata/wsmg/commons/storage/WsmgInMemoryStorage.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- *
- * 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.airavata.wsmg.commons.storage;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.airavata.wsmg.broker.subscription.SubscriptionEntry;
-import org.apache.airavata.wsmg.broker.subscription.SubscriptionState;
-
-public class WsmgInMemoryStorage implements WsmgStorage, WsmgQueue {
-
-    private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();
-
-    private Map<String, SubscriptionState> expirableSubscriptions = new ConcurrentHashMap<String, SubscriptionState>();
-
-    private Map<String, SubscriptionState> unexpirableSubscriptions = new ConcurrentHashMap<String, SubscriptionState>();
-
-    public void dispose() {
-        queue.clear();
-        expirableSubscriptions.clear();
-        unexpirableSubscriptions.clear();
-    }
-
-    public int insert(SubscriptionState subscription) {
-        if (subscription.isNeverExpire()) {
-            unexpirableSubscriptions.put(subscription.getId(), subscription);
-        } else {
-            expirableSubscriptions.put(subscription.getId(), subscription);
-        }
-        return 0;
-    }
-
-    public int delete(String subscriptionId) {
-        expirableSubscriptions.remove(subscriptionId);
-        unexpirableSubscriptions.remove(subscriptionId);
-        return 0;
-    }
-
-    public List<SubscriptionEntry> getAllSubscription() {
-
-        List<SubscriptionEntry> ret = new ArrayList<SubscriptionEntry>(expirableSubscriptions.size()
-                + unexpirableSubscriptions.size());
-
-        Collection<SubscriptionState> entries = expirableSubscriptions.values();
-
-        for (SubscriptionState s : entries) {
-            SubscriptionEntry se = new SubscriptionEntry();
-            se.setSubscribeXml(s.getSubscribeXml());
-            se.setSubscriptionId(s.getId());
-            ret.add(se);
-        }
-        entries = unexpirableSubscriptions.values();
-        for (SubscriptionState s : entries) {
-            SubscriptionEntry se = new SubscriptionEntry();
-            se.setSubscribeXml(s.getSubscribeXml());
-            se.setSubscriptionId(s.getId());
-            ret.add(se);
-        }
-
-        return ret;
-    }
-
-    public Object blockingDequeue() {
-        Object obj = null;
-
-        try {
-            obj = queue.take();
-        } catch (InterruptedException ie) {
-            throw new RuntimeException("interruped exception occured", ie);
-        }
-
-        return obj;
-    }
-
-    public void cleanup() {
-        queue.clear();
-    }
-
-    public void enqueue(Object object, String trackId) {
-        queue.offer(object);
-    }
-}


Mime
View raw message