/*

  * Copyright 2006-2011 The Kuali Foundation

  *

  * Licensed under the Educational Community 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.opensource.org/licenses/ecl2.php

  *

  * 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.kuali.rice.kcb.quartz;

 import org.apache.log4j.Logger;

 import org.kuali.rice.core.framework.persistence.dao.GenericDao;

 import org.kuali.rice.kcb.bo.Message;

 import org.kuali.rice.kcb.bo.MessageDelivery;

 import org.kuali.rice.kcb.bo.MessageDeliveryStatus;

 import org.kuali.rice.kcb.deliverer.BulkMessageDeliverer;

 import org.kuali.rice.kcb.deliverer.MessageDeliverer;

 import org.kuali.rice.kcb.exception.MessageDeliveryProcessingException;

 import org.kuali.rice.kcb.quartz.ProcessingResult.Failure;

 import org.kuali.rice.kcb.service.GlobalKCBServiceLocator;

 import org.kuali.rice.kcb.service.MessageDelivererRegistryService;

 import org.kuali.rice.kcb.service.MessageDeliveryService;

 import org.kuali.rice.kcb.service.MessageService;

 import org.quartz.JobExecutionContext;

 import org.quartz.JobExecutionException;

 import org.quartz.StatefulJob;

 import org.springframework.beans.factory.annotation.Required;

 import java.util.ArrayList;

 import java.util.Collection;

 import java.util.HashMap;

 import java.util.HashSet;

 import java.util.LinkedList;

 import java.util.List;

 import java.util.Map;

 import java.util.Set;

 /**

  * Job that delivers messages to endpoints.  This job is not really stateful,

  * but should not be executed concurrently.

  * 

  * @author Kuali Rice Team (rice.collab@kuali.org)

  */

 public class MessageProcessingJob extends ConcurrentJob<MessageDelivery> implements StatefulJob {

     public static final String NAME = "MessageProcessingJobDetail";

     public static final String GROUP = "KCB-Delivery";

     public static enum Mode {

         DELIVER, REMOVE

     }

     private static final Logger LOG = Logger.getLogger(MessageProcessingJob.class);

     private GenericDao dao;

     private MessageDelivererRegistryService registry;

     private MessageDeliveryService messageDeliveryService;

     private Long messageId;

     private Mode mode = null;

     private String user;

     private String cause;

     public MessageProcessingJob(Long messageId, Mode mode, String user, String cause) {

         this();

         this.messageId = messageId;

         this.mode = mode;

         this.user = user;

         this.cause = cause;

     }

     public MessageProcessingJob() {

         dao = GlobalKCBServiceLocator.getInstance().getKcbGenericDao();

         registry = GlobalKCBServiceLocator.getInstance().getMessageDelivererRegistryService();

         messageDeliveryService = GlobalKCBServiceLocator.getInstance().getMessageDeliveryService();

         txManager = GlobalKCBServiceLocator.getInstance().getTransactionManager();

     }

     /**

      * Sets the {@link GenericDao}

      * @param dao the {@link GenericDao}

      */

     @Required

     public void setGenericDao(GenericDao dao) {

         this.dao = dao;

     }

     /**

      * Sets the {@link MessageDelivererRegistryService}

      * @param registry the {@link MessageDelivererRegistryService}

      */

     @Required

     public void setMessageDelivererRegistry(MessageDelivererRegistryService registry) {

         this.registry = registry;

     }

     /**

      * Sets the {@link MessageDeliveryService}

      * @param messageDeliveryService the {@link MessageDeliveryService}

      */

     @Required

     public void setMessageDeliveryService(MessageDeliveryService messageDeliveryService) {

         this.messageDeliveryService = messageDeliveryService;

     }

     @Override

     protected Collection<MessageDelivery> takeAvailableWorkItems() {

         MessageDeliveryStatus[] statuses;

         switch (mode) {

             case DELIVER: {

                 statuses = new MessageDeliveryStatus[] { MessageDeliveryStatus.UNDELIVERED };

                 break;

             }

             case REMOVE: {

                 if (messageId == null) {

                     throw new IllegalStateException("Message id must be specified for message removal mode");

                 }

                 statuses = new MessageDeliveryStatus[] { MessageDeliveryStatus.DELIVERED, MessageDeliveryStatus.UNDELIVERED };

                 break;

             }

             default:

                 throw new RuntimeException("Invalid mode: " + mode);

         }

         for (MessageDeliveryStatus status: statuses) {

             LOG.debug("Taking message deliveries with status: " + status);

         }

         Collection<MessageDelivery> ds = messageDeliveryService.lockAndTakeMessageDeliveries(messageId, statuses);

         LOG.debug("Took " + ds.size() + " deliveries");

         for (MessageDelivery md: ds) {

             LOG.debug(md);

             md.setProcessCount(md.getProcessCount().intValue() + 1);

         }

         return ds;

     }

     @Override

     protected void unlockWorkItem(MessageDelivery item) {

         item.setLockedDate(null);

         dao.save(item);

     }

     /**

      * Group work items by deliverer and notification, so that deliveries to bulk deliverers are grouped

      * by notification

      * @see org.kuali.rice.ken.service.impl.ConcurrentJob#groupWorkItems(java.util.Collection)

      */

     @Override

     protected Collection<Collection<MessageDelivery>> groupWorkItems(Collection<MessageDelivery> workItems, ProcessingResult<MessageDelivery> result) {

         Collection<Collection<MessageDelivery>> groupedWorkItems = new ArrayList<Collection<MessageDelivery>>(workItems.size());

         Map<String, Collection<MessageDelivery>> bulkWorkUnits = new HashMap<String, Collection<MessageDelivery>>();

         for (MessageDelivery messageDelivery: workItems) {

             MessageDeliverer deliverer = registry.getDeliverer(messageDelivery);

             if (deliverer == null) {

                 LOG.error("Error obtaining message deliverer for message delivery: " + messageDelivery);

                 result.addFailure(new Failure<MessageDelivery>(messageDelivery, "Error obtaining message deliverer for message delivery"));

                 unlockWorkItemAtomically(messageDelivery);

                 continue;

             }

             if (deliverer instanceof BulkMessageDeliverer) {

                 // group by bulk-deliverer+message combo

                 String key = messageDelivery.getDelivererTypeName() + ":" + messageDelivery.getMessage().getId();

                 Collection<MessageDelivery> workUnit = bulkWorkUnits.get(key);

                 if (workUnit == null) {

                     workUnit = new LinkedList<MessageDelivery>();

                     bulkWorkUnits.put(key, workUnit);

                 }

                 workUnit.add(messageDelivery);

             } else {

                 ArrayList<MessageDelivery> l = new ArrayList<MessageDelivery>(1);

                 l.add(messageDelivery);

                 groupedWorkItems.add(l);

             }

         }

         return groupedWorkItems;

     }

     @Override

     protected Collection<MessageDelivery> processWorkItems(Collection<MessageDelivery> messageDeliveries) {

         MessageDelivery firstMessageDelivery = messageDeliveries.iterator().next();

         // get our hands on the appropriate MessageDeliverer instance

         MessageDeliverer messageDeliverer = registry.getDeliverer(firstMessageDelivery);

         if (messageDeliverer == null) {

             throw new RuntimeException("Message deliverer could not be obtained");

         }

         if (messageDeliveries.size() > 1) {

             // this is a bulk deliverer, so we need to batch the MessageDeliveries

             if (!(messageDeliverer instanceof BulkMessageDeliverer)) {

                 throw new RuntimeException("Discrepency in dispatch service: deliverer for list of message deliveries is not a BulkMessageDeliverer");

             }

             return bulkProcess((BulkMessageDeliverer) messageDeliverer, messageDeliveries, mode);

         } else {

             return process(messageDeliverer, firstMessageDelivery, mode);

         }

     }

     /**

      * Implements delivery of a single MessageDelivery

      * @param deliverer the deliverer

      * @param messageDelivery the delivery

      * @return collection of strings indicating successful deliveries

      */

     protected Collection<MessageDelivery> process(MessageDeliverer messageDeliverer, MessageDelivery messageDelivery, Mode mode) {

         // we have our message deliverer, so tell it to deliver the message

         try {

             if (mode == Mode.DELIVER) {

                 messageDeliverer.deliver(messageDelivery);

                 // if processing was successful, set the count back to zero

                 messageDelivery.setProcessCount(Integer.valueOf(0));

                 // by definition we have succeeded at this point if no exception was thrown by the messageDeliverer

                 // so update the status of the delivery message instance to DELIVERED (and unmark as taken)

                 // and persist

                 updateStatusAndUnlock(messageDelivery, mode == Mode.DELIVER ? MessageDeliveryStatus.DELIVERED : MessageDeliveryStatus.REMOVED);

             } else {

                 messageDeliverer.dismiss(messageDelivery, user, cause);

                 // don't need to set the processing count down to zero because we are just deleting the record entirely

                 messageDeliveryService.deleteMessageDelivery(messageDelivery);

             }

         } catch (MessageDeliveryProcessingException nmde) {

             LOG.error("Error processing message delivery " + messageDelivery, nmde);

             throw new RuntimeException(nmde);

         }

         LOG.debug("Message delivery '" + messageDelivery.getId() + "' for message '" + messageDelivery.getMessage().getId() + "' was successfully processed.");

         //PerformanceLog.logDuration("Time to dispatch notification delivery for notification " + messageDelivery.getMessage().getId(), System.currentTimeMillis() - messageDelivery.getNotification().getSendDateTime().getTime());

         List<MessageDelivery> success = new ArrayList<MessageDelivery>(1);

         success.add(messageDelivery);

         return success;

     }

     /**

      * Implements bulk delivery of a collection of {@link MessageDelivery}s

      * @param deliverer the deliverer

      * @param messageDeliveries the deliveries

      * @return collection of strings indicating successful deliveries

      */

     protected Collection<MessageDelivery> bulkProcess(BulkMessageDeliverer messageDeliverer, Collection<MessageDelivery> messageDeliveries,  Mode mode) {

         MessageDeliveryStatus targetStatus = (mode == Mode.DELIVER ? MessageDeliveryStatus.DELIVERED : MessageDeliveryStatus.REMOVED);

         // we have our message deliverer, so tell it to deliver the message

         try {

             if (mode == Mode.DELIVER) {

                 messageDeliverer.bulkDeliver(messageDeliveries);

             } else {

                 messageDeliverer.bulkDismiss(messageDeliveries);

             }

         } catch (MessageDeliveryProcessingException nmde) {

             LOG.error("Error bulk-delivering messages " + messageDeliveries, nmde);

             throw new RuntimeException(nmde);

         }

         // by definition we have succeeded at this point if no exception was thrown by the messageDeliverer

         // so update the status of the delivery message instance to DELIVERED (and unmark as taken)

         // and persist

         List<MessageDelivery> successes = new ArrayList<MessageDelivery>(messageDeliveries.size());

         for (MessageDelivery nmd: messageDeliveries) {

             successes.add(nmd);

             LOG.debug("Message delivery '" + nmd.getId() + "' for notification '" + nmd.getMessage().getId() + "' was successfully delivered.");

             //PerformanceLog.logDuration("Time to dispatch notification delivery for notification " + nmd.getMessage().getId(), System.currentTimeMillis() - nmd.getNotification().getSendDateTime().getTime());

             if (mode == Mode.REMOVE) {

                 messageDeliveryService.deleteMessageDelivery(nmd);

             } else {

                 nmd.setProcessCount(0);

                 updateStatusAndUnlock(nmd, targetStatus);                

             }

         }

         return successes;

     }

     @Override

     protected void finishProcessing(ProcessingResult<MessageDelivery> result) {

         LOG.debug("Message processing job: " + result.getSuccesses().size() + " processed, " + result.getFailures().size() + " failures");

         Set<Long> messageIds = new HashSet<Long>(result.getSuccesses().size());

         for (MessageDelivery md: result.getSuccesses()) {

             messageIds.add(md.getMessage().getId());

         }

         MessageService ms = GlobalKCBServiceLocator.getInstance().getMessageService();

         for (Long id: messageIds) {

             LOG.debug("Finishing processing message " + id);

             //if (Mode.REMOVE == mode) {

             Message m = ms.getMessage(id);

             Collection<MessageDelivery> c = messageDeliveryService.getMessageDeliveries(m);

             if (c.size() == 0) {

                 LOG.debug("Deleting message " + m);

                 ms.deleteMessage(m);

             } else {

                 LOG.debug("Message " + m.getId() + " has " + c.size() + " deliveries");

                 for (MessageDelivery md: c) {

                     LOG.debug(md);

                 }

             }

         }

     }

     /**

      * Marks a MessageDelivery as having been delivered, and unlocks it

      * @param messageDelivery the messageDelivery instance to mark

      */

     protected void updateStatusAndUnlock(MessageDelivery messageDelivery, MessageDeliveryStatus status) {

         messageDelivery.setDeliveryStatus(status);

         // mark as unlocked

         messageDelivery.setLockedDate(null);

         dao.save(messageDelivery);

     }

     @Override

     public ProcessingResult<MessageDelivery> run() {

         LOG.debug("MessageProcessingJob running in Thread " + Thread.currentThread() + ": " + mode + " " + user + " " + cause);

         return super.run();

     }

     public void execute(JobExecutionContext context) throws JobExecutionException {

         String mode = context.getMergedJobDataMap().getString("mode");

         if (mode != null) {

             this.mode = Mode.valueOf(mode);

         } else {

             this.mode = Mode.DELIVER;

         }

         this.user = context.getMergedJobDataMap().getString("user");

         this.cause = context.getMergedJobDataMap().getString("cause");

         if (context.getMergedJobDataMap().containsKey("messageId")) {

             this.messageId = context.getMergedJobDataMap().getLong("messageId");

         }

         LOG.debug("==== message processing job: " + this.mode + " message id: " + this.messageId + "====");

         super.run();

     }

 }