This project has retired. For details please refer to its Attic page.
SubscriptionNotifier.java

SubscriptionNotifier.java

/*
 * Copyright 2001-2008 The Apache Software Foundation.
 * 
 * Licensed 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.juddi.subscription;

import java.io.IOException;
import java.util.Collection;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.Query;
import javax.xml.datatype.DatatypeConfigurationException;
import javax.xml.datatype.DatatypeFactory;
import javax.xml.datatype.Duration;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.juddi.api.impl.ServiceCounterLifecycleResource;
import org.apache.juddi.api.impl.UDDIPublicationImpl;
import org.apache.juddi.api.impl.UDDISecurityImpl;
import org.apache.juddi.api.impl.UDDIServiceCounter;
import org.apache.juddi.api.impl.UDDISubscriptionImpl;
import org.apache.juddi.api_v3.AccessPointType;
import org.apache.juddi.config.AppConfig;
import org.apache.juddi.config.PersistenceManager;
import org.apache.juddi.config.Property;
import org.apache.juddi.model.Subscription;
import org.apache.juddi.model.UddiEntityPublisher;
import org.apache.juddi.subscription.notify.Notifier;
import org.apache.juddi.subscription.notify.NotifierFactory;
import org.uddi.sub_v3.CoveragePeriod;
import org.uddi.sub_v3.GetSubscriptionResults;
import org.uddi.sub_v3.SubscriptionResultsList;
import org.uddi.subr_v3.NotifySubscriptionListener;
import org.uddi.v3_service.DispositionReportFaultMessage;

/**
 *
 * @author <a href="mailto:kstam@apache.org">Kurt T Stam</a>
 *
 */
public class SubscriptionNotifier extends TimerTask {

        private static final Log log = LogFactory.getLog(SubscriptionNotifier.class);
        private Timer timer = null;
        private final long startBuffer = AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_START_BUFFER, 20000l); // 20s startup delay default 
        private final long interval = AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_INTERVAL, 300000l); //5 min default
        private final long acceptableLagTime = AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_ACCEPTABLE_LAGTIME, 1000l); //1000 milliseconds
        private final int maxTries = AppConfig.getConfiguration().getInt(Property.JUDDI_NOTIFICATION_MAX_TRIES, 3);
        private final long badListResetInterval = AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_LIST_RESET_INTERVAL, 1000l * 3600); //one hour
        /**
         * @since 3.2
         */
        private final boolean sendToken = AppConfig.getConfiguration().getBoolean(Property.JUDDI_NOTIFICATION_SENDAUTHTOKEN, false);
        private final UDDISubscriptionImpl subscriptionImpl = new UDDISubscriptionImpl();
        private final Boolean alwaysNotify = false;
        private Date desiredDate = null;
        private int lastUpdateCounter;
        private final UDDIServiceCounter serviceCounter = ServiceCounterLifecycleResource.getServiceCounter(UDDIPublicationImpl.class);
        private final String[] attributes = {
                "save_business", "save_service", "save_binding", "save_tmodel",
                "delete_business", "delete_service", "delete_binding", "delete_tmodel",
                "add_publisherassertions", "set_publisherassertions", "delete_publisherassertions"
        };
        private static Map<String, Integer> badNotifications = new ConcurrentHashMap<String, Integer>();
        private static Date lastBadNotificationReset = new Date();

        /**
         * default constructor
         *
         * @throws ConfigurationException
         */
        public SubscriptionNotifier() throws ConfigurationException {
                super();
                timer = new Timer(true);
                timer.scheduleAtFixedRate(this, startBuffer, interval);
        }

        @Override
        public boolean cancel() {
                timer.cancel();
                return super.cancel();
        }

        /**
         * If the CRUD methods on the publication API where not called, this
         * registry node does not contain changes. If the registry database is
         * shared with other registry nodes and one of those registries pushed
         * in a change, then that registry node will take care of sending out
         * notifications.
         *
         * @return true/false
         */
        protected boolean registryMayContainUpdates() {
                boolean isUpdated = false;
                int updateCounter = 0;
		//if the desiredDate is set it means that we've declined sending out a notification before
                //because the a client did not want a notification yet. However if this desired
                //notification time has come we should try sending out the notification now.
                if (desiredDate != null && new Date().getTime() > desiredDate.getTime()) {
                        return true;
                }
                try {
                        for (String attribute : attributes) {
                                String counter = serviceCounter.getAttribute(attribute + " successful queries").toString();
                                updateCounter += Integer.valueOf(counter);
                        }
			// if the counts are not the same something has changed, 
                        // this accounts for the case where the counters where reset.
                        if (updateCounter != lastUpdateCounter) {
                                lastUpdateCounter = updateCounter;
                                isUpdated = true;
                        }
                } catch (Exception e) {
                        log.error(e.getMessage(), e);
                }
                return isUpdated;
        }

        @Override
        public synchronized void run() {
                if (badListResetInterval > 0 && new Date().getTime() > lastBadNotificationReset.getTime() + badListResetInterval) {
                        badNotifications = new ConcurrentHashMap<String, Integer>();
                        lastBadNotificationReset = new Date();
                        log.debug("badNotificationList was reset");
                }
                if ((firedOnTime(scheduledExecutionTime()) || alwaysNotify) && registryMayContainUpdates()) {
                        long startTime = System.currentTimeMillis();
                        desiredDate = null;
                        log.info("Start Notification background task; checking if subscription notifications need to be send out..");

                        Collection<Subscription> subscriptions = getAllAsyncSubscriptions();
                        for (Subscription subscription : subscriptions) {

                                if (subscription.getExpiresAfter() == null || subscription.getExpiresAfter().getTime() > startTime
                                        || !isTemporarilyDisabled(subscription.getSubscriptionKey())) {
                                        try {
						//build a query with a coverage period from the lastNotified time to 
                                                //now (the scheduled Execution time)
                                                Date notificationDate = new Date(scheduledExecutionTime());
                                                GetSubscriptionResults getSubscriptionResults
                                                        = buildGetSubscriptionResults(subscription, notificationDate);
                                                if (getSubscriptionResults != null) {
                                                        getSubscriptionResults.setSubscriptionKey(subscription.getSubscriptionKey());
                                                        UddiEntityPublisher publisher = new UddiEntityPublisher();
                                                        publisher.setAuthorizedName(subscription.getAuthorizedName());
                                                        SubscriptionResultsList resultList = subscriptionImpl.getSubscriptionResults(getSubscriptionResults, publisher);
                                                        String token = resultList.getChunkToken();
                                                        if (resultListContainsChanges(resultList)) {
                                                                log.info("We have a change and need to notify " + subscription.getSubscriptionKey());
                                                                resultList.setChunkToken(null);
                                                                //Note that the chunkToken is not returned with this structure for this API.  
                                                                notify(getSubscriptionResults, resultList, notificationDate);
                                                        } else {
                                                                log.info("No changes where recorded, no need to notify.");
                                                        }
                                                        while (!token.equalsIgnoreCase("0")) {
                                                                resultList = subscriptionImpl.getSubscriptionResults(getSubscriptionResults, publisher);
                                                                if (resultListContainsChanges(resultList)) {
                                                                        log.info("We have a change and need to notify " + subscription.getSubscriptionKey());
                                                                        resultList.setChunkToken(null);
                                                                        //Note that the chunkToken is not returned with this structure for this API.  
                                                                        notify(getSubscriptionResults, resultList, notificationDate);
                                                                } else {
                                                                        log.info("No changes where recorded, no need to notify.");
                                                                }
                                                        }

                                                }
                                        } catch (Exception e) {
                                                log.error("Could not obtain subscriptionResult for subscriptionKey "
                                                        + subscription.getSubscriptionKey() + ". " + e.getMessage(), e);
                                        }
                                } else {
                                        // the subscription expired, we should delete it
                                        log.info("Subcription with key " + subscription.getSubscriptionKey()
                                                + " expired " + subscription.getExpiresAfter());
                                        deleteSubscription(subscription);
                                }
                        }
                        long endTime = System.currentTimeMillis();

                        if ((endTime - startTime) > interval) {
                                log.info("Notification background task duration exceeds the JUDDI_NOTIFICATION_INTERVAL"
                                        + " of " + interval + ". Notification background task took "
                                        + (endTime - startTime) + " milliseconds.");
                        } else {
                                log.info("Notification background task took " + (endTime - startTime) + " milliseconds.");
                        }
                } else {
                        log.debug("Skipping current notification cycle because lagtime is too great.");
                }
        }

        /**
         * Checks to see that the event are fired on time. If they are late this
         * may indicate that the server is under load. The acceptableLagTime is
         * configurable using the "juddi.notification.acceptable.lagtime"
         * property and is defaulted to 500ms. A negative value means that you
         * do not care about the lag time and you simply always want to go do
         * the notification work.
         *
         * @param scheduleExecutionTime
         * @return true if the server is within the acceptable latency lag.
         */
        private boolean firedOnTime(long scheduleExecutionTime) {
                long lagTime = System.currentTimeMillis() - scheduleExecutionTime;
                if (lagTime <= acceptableLagTime || acceptableLagTime < 0) {
                        return true;
                } else {
                        log.debug("NotificationTimer is lagging " + lagTime + " milli seconds behind. A lag time "
                                + "which exceeds an acceptable lagtime of " + acceptableLagTime + "ms indicates "
                                + "that the registry server is under load or was in sleep mode. We are therefore skipping this notification "
                                + "cycle.");
                        return false;
                }
        }

        protected GetSubscriptionResults buildGetSubscriptionResults(Subscription subscription, Date endPoint)
                throws DispositionReportFaultMessage, DatatypeConfigurationException {

                GetSubscriptionResults getSubscriptionResults = null;
                Duration duration = TypeConvertor.convertStringToDuration(subscription.getNotificationInterval());
                Date startPoint = subscription.getLastNotified();
                Date nextDesiredNotificationDate = null;
                if (startPoint == null) {
                        startPoint = subscription.getCreateDate();
                }
                nextDesiredNotificationDate = new Date(startPoint.getTime());
                duration.addTo(nextDesiredNotificationDate);
		//nextDesiredNotificationDate = lastTime + the Interval Duration, which should be:
                //AFTER the lastNotified time and BEFORE the endTime (current time). If it is
                //after the endTime, then the user does not want a notification yet, so we accumulate.
                if (subscription.getLastNotified() == null || nextDesiredNotificationDate.after(startPoint) && nextDesiredNotificationDate.before(endPoint)) {
                        getSubscriptionResults = new GetSubscriptionResults();
                        CoveragePeriod period = new CoveragePeriod();
                        GregorianCalendar calendar = new GregorianCalendar();
                        calendar.setTimeInMillis(startPoint.getTime());
                        period.setStartPoint(DatatypeFactory.newInstance().newXMLGregorianCalendar(calendar));
                        calendar.setTimeInMillis(endPoint.getTime());
                        period.setEndPoint(DatatypeFactory.newInstance().newXMLGregorianCalendar(calendar));
                        if (log.isDebugEnabled()) {
                                log.debug("Period " + period.getStartPoint() + " " + period.getEndPoint());
                        }
                        getSubscriptionResults.setCoveragePeriod(period);
                } else {
                        log.info("Client does not yet want a notification. The next desidered notification Date " + nextDesiredNotificationDate + ". The current interval ["
                                + startPoint + " , " + endPoint + "] therefore skipping this notification cycle.");
                        if (desiredDate == null || nextDesiredNotificationDate.getTime() < desiredDate.getTime()) {
                                desiredDate = nextDesiredNotificationDate;
                        }
                }
                return getSubscriptionResults;

        }

        protected boolean resultListContainsChanges(SubscriptionResultsList resultList) {
                if (resultList == null) {
                        return false;
                }
                if (resultList.getBindingDetail() != null || resultList.getBusinessDetail() != null
                        || resultList.getBusinessList() != null || resultList.getServiceDetail() != null
                        || resultList.getServiceList() != null || resultList.getTModelDetail() != null
                        || resultList.getTModelList() != null || resultList.getRelatedBusinessesList() != null
                        || resultList.getAssertionStatusReport() !=null) {
                        return true;
                }
                //When the response is 'brief', or when there are deleted only keyBags are used.
                if (resultList.getKeyBag() != null && resultList.getKeyBag().size() > 0) {
                        return true;
                }
                //there are no changes to what was subscribed to
                return false;
        }

        /**
         * Obtains all subscriptions in the system.
         *
         * @return Collection of All Subscriptions in the system.
         */
        @SuppressWarnings("unchecked")
        protected Collection<Subscription> getAllAsyncSubscriptions() {
                Collection<Subscription> subscriptions = null;
                EntityManager em = PersistenceManager.getEntityManager();
                EntityTransaction tx = em.getTransaction();
                try {
                        tx.begin();
                        Query query = em.createQuery("SELECT s FROM Subscription s WHERE s.bindingKey IS NOT NULL");
                        subscriptions = (Collection<Subscription>) query.getResultList();
                        tx.commit();
                } finally {
                        if (tx.isActive()) {
                                tx.rollback();
                        }
                        em.close();
                }
                return subscriptions;
        }

        /**
         * Deletes the subscription. i.e. when it is expired.
         *
         * @param subscription
         */
        protected void deleteSubscription(Subscription subscription) {
                EntityManager em = PersistenceManager.getEntityManager();
                EntityTransaction tx = em.getTransaction();
                try {
                        tx.begin();
                        em.remove(subscription);
                        tx.commit();
                } finally {
                        if (tx.isActive()) {
                                tx.rollback();
                        }
                        em.close();
                }
        }

        /**
         * Sends out the notifications.
         *
         * @param getSubscriptionResults
         * @param resultList
         * @param notificationDate
         */
        protected void notify(GetSubscriptionResults getSubscriptionResults, SubscriptionResultsList resultList, Date notificationDate) {
                EntityManager em = PersistenceManager.getEntityManager();
                EntityTransaction tx = em.getTransaction();
                try {
                        String subscriptionKey = resultList.getSubscription().getSubscriptionKey();
                        org.apache.juddi.model.Subscription modelSubscription
                                = em.find(org.apache.juddi.model.Subscription.class, subscriptionKey);
                        Date lastNotifiedDate = modelSubscription.getLastNotified();
                        //now log to the db that we are sending the notification.
                        tx.begin();
                        modelSubscription.setLastNotified(notificationDate);
                        em.persist(modelSubscription);
                        tx.commit();

                        org.apache.juddi.model.BindingTemplate bindingTemplate = em.find(org.apache.juddi.model.BindingTemplate.class, modelSubscription.getBindingKey());
                        NotifySubscriptionListener body = new NotifySubscriptionListener();
//			if (resultList.getServiceList()!=null && resultList.getServiceList().getServiceInfos()!=null &&
//					resultList.getServiceList().getServiceInfos().getServiceInfo().size() == 0) {
//				resultList.getServiceList().setServiceInfos(null);
//			}
                        body.setSubscriptionResultsList(resultList);

                         //TODO if the endpoint requires an auth token, look up the security endpoint of the remote registry
                        //via ClientSubscriptionInfo
                        if (sendToken) {
                                String authorizedName = modelSubscription.getAuthorizedName();
                                UDDISecurityImpl security = new UDDISecurityImpl();

                                if (authorizedName != null) { // add a security token if needed
                                        try {
                                                //obtain a token for this publisher
                                                org.uddi.api_v3.AuthToken token = security.getAuthToken(authorizedName);
                                                body.setAuthInfo(token.getAuthInfo());
                                        } catch (DispositionReportFaultMessage e) {
                                                body.setAuthInfo("Failed to generate token, please contact UDDI admin");
                                                log.error(e.getMessage(), e);
                                        }
                                }
                        }

                        if (bindingTemplate != null) {
                                if (AccessPointType.END_POINT.toString().equalsIgnoreCase(bindingTemplate.getAccessPointType())
                                        || AccessPointType.WSDL_DEPLOYMENT.toString().equalsIgnoreCase(bindingTemplate.getAccessPointType())) {
                                        try {
                                                Notifier notifier = new NotifierFactory().getNotifier(bindingTemplate);
                                                if (notifier != null) {
                                                        log.info("Sending out notification to " + bindingTemplate.getAccessPointUrl());
                                                        notifier.notifySubscriptionListener(body);
                                                        //there maybe more chunks we have to send
                                                        String chunkToken = body.getSubscriptionResultsList().getChunkToken();
                                                        while (chunkToken != null) {
                                                                UddiEntityPublisher publisher = new UddiEntityPublisher();
                                                                publisher.setAuthorizedName(modelSubscription.getAuthorizedName());
                                                                log.debug("Sending out next chunk: " + chunkToken + " to " + bindingTemplate.getAccessPointUrl());
                                                                getSubscriptionResults.setChunkToken(chunkToken);
                                                                resultList = subscriptionImpl.getSubscriptionResults(getSubscriptionResults, publisher);
                                                                body.setSubscriptionResultsList(resultList);
                                                                if (resultListContainsChanges(resultList)) //if (!IsEmpty(resultList))
                                                                {
                                                                        notifier.notifySubscriptionListener(body);
                                                                }
                                                                chunkToken = body.getSubscriptionResultsList().getChunkToken();
                                                        }
                                                        //successful notification so remove from the badNotificationList
                                                        if (badNotifications.containsKey(resultList.getSubscription().getSubscriptionKey())) {
                                                                badNotifications.remove(resultList.getSubscription().getSubscriptionKey());
                                                        }
                                                }
                                        } catch (Exception e) {
                                                if (e.getCause() instanceof IOException) {
                                                        addBadNotificationToList(subscriptionKey, bindingTemplate.getAccessPointUrl());
                                                        //we could not notify so compensate the transaction above
                                                        modelSubscription.setLastNotified(lastNotifiedDate);
                                                        tx.begin();
                                                        em.persist(modelSubscription);
                                                        tx.commit();
						//} else {
                                                        //log.warn("Unexpected WebServiceException " + e.getMessage() + e.getCause());
                                                }
                                                log.error("Unexpected notification exception:" + e.getClass().getCanonicalName() + " " + e.getMessage() + " " + e.getCause());
                                                log.debug("Unexpected notification exception:" + e.getClass().getCanonicalName() + " " + e.getMessage() + " " + e.getCause(), e);
                                        }
                                } else {
                                        log.info("Binding " + bindingTemplate.getEntityKey() + " has an unsupported binding type of "
                                                + bindingTemplate.getAccessPointType() + ". Only "
                                                + AccessPointType.END_POINT.toString() + " and "
                                                + AccessPointType.WSDL_DEPLOYMENT.toString() + " are supported.");
                                        addBadNotificationToList(subscriptionKey, bindingTemplate.getAccessPointType() + " not supported");
                                }
                        } else {
                                log.info("There is no valid binding template defined for this subscription: " + modelSubscription.getBindingKey());
                                addBadNotificationToList(subscriptionKey, modelSubscription.getBindingKey() + " not found");
                        }

                } finally {
                        if (tx.isActive()) {
                                tx.rollback();
                        }
                        em.close();
                }
        }

        protected UDDISubscriptionImpl getSubscriptionImpl() {
                return subscriptionImpl;
        }

        private boolean isTemporarilyDisabled(String subscriptionKey) {
                if (maxTries > 0 && badNotifications.containsKey(subscriptionKey) && badNotifications.get(subscriptionKey) > maxTries) {
                        log.debug("Subscription " + subscriptionKey + " is temperarily disabled. The notification endpoint"
                                + " could not be reached more then " + maxTries + " times");
                        return true;
                }
                return false;
        }

        private int addBadNotificationToList(String subscriptionKey, String endPoint) {
                Integer numberOfBadNotifications = 0;
                if (badNotifications.containsKey(subscriptionKey)) {
                        numberOfBadNotifications = badNotifications.get(subscriptionKey);
                }
                badNotifications.put(subscriptionKey, ++numberOfBadNotifications);
                log.debug("bad notification number " + numberOfBadNotifications + " for subscription "
                        + subscriptionKey + " " + endPoint);
                return numberOfBadNotifications;
        }

}