trunk/discovery/distributed/cxf-discovery/src/main/java/org/apache/cxf/dosgi/discovery/zookeeper/subscribe/InterfaceMonitor.java - cxf-dosgi - Git at Google

 /**
  * 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.cxf.dosgi.discovery.zookeeper.subscribe;

 import java.io.ByteArrayInputStream;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

 import org.apache.cxf.dosgi.discovery.zookeeper.util.Utils;
 import org.apache.cxf.dosgi.endpointdesc.EndpointDescriptionParser;
 import org.apache.cxf.dosgi.endpointdesc.PropertiesMapper;
 import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.Stat;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.osgi.service.remoteserviceadmin.EndpointListener;
 import org.osgi.xmlns.rsa.v1_0.EndpointDescriptionType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Monitors ZooKeeper for changes in published endpoints.
  * <p>
  * Specifically, it monitors the node path associated with a given interface class,
  * whose data is a serialized version of an EndpointDescription, and notifies an
  * EndpointListener when changes are detected (which can then propagate the
  * notification to other EndpointListeners with a matching scope).
  * <p>
  * Note that the EndpointListener is used here as a decoupling interface for
  * convenience, and is not necessarily used according to its documented contract.
  */
 public class InterfaceMonitor implements Watcher, StatCallback {

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

     private final String znode;
     private final ZooKeeper zk;
     private final EndpointListener endpointListener;
     private final boolean recursive;
     private volatile boolean closed;

     // This map reference changes, so don't synchronize on it
     private Map<String, EndpointDescription> nodes = new HashMap<String, EndpointDescription>();

     private EndpointDescriptionParser parser;

     public InterfaceMonitor(ZooKeeper zk, String objClass, EndpointListener endpointListener, String scope) {
         this.zk = zk;
         this.znode = Utils.getZooKeeperPath(objClass);
         this.recursive = objClass == null || objClass.isEmpty();
         this.endpointListener = endpointListener;
         this.parser = new EndpointDescriptionParser();
         LOG.debug("Creating new InterfaceMonitor {} for scope [{}] and objectClass [{}]",
                 new Object[] {recursive ? "(recursive)" : "", scope, objClass});
     }

     /**
      * Returns all endpoints that are currently known to this monitor.
      *
      * @return all endpoints that are currently known to this monitor
      */
     public synchronized List<EndpointDescription> getEndpoints() {
         return new ArrayList<EndpointDescription>(nodes.values());
     }

     public void start() {
         watch();
     }

     private void watch() {
         LOG.debug("registering a ZooKeeper.exists({}) callback", znode);
         zk.exists(znode, this, this, null);
     }

     /**
      * Zookeeper Watcher interface callback.
      */
     public void process(WatchedEvent event) {
         LOG.debug("ZooKeeper watcher callback on node {} for event {}", znode, event);
         processDelta();
     }

     /**
      * Zookeeper StatCallback interface callback.
      */
     @SuppressWarnings("deprecation")
     public void processResult(int rc, String path, Object ctx, Stat stat) {
         LOG.debug("ZooKeeper callback on node: {} code: {}", znode, rc);

         switch (rc) {
         case Code.Ok:
         case Code.NoNode:
             processDelta();
             return;

         case Code.SessionExpired:
         case Code.NoAuth:
         case Code.ConnectionLoss:
             return;

         default:
             watch();
         }
     }

     private void processDelta() {
         if (closed) {
             return;
         }

         if (zk.getState() != ZooKeeper.States.CONNECTED) {
             LOG.debug("ZooKeeper connection was already closed! Not processing changed event.");
             return;
         }

         try {
             if (zk.exists(znode, false) != null) {
                 zk.getChildren(znode, this);
                 refreshNodes();
             } else {
                 LOG.debug("znode {} doesn't exist -> not processing any changes", znode);
             }
         } catch (Exception e) {
             if (zk.getState() != ZooKeeper.States.CONNECTED) {
                 LOG.debug("Error getting Zookeeper data: " + e); // e.g. session expired, handled by ZooKeeperDiscovery
             } else {
                 LOG.error("Error getting ZooKeeper data.", e);
             }
         }
     }

     public synchronized void close() {
         closed = true;
         for (EndpointDescription endpoint : nodes.values()) {
             endpointListener.endpointRemoved(endpoint, null);
         }
         nodes.clear();
     }

     private synchronized void refreshNodes() {
         if (closed) {
             return;
         }
         LOG.info("Processing change on node: {}", znode);

         Map<String, EndpointDescription> newNodes = new HashMap<String, EndpointDescription>();
         Map<String, EndpointDescription> prevNodes = new HashMap<String, EndpointDescription>(nodes);
         processChildren(znode, newNodes, prevNodes);

         // whatever is left in prevNodes now has been removed from Discovery
         LOG.debug("processChildren done. Nodes that are missing now and need to be removed: {}", prevNodes.values());
         for (EndpointDescription endpoint : prevNodes.values()) {
             endpointListener.endpointRemoved(endpoint, null);
         }
         nodes = newNodes;
     }

     /**
      * Iterates through all child nodes of the given node and tries to find
      * endpoints. If the recursive flag is set it also traverses into the child
      * nodes.
      *
      * @return true if an endpoint was found and if the node therefore needs to
      *         be monitored for changes
      */
     private boolean processChildren(String zn, Map<String, EndpointDescription> newNodes,
             Map<String, EndpointDescription> prevNodes) {
         List<String> children;
         try {
             LOG.debug("Processing the children of {}", zn);
             children = zk.getChildren(zn, false);

             boolean foundANode = false;
             for (String child : children) {
                 String childZNode = zn + '/' + child;
                 EndpointDescription endpoint = getEndpointDescriptionFromNode(childZNode);
                 if (endpoint != null) {
                     EndpointDescription prevEndpoint = prevNodes.get(child);
                     LOG.info("found new node " + zn + "/[" + child + "]   ( []->child )  props: "
                             + endpoint.getProperties().values());
                     newNodes.put(child, endpoint);
                     prevNodes.remove(child);
                     foundANode = true;
                     LOG.debug("Properties: {}", endpoint.getProperties());
                     if (prevEndpoint == null) {
                         // This guy is new
                         endpointListener.endpointAdded(endpoint, null);
                     } else if (!prevEndpoint.getProperties().equals(endpoint.getProperties())) {
                         // TODO
                     }
                 }
                 if (recursive && processChildren(childZNode, newNodes, prevNodes)) {
                     zk.getChildren(childZNode, this);
                 }
             }

             return foundANode;
         } catch (KeeperException e) {
             LOG.error("Problem processing ZooKeeper node", e);
         } catch (InterruptedException e) {
             LOG.error("Problem processing ZooKeeper node", e);
         }
         return false;
     }

     /**
      * Retrieves data from the given node and parses it into an EndpointDescription.
      *
      * @param node a node path
      * @return endpoint found in the node or null if no endpoint was found
      */
     private EndpointDescription getEndpointDescriptionFromNode(String node) {
         try {
             Stat stat = zk.exists(node, false);
             if (stat == null || stat.getDataLength() <= 0) {
                 return null;
             }
             byte[] data = zk.getData(node, false, null);
             LOG.debug("Got data for node: {}", node);

             EndpointDescription endpoint = getFirstEnpointDescription(data);
             if (endpoint != null) {
                 return endpoint;
             }
             LOG.warn("No Discovery information found for node: {}", node);
         } catch (Exception e) {
             LOG.error("Problem getting EndpointDescription from node " + node, e);
         }
         return null;
     }

     public EndpointDescription getFirstEnpointDescription(byte[] data) {
         List<EndpointDescriptionType> elements = parser.getEndpointDescriptions(new ByteArrayInputStream(data));
         if (elements.isEmpty()) {
             return null;
         }
         Map<String, Object> props = new PropertiesMapper().toProps(elements.get(0).getProperty());
         return new EndpointDescription(props);
     }
 }
	/**
	* 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.cxf.dosgi.discovery.zookeeper.subscribe;

	import java.io.ByteArrayInputStream;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;

	import org.apache.cxf.dosgi.discovery.zookeeper.util.Utils;
	import org.apache.cxf.dosgi.endpointdesc.EndpointDescriptionParser;
	import org.apache.cxf.dosgi.endpointdesc.PropertiesMapper;
	import org.apache.zookeeper.AsyncCallback.StatCallback;
	import org.apache.zookeeper.KeeperException;
	import org.apache.zookeeper.KeeperException.Code;
	import org.apache.zookeeper.WatchedEvent;
	import org.apache.zookeeper.Watcher;
	import org.apache.zookeeper.ZooKeeper;
	import org.apache.zookeeper.data.Stat;
	import org.osgi.service.remoteserviceadmin.EndpointDescription;
	import org.osgi.service.remoteserviceadmin.EndpointListener;
	import org.osgi.xmlns.rsa.v1_0.EndpointDescriptionType;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Monitors ZooKeeper for changes in published endpoints.
	* <p>
	* Specifically, it monitors the node path associated with a given interface class,
	* whose data is a serialized version of an EndpointDescription, and notifies an
	* EndpointListener when changes are detected (which can then propagate the
	* notification to other EndpointListeners with a matching scope).
	* <p>
	* Note that the EndpointListener is used here as a decoupling interface for
	* convenience, and is not necessarily used according to its documented contract.
	*/
	public class InterfaceMonitor implements Watcher, StatCallback {

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

	private final String znode;
	private final ZooKeeper zk;
	private final EndpointListener endpointListener;
	private final boolean recursive;
	private volatile boolean closed;

	// This map reference changes, so don't synchronize on it
	private Map<String, EndpointDescription> nodes = new HashMap<String, EndpointDescription>();

	private EndpointDescriptionParser parser;

	public InterfaceMonitor(ZooKeeper zk, String objClass, EndpointListener endpointListener, String scope) {
	this.zk = zk;
	this.znode = Utils.getZooKeeperPath(objClass);
	this.recursive = objClass == null \|\| objClass.isEmpty();
	this.endpointListener = endpointListener;
	this.parser = new EndpointDescriptionParser();
	LOG.debug("Creating new InterfaceMonitor {} for scope [{}] and objectClass [{}]",
	new Object[] {recursive ? "(recursive)" : "", scope, objClass});
	}

	/**
	* Returns all endpoints that are currently known to this monitor.
	*
	* @return all endpoints that are currently known to this monitor
	*/
	public synchronized List<EndpointDescription> getEndpoints() {
	return new ArrayList<EndpointDescription>(nodes.values());
	}

	public void start() {
	watch();
	}

	private void watch() {
	LOG.debug("registering a ZooKeeper.exists({}) callback", znode);
	zk.exists(znode, this, this, null);
	}

	/**
	* Zookeeper Watcher interface callback.
	*/
	public void process(WatchedEvent event) {
	LOG.debug("ZooKeeper watcher callback on node {} for event {}", znode, event);
	processDelta();
	}

	/**
	* Zookeeper StatCallback interface callback.
	*/
	@SuppressWarnings("deprecation")
	public void processResult(int rc, String path, Object ctx, Stat stat) {
	LOG.debug("ZooKeeper callback on node: {} code: {}", znode, rc);

	switch (rc) {
	case Code.Ok:
	case Code.NoNode:
	processDelta();
	return;

	case Code.SessionExpired:
	case Code.NoAuth:
	case Code.ConnectionLoss:
	return;

	default:
	watch();
	}
	}

	private void processDelta() {
	if (closed) {
	return;
	}

	if (zk.getState() != ZooKeeper.States.CONNECTED) {
	LOG.debug("ZooKeeper connection was already closed! Not processing changed event.");
	return;
	}

	try {
	if (zk.exists(znode, false) != null) {
	zk.getChildren(znode, this);
	refreshNodes();
	} else {
	LOG.debug("znode {} doesn't exist -> not processing any changes", znode);
	}
	} catch (Exception e) {
	if (zk.getState() != ZooKeeper.States.CONNECTED) {
	LOG.debug("Error getting Zookeeper data: " + e); // e.g. session expired, handled by ZooKeeperDiscovery
	} else {
	LOG.error("Error getting ZooKeeper data.", e);
	}
	}
	}

	public synchronized void close() {
	closed = true;
	for (EndpointDescription endpoint : nodes.values()) {
	endpointListener.endpointRemoved(endpoint, null);
	}
	nodes.clear();
	}

	private synchronized void refreshNodes() {
	if (closed) {
	return;
	}
	LOG.info("Processing change on node: {}", znode);

	Map<String, EndpointDescription> newNodes = new HashMap<String, EndpointDescription>();
	Map<String, EndpointDescription> prevNodes = new HashMap<String, EndpointDescription>(nodes);
	processChildren(znode, newNodes, prevNodes);

	// whatever is left in prevNodes now has been removed from Discovery
	LOG.debug("processChildren done. Nodes that are missing now and need to be removed: {}", prevNodes.values());
	for (EndpointDescription endpoint : prevNodes.values()) {
	endpointListener.endpointRemoved(endpoint, null);
	}
	nodes = newNodes;
	}

	/**
	* Iterates through all child nodes of the given node and tries to find
	* endpoints. If the recursive flag is set it also traverses into the child
	* nodes.
	*
	* @return true if an endpoint was found and if the node therefore needs to
	* be monitored for changes
	*/
	private boolean processChildren(String zn, Map<String, EndpointDescription> newNodes,
	Map<String, EndpointDescription> prevNodes) {
	List<String> children;
	try {
	LOG.debug("Processing the children of {}", zn);
	children = zk.getChildren(zn, false);

	boolean foundANode = false;
	for (String child : children) {
	String childZNode = zn + '/' + child;
	EndpointDescription endpoint = getEndpointDescriptionFromNode(childZNode);
	if (endpoint != null) {
	EndpointDescription prevEndpoint = prevNodes.get(child);
	LOG.info("found new node " + zn + "/[" + child + "] ( []->child ) props: "
	+ endpoint.getProperties().values());
	newNodes.put(child, endpoint);
	prevNodes.remove(child);
	foundANode = true;
	LOG.debug("Properties: {}", endpoint.getProperties());
	if (prevEndpoint == null) {
	// This guy is new
	endpointListener.endpointAdded(endpoint, null);
	} else if (!prevEndpoint.getProperties().equals(endpoint.getProperties())) {
	// TODO
	}
	}
	if (recursive && processChildren(childZNode, newNodes, prevNodes)) {
	zk.getChildren(childZNode, this);
	}
	}

	return foundANode;
	} catch (KeeperException e) {
	LOG.error("Problem processing ZooKeeper node", e);
	} catch (InterruptedException e) {
	LOG.error("Problem processing ZooKeeper node", e);
	}
	return false;
	}

	/**
	* Retrieves data from the given node and parses it into an EndpointDescription.
	*
	* @param node a node path
	* @return endpoint found in the node or null if no endpoint was found
	*/
	private EndpointDescription getEndpointDescriptionFromNode(String node) {
	try {
	Stat stat = zk.exists(node, false);
	if (stat == null \|\| stat.getDataLength() <= 0) {
	return null;
	}
	byte[] data = zk.getData(node, false, null);
	LOG.debug("Got data for node: {}", node);

	EndpointDescription endpoint = getFirstEnpointDescription(data);
	if (endpoint != null) {
	return endpoint;
	}
	LOG.warn("No Discovery information found for node: {}", node);
	} catch (Exception e) {
	LOG.error("Problem getting EndpointDescription from node " + node, e);
	}
	return null;
	}

	public EndpointDescription getFirstEnpointDescription(byte[] data) {
	List<EndpointDescriptionType> elements = parser.getEndpointDescriptions(new ByteArrayInputStream(data));
	if (elements.isEmpty()) {
	return null;
	}
	Map<String, Object> props = new PropertiesMapper().toProps(elements.get(0).getProperty());
	return new EndpointDescription(props);
	}
	}