/*
    * 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.replication;
  
  import java.io.UnsupportedEncodingException;
  import java.util.ArrayList;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Queue;
  import java.util.Set;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.ConcurrentLinkedQueue;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  import javax.persistence.EntityManager;
  import javax.persistence.EntityTransaction;
  import javax.persistence.Query;
  import javax.xml.bind.JAXB;
  import javax.xml.ws.BindingProvider;
  import org.apache.commons.configuration.ConfigurationException;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.juddi.api_v3.Node;
  import org.apache.juddi.config.AppConfig;
  import org.apache.juddi.config.PersistenceManager;
  import org.apache.juddi.config.Property;
  import org.apache.juddi.mapping.MappingApiToModel;
  import org.apache.juddi.mapping.MappingModelToApi;
  import org.apache.juddi.model.ChangeRecord;
  import org.apache.juddi.model.ReplicationConfiguration;
  import org.apache.juddi.v3.client.UDDIService;
  import org.apache.juddi.v3.client.cryptor.TransportSecurityHelper;
  import org.uddi.repl_v3.ChangeRecordIDType;
  import org.uddi.repl_v3.CommunicationGraph.Edge;
  import org.uddi.repl_v3.HighWaterMarkVectorType;
  import org.uddi.repl_v3.NotifyChangeRecordsAvailable;
  import org.uddi.repl_v3.Operator;
  import org.uddi.v3_service.UDDIReplicationPortType;
  
  /**
   * Handles when local records have been changed, change journal storage and
   * notifications to all remote replication nodes that something has been
   * altered.
   *
   * @author <a href="mailto:alexoree@apache.org">Alex O'Ree</a>
   *
   */
  public class ReplicationNotifier extends TimerTask {
  
          private static Log log = LogFactory.getLog(ReplicationNotifier.class);
          private Timer timer = null;
          private long startBuffer = 5000;//AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_START_BUFFER, 20000l); // 20s startup delay default 
          private long interval = 5000;//AppConfig.getConfiguration().getLong(Property.JUDDI_NOTIFICATION_INTERVAL, 300000l); //5 min default
          private static String node = null;
          private static UDDIService uddiService = new UDDIService();
  
          /**
           * default constructor
           *
           * @throws ConfigurationException
           */
          public ReplicationNotifier() throws ConfigurationException {
                 super();
                 init();
          }
          private synchronized void init() throws ConfigurationException {
                  timer = new Timer(true);
                  startBuffer=AppConfig.getConfiguration().getLong(Property.JUDDI_REPLICATION_START_BUFFER, 5000l);
                  interval = AppConfig.getConfiguration().getLong(Property.JUDDI_REPLICATION_INTERVAL, 5000l);
                  timer.scheduleAtFixedRate(this, startBuffer, interval);
                  if (queue == null) {
                          queue = new ConcurrentLinkedQueue();
                  }
                  node = AppConfig.getConfiguration().getString(Property.JUDDI_NODE_ID, "UNDEFINED_NODE_NAME");
          }
  
          @Override
          public boolean cancel() {
                  timer.cancel();
                  //TODO notify other nodes that i'm going down
                  return super.cancel();
          }
  
         //ReplicationNotifier.Enqueue(this);
         public synchronized static void enqueue(org.apache.juddi.model.ChangeRecord change) {
                 if (queue == null) {
                         queue = new ConcurrentLinkedQueue<org.apache.juddi.model.ChangeRecord>();
                 }
                 queue.add(change);
         }
 
         public synchronized static void EnqueueRetransmit(org.uddi.repl_v3.ChangeRecord change) {
                 if (queue2 == null) {
                         queue2 = new ConcurrentLinkedQueue<org.uddi.repl_v3.ChangeRecord>();
                 }
                 queue2.add(change);
         }
         static Queue<org.apache.juddi.model.ChangeRecord> queue;
         static Queue<org.uddi.repl_v3.ChangeRecord> queue2;
 
         /**
          * Note: this is for locally originated changes only, see null null null         {@link org.apache.juddi.api.impl.UDDIReplicationImpl.PullTimerTask#PersistChangeRecord PersistChangeRecord
          * } for how remote changes are processed
          *
          * @param j must be one of the UDDI save APIs
          *
          */
         protected void ProcessChangeRecord(org.apache.juddi.model.ChangeRecord j) {
                 //store and convert the changes to database model
 
                 //TODO need a switch to send the notification without persisting the record
                 //this is to support multihop notifications
                 EntityManager em = PersistenceManager.getEntityManager();
                 EntityTransaction tx = null;
                 try {
                         tx = em.getTransaction();
                         tx.begin();
                         j.setIsAppliedLocally(true);
                         em.persist(j);
                         j.setOriginatingUSN(j.getId());
                         em.merge(j);
                         log.info("CR saved locally, it was from " + j.getNodeID()
                                 + " USN:" + j.getOriginatingUSN()
                                 + " Type:" + j.getRecordType().name()
                                 + " Key:" + j.getEntityKey()
                                 + " Local id:" + j.getId());
                         tx.commit();
                 } catch (Exception ex) {
                         log.fatal("unable to store local change record locally!!", ex);
                         if (tx != null && tx.isActive()) {
                                 tx.rollback();
                         }
                         JAXB.marshal(MappingModelToApi.mapChangeRecord(j), System.out);
                 } finally {
                         em.close();
                 }
 
                 log.debug("ChangeRecord: " + j.getId() + "," + j.getEntityKey() + "," + j.getNodeID() + "," + j.getOriginatingUSN() + "," + j.getRecordType().toString());
                 SendNotifications(j.getId(), j.getNodeID(), false);
 
         }
 
         private void SendNotifications(Long id, String origin_node, boolean isRetrans) {
 
                 org.uddi.repl_v3.ReplicationConfiguration repcfg = FetchEdges();
 
                 if (repcfg == null) {
                         log.debug("No replication configuration is defined!");
                         return;
 
                 }
                 if (id == null || origin_node == null) {
                         log.fatal("Either the id is null or the origin_node is null. I can't send out this alert!!");
                         //throw new Exception(node);
                         return;
                 }
 
                 Set<Object> destinationUrls = new HashSet<Object>();
 
                 /**
                  * In the absence of a communicationGraph element from the
                  * Replication Configuration Structure (although it's mandatory
                  * in the xsd), all nodes listed in the node element MAY send
                  * any and all messages to any other node of the registry.
                  */
                 if (repcfg.getCommunicationGraph() == null
                         || repcfg.getCommunicationGraph().getEdge().isEmpty() && !isRetrans) {
                         //no edges or graph defined, default to the operator list
                         //retransmission only applies to non-directed-edge replication, thus the extra check
                         for (Operator o : repcfg.getOperator()) {
                                 //no need to tell myself about a change at myself or the origin
                                 if (!o.getOperatorNodeID().equalsIgnoreCase(node) && !o.getOperatorNodeID().equalsIgnoreCase(origin_node)) {
                                         destinationUrls.add(o.getSoapReplicationURL());
                                 }
                         }
                 } else {
                         //this is for directed graph replication
                         //find all nodes that i need to notify
                         Iterator<Edge> iterator = repcfg.getCommunicationGraph().getEdge().iterator();
                         while (iterator.hasNext()) {
                                 Edge next = iterator.next();
 
                                 if (next.getMessageSender().equalsIgnoreCase(node)) {
 
                                         //this is my server and i need to transmit the notification to
                                         String messageReceiver = next.getMessageReceiver();
                                         PrimaryAlternate container = new PrimaryAlternate();
                                         //pointless to send a notification to myself or the origin
                                         if (!messageReceiver.equalsIgnoreCase(node) && !messageReceiver.equalsIgnoreCase(origin_node)) {
                                                 //look up the endpoint urls
                                                 for (int x = 0; x < repcfg.getOperator().size(); x++) {
                                                         if (repcfg.getOperator().get(x).getOperatorNodeID().equalsIgnoreCase(messageReceiver)) {
                                                                 container.primaryUrl = repcfg.getOperator().get(x).getSoapReplicationURL();
                                                         }
                                                 }
                                                 for (int y = 0; y < next.getMessageReceiverAlternate().size(); y++) {
                                                         for (int x = 0; x < repcfg.getOperator().size(); x++) {
                                                                 if (repcfg.getOperator().get(x).getOperatorNodeID().equalsIgnoreCase(next.getMessageReceiverAlternate().get(y))) {
                                                                         container.alternateUrls.add(repcfg.getOperator().get(x).getSoapReplicationURL());
                                                                 }
                                                         }
                                                 }
                                         }
                                         if (container.primaryUrl != null) {
                                                 destinationUrls.add(container);
                                         } else {
                                                 log.warn("Unable to find primary url for directed edge graph replication from this node " + node + " to "
                                                         + "destination node " + next.getMessageReceiver() + " it will be ignored!");
                                         }
 
                                 }
 
                         }
 
                 }
 
                 if (destinationUrls.isEmpty()) {
                         log.debug("Something is bizarre with the replication config. I should have had at least one node to notify, but I have none!");
                         return;
                 }
                 UDDIReplicationPortType x = uddiService.getUDDIReplicationPort();
                 TransportSecurityHelper.applyTransportSecurity((BindingProvider) x);
 
                 for (Object s : destinationUrls) {
 
                         NotifyChangeRecordsAvailable req = new NotifyChangeRecordsAvailable();
 
                         req.setNotifyingNode(node);
                         HighWaterMarkVectorType highWaterMarkVectorType = new HighWaterMarkVectorType();
 
                         highWaterMarkVectorType.getHighWaterMark().add(new ChangeRecordIDType(origin_node, id));
                         req.setChangesAvailable(highWaterMarkVectorType);
 
                         if (s instanceof String) {
                                 SendNotification(x, (String) s, req);
                         } else if (s instanceof PrimaryAlternate) {
                                 //more complex directed graph stuff
                                 PrimaryAlternate pa = (PrimaryAlternate) s;
                                 if (!SendNotification(x, pa.primaryUrl, req)) {
                                         for (String url : pa.alternateUrls) {
                                                 if (SendNotification(x, url, req)) {
                                                         break;
                                                 }
                                                 //no need to continue to additional alternates
                                         }
                                 } else {
                                         //primary url succeeded, no further action required
                                 }
 
                         }
 
                         //TODO the spec talks about control messages, should we even support it? seems pointless
                 }
 
         }
 
         /**
          * return true if successful
          *
          * @param x
          * @param s
          * @param req
          * @return
          */
         private boolean SendNotification(UDDIReplicationPortType x, String s, NotifyChangeRecordsAvailable req) {
                 ((BindingProvider) x).getRequestContext().put(BindingProvider.ENDPOINT_ADDRESS_PROPERTY, s);
                 try {
                         x.notifyChangeRecordsAvailable(req);
                         log.info("Successfully sent change record available message to " + s + " this node: " + node);
                         return true;
                 } catch (Exception ex) {
                         log.warn("Unable to send change notification to " + s + " this node: " + node + " reason: " + ex.getMessage());
                         log.debug("Unable to send change notification to " + s, ex);
                 }
                 return false;
         }
 
         private static class PrimaryAlternate {
 
                 String primaryUrl = null;
                 List<String> alternateUrls = new ArrayList<String>();
         }
 
         public synchronized void run() {
                 log.debug("Replication thread triggered");
                 if (queue == null) {
                         queue = new ConcurrentLinkedQueue();
                 }
                 if (queue2 == null) {
                         queue2 = new ConcurrentLinkedQueue();
                 }
                 //TODO revisie this
                 if (!queue.isEmpty()) {
                         log.info("Replication, Notifying nodes of new change records. " + queue.size() + " queued");
                 }
 
                 //TODO check for replication config changes
                 while (!queue.isEmpty()) {
                         //for each change at this node
 
                         ChangeRecord j = queue.poll();
                         ProcessChangeRecord(j);
 
                 }
 
                 while (!queue2.isEmpty()) {
                         //for each change at this node
 
                         org.uddi.repl_v3.ChangeRecord j = queue2.poll();
 
                         ChangeRecord model = new ChangeRecord();
                         try {
                                 model = MappingApiToModel.mapChangeRecord(j);
                         } catch (UnsupportedEncodingException ex) {
                                 Logger.getLogger(ReplicationNotifier.class.getName()).log(Level.SEVERE, null, ex);
                         }
                         log.info("retransmitting CR notificationm entity owner: " + j.getChangeID().getNodeID() + " CR: " + j.getChangeID().getOriginatingUSN() + " key:" + model.getEntityKey() + " " + model.getRecordType().name() + " accepted locally:" + model.getIsAppliedLocally());
                         SendNotifications(j.getChangeID().getOriginatingUSN(), j.getChangeID().getNodeID(), true);
 
                 }
         }
 
         /**
          * returns the latest version of the replication config or null if there
          * is no config
          *
          * @return
          */
         public static org.uddi.repl_v3.ReplicationConfiguration FetchEdges() {
 
                 EntityManager em = PersistenceManager.getEntityManager();
                 EntityTransaction tx = null;
                 org.uddi.repl_v3.ReplicationConfiguration item = new org.uddi.repl_v3.ReplicationConfiguration();
                 try {
                         tx = em.getTransaction();
                         tx.begin();
                         Query q = em.createQuery("SELECT item FROM ReplicationConfiguration item order by item.serialNumber DESC");
                         q.setMaxResults(1);
                         ReplicationConfiguration results = (ReplicationConfiguration) q.getSingleResult();
                         //   ReplicationConfiguration find = em.find(ReplicationConfiguration.class, null);
                         if (results != null) {
                                 MappingModelToApi.mapReplicationConfiguration(results, item);
                         }
 
                         tx.commit();
                         return item;
                 } catch (Exception ex) {
                         //log.error("error", ex);
                         //no config available
                         if (tx != null && tx.isActive()) {
                                 tx.rollback();
                         }
                 } finally {
                         em.close();
                 }
                 return null;
         }
 
         @Deprecated
         private Node getNode(String messageSender) {
                 EntityManager em = PersistenceManager.getEntityManager();
                 EntityTransaction tx = null;
                 try {
                         tx = em.getTransaction();
                         tx.begin();
                         Node api = new Node();
                         org.apache.juddi.model.Node find = em.find(org.apache.juddi.model.Node.class, messageSender);
                         if (find != null) {
                                 MappingModelToApi.mapNode(find, api);
                         }
                         tx.commit();
                         return api;
                 } catch (Exception ex) {
                         log.error("error", ex);
                         if (tx != null && tx.isActive()) {
                                 tx.rollback();
                         }
                 } finally {
                         em.close();
                 }
                 return null;
         }
 }