private Map<String, TrackedNode<T>> trackNode(String path,
                                                Map<String, TrackedNode<T>> tree,
                                                Collection<NodeEvent<T>> events,
                                                int depth)
    throws InterruptedException, KeeperException
  {
    if(depth > _depth)
    {
      if(log.isDebugEnabled())
        log.debug(logString(path, "max depth reached ${depth}"));

      return tree;
    }

    TrackedNode<T> oldTrackedNode = tree.get(path);

    try
    {
      ZKData<T> res = _zkDataReader.readData(_zk, path, _treeWacher);

      TrackedNode<T> newTrackedNode =
        new TrackedNode<T>(path, res.getData(), res.getStat(), depth);
      if(oldTrackedNode != null)
      {
        if(!_zkDataReader.isEqual(oldTrackedNode.getData(), newTrackedNode.getData()))
        {
          events.add(new NodeEvent<T>(NodeEventType.UPDATED,
                                      newTrackedNode));
        }
      }
      else
      {
        events.add(new NodeEvent<T>(NodeEventType.ADDED,
                                    newTrackedNode));
      }

      tree.put(path, newTrackedNode);

      if(depth < _depth)
      {
        ZKChildren children = _zk.getZKChildren(path, _treeWacher);
        // the stat may change between the 2 calls
        if(!newTrackedNode.getStat().equals(children.getStat()))
          newTrackedNode.setStat(children.getStat());
        Collections.sort(children.getChildren());
        for(String child : children.getChildren())
        {
          String childPath = PathUtils.addPaths(path, child);
          if(!tree.containsKey(childPath))
            trackNode(childPath, tree, events, depth + 1);
        }
      }

      _lastZkTxId = Math.max(_lastZkTxId, newTrackedNode.getZkTxId());
      _lock.notifyAll();
      if(log.isDebugEnabled())
        log.debug(logString(path,
                            "start tracking " + (depth < _depth ? "": "leaf ") +
                            "node zkTxId=" + newTrackedNode.getZkTxId()));

    }
    catch (KeeperException.NoNodeException e)
    {
      // this is a race condition which could happen between the moment the event is received
      // and this call
      if(log.isDebugEnabled())
        log.debug(logString(path, "no such node"));
      // it means that the node has disappeared

      tree.remove(path);

      if(oldTrackedNode != null)
      {
        events.add(new NodeEvent<T>(NodeEventType.DELETED, 
                                    oldTrackedNode));
      }
    }

    return tree;
  }
private Map < String , TrackedNode < T > > trackNode ( String path , Map < String , TrackedNode < T > > tree , Collection < NodeEvent < T > > events , int depth ) throws InterruptedException , KeeperException { if ( depth > _depth ) { if ( log . isDebugEnabled ( ) ) log . debug ( logString ( path , "max depth reached ${depth}" ) ) ; return tree ; } TrackedNode < T > oldTrackedNode = tree . get ( path ) ; try { ZKData < T > res = _zkDataReader . readData ( _zk , path , _treeWacher ) ; TrackedNode < T > newTrackedNode = new TrackedNode < T > ( path , res . getData ( ) , res . getStat ( ) , depth ) ; if ( oldTrackedNode != null ) { if ( ! _zkDataReader . isEqual ( oldTrackedNode . getData ( ) , newTrackedNode . getData ( ) ) ) { events . add ( new NodeEvent < T > ( NodeEventType . UPDATED , newTrackedNode ) ) ; } } else { events . add ( new NodeEvent < T > ( NodeEventType . ADDED , newTrackedNode ) ) ; } tree . put ( path , newTrackedNode ) ; if ( depth < _depth ) { ZKChildren children = _zk . getZKChildren ( path , _treeWacher ) ; if ( ! newTrackedNode . getStat ( ) . equals ( children . getStat ( ) ) ) newTrackedNode . setStat ( children . getStat ( ) ) ; Collections . sort ( children . getChildren ( ) ) ; for ( String child : children . getChildren ( ) ) { String childPath = PathUtils . addPaths ( path , child ) ; if ( ! tree . containsKey ( childPath ) ) trackNode ( childPath , tree , events , depth + 1 ) ; } } _lastZkTxId = Math . max ( _lastZkTxId , newTrackedNode . getZkTxId ( ) ) ; _lock . notifyAll ( ) ; if ( log . isDebugEnabled ( ) ) log . debug ( logString ( path , "start tracking " + ( depth < _depth ? "" : "leaf " ) + "node zkTxId=" + newTrackedNode . getZkTxId ( ) ) ) ; } catch ( KeeperException . NoNodeException e ) { if ( log . isDebugEnabled ( ) ) log . debug ( logString ( path , "no such node" ) ) ; tree . remove ( path ) ; if ( oldTrackedNode != null ) { events . add ( new NodeEvent < T > ( NodeEventType . DELETED , oldTrackedNode ) ) ; } } return tree ; }
Must be called from a synchronized section