replicatedtree.java

来自「JGRoups源码」· Java 代码 · 共 1,098 行 · 第 1/3 页

        if(n == null) return null;
        data=n.getData();
        if(data == null) return null;
        return data.keySet();
    }


    /**
     * Finds a node given its name and returns the value associated with a given key in its <code>data</code>
     * map. Returns null if the node was not found in the tree or the key was not found in the hashmap.
     * @param fqn The fully qualified name of the node.
     * @param key The key.
     */
    public Object get(String fqn, String key) {
        Node n=findNode(fqn);

        if(n == null) return null;
        return n.getData(key);
    }


    /**
     * Returns the data hashmap for a given node. This method can only be used by callers that are inside
     * the same package. The reason is that callers must not modify the return value, as these modifications
     * would not be replicated, thus rendering the replicas inconsistent.
     * @param fqn The fully qualified name of the node
     * @return HashMap The data hashmap for the given node
     */
    HashMap get(String fqn) {
        Node n=findNode(fqn);

        if(n == null) return null;
        return n.getData();
    }


    /**
     * Prints a representation of the node defined by <code>fqn</code>. Output includes name, fqn and
     * data.
     */
    public String print(String fqn) {
        Node n=findNode(fqn);
        if(n == null) return null;
        return n.toString();
    }


    /**
     * Returns all children of a given node
     * @param fqn The fully qualified name of the node
     * @return Set A list of child names (as Strings)
     */
    public Set getChildrenNames(String fqn) {
        Node n=findNode(fqn);
        Map m;

        if(n == null) return null;
        m=n.getChildren();
        if(m != null)
            return m.keySet();
        else
            return null;
    }


    public String toString() {
        StringBuffer sb=new StringBuffer();
        int indent=0;
        Map children;

        children=root.getChildren();
        if(children != null && children.size() > 0) {
            Collection nodes=children.values();
            for(Iterator it=nodes.iterator(); it.hasNext();) {
                ((Node)it.next()).print(sb, indent);
                sb.append('\n');
            }
        }
        else
            sb.append(SEPARATOR);
        return sb.toString();
    }

   /**
    * Returns the name of the group that the DistributedTree is connected to
    * @return String
    */
    public String  getGroupName()           {return groupname;}
	 	
   /**
    * Returns the Channel the DistributedTree is connected to 
    * @return Channel
    */
    public Channel getChannel()             {return channel;}

   /**
    * Returns the number of current members joined to the group
    * @return int
    */
    public int getGroupMembersNumber()			{return members.size();}




    /* --------------------- Callbacks -------------------------- */


    public void _put(String fqn, HashMap data) {
        Node n;
        StringHolder child_name=new StringHolder();
        boolean child_exists=false;

        if(fqn == null) return;
        n=findParentNode(fqn, child_name, true); // create all nodes if they don't exist
        if(child_name.getValue() != null) {
            child_exists=n.childExists(child_name.getValue());
            n.createChild(child_name.getValue(), fqn, n, data);
        }
        else {
            child_exists=true;
            n.setData(data);
        }
        if(child_exists)
            notifyNodeModified(fqn);
        else
            notifyNodeAdded(fqn);
    }


    public void _put(String fqn, String key, Object value) {
        Node n;
        StringHolder child_name=new StringHolder();
        boolean child_exists=false;

        if(fqn == null || key == null || value == null) return;
        n=findParentNode(fqn, child_name, true);
        if(child_name.getValue() != null) {
            child_exists=n.childExists(child_name.getValue());
            n.createChild(child_name.getValue(), fqn, n, key, value);
        }
        else {
            child_exists=true;
            n.setData(key, value);
        }
        if(child_exists)
            notifyNodeModified(fqn);
        else
            notifyNodeAdded(fqn);
    }


    public void _remove(String fqn) {
        Node n;
        StringHolder child_name=new StringHolder();

        if(fqn == null) return;
        if(fqn.equals(SEPARATOR)) {
            root.removeAll();
            notifyNodeRemoved(fqn);
            return;
        }
        n=findParentNode(fqn, child_name, false);
        if(n == null) return;
        n.removeChild(child_name.getValue(), fqn);
        notifyNodeRemoved(fqn);
    }


    public void _remove(String fqn, String key) {
        Node n;

        if(fqn == null || key == null) return;
        n=findNode(fqn);
        if(n != null)
            n.removeData(key);
    }


    public void _removeData(String fqn) {
        Node n;

        if(fqn == null) return;
        n=findNode(fqn);
        if(n != null)
            n.removeData();
    }


    /* ----------------- End of  Callbacks ---------------------- */






    /*-------------------- MessageListener ----------------------*/

    /** Callback. Process the contents of the message; typically an _add() or _set() request */
    public void receive(Message msg) {
        Request req=null;

        if(msg == null || msg.getLength() == 0)
            return;
        try {
            req=(Request)msg.getObject();
            request_queue.add(req);
        }
        catch(QueueClosedException queue_closed_ex) {
            if(log.isErrorEnabled()) log.error("request queue is null");
        }
        catch(Exception ex) {
            if(log.isErrorEnabled()) log.error("failed unmarshalling request: " + ex);
        }
    }

    /** Return a copy of the current cache (tree) */
    public byte[] getState() {
        try {
            return Util.objectToByteBuffer(root.clone());
        }
        catch(Throwable ex) {
            if(log.isErrorEnabled()) log.error("exception returning cache: " + ex);
            return null;
        }
    }

    /** Set the cache (tree) to this value */
    public void setState(byte[] new_state) {
        Node new_root=null;
        Object obj;

        if(new_state == null) {
            if(log.isInfoEnabled()) log.info("new cache is null");
            return;
        }
        try {
            obj=Util.objectFromByteBuffer(new_state);
            new_root=(Node)((Node)obj).clone();
            root=new_root;
            notifyAllNodesCreated(root);
        }
        catch(Throwable ex) {
            if(log.isErrorEnabled()) log.error("could not set cache: " + ex);
        }
    }

    /*-------------------- End of MessageListener ----------------------*/





    /*----------------------- MembershipListener ------------------------*/

    public void viewAccepted(View new_view) {
        Vector new_mbrs=new_view.getMembers();

        // todo: if MergeView, fetch and reconcile state from coordinator
        // actually maybe this is best left up to the application ? we just notify them and let
        // the appl handle it ?

        if(new_mbrs != null) {
            notifyViewChange(new_view);
            members.removeAllElements();
            for(int i=0; i < new_mbrs.size(); i++)
                members.addElement(new_mbrs.elementAt(i));
        }
		//if size is bigger than one, there are more peers in the group
		//otherwise there is only one server.
        send_message=members.size() > 1;
    }


    /** Called when a member is suspected */
    public void suspect(Address suspected_mbr) {
        ;
    }


    /** Block sending and receiving of messages until viewAccepted() is called */
    public void block() {
    }

    /*------------------- End of MembershipListener ----------------------*/



    /** Request handler thread */
    public void run() {
        Request req;
        String fqn=null;

        while(request_handler != null) {
            try {
                req=(Request)request_queue.remove(0);
                fqn=req.fqn;
                switch(req.type) {
                    case Request.PUT:
                        if(req.key != null && req.value != null)
                            _put(fqn, req.key, req.value);
                        else
                            _put(fqn, req.data);
                        break;
                    case Request.REMOVE:
                        if(req.key != null)
                            _remove(fqn, req.key);
                        else
                            _remove(fqn);
                        break;
                    default:
                        if(log.isErrorEnabled()) log.error("type " + req.type + " unknown");
                        break;
                }
            }
            catch(QueueClosedException queue_closed_ex) {
                request_handler=null;
                break;
            }
            catch(Throwable other_ex) {
                if(log.isWarnEnabled()) log.warn("exception processing request: " + other_ex);
            }
        }
    }


    /**
     * Find the node just <em>above</em> the one indicated by <code>fqn</code>. This is needed in many cases,
     * e.g. to add a new node or remove an existing node.
     * @param fqn The fully qualified name of the node.
     * @param child_name Will be filled with the name of the child when this method returns. The child name
     *                   is the last relative name of the <code>fqn</code>, e.g. in "/a/b/c" it would be "c".
     * @param create_if_not_exists Create parent nodes along the way if they don't exist. Otherwise, this method
     *                             will return when a node cannot be found.
     */
    Node findParentNode(String fqn, StringHolder child_name, boolean create_if_not_exists) {
        Node curr=root, node;
        StringTokenizer tok;
        String name;
        StringBuffer sb=null;

        if(fqn == null || fqn.equals(SEPARATOR) || "".equals(fqn))
            return curr;

        sb=new StringBuffer();
        tok=new StringTokenizer(fqn, SEPARATOR);
        while(tok.countTokens() > 1) {
            name=tok.nextToken();
            sb.append(SEPARATOR).append(name);
            node=curr.getChild(name);
            if(node == null && create_if_not_exists)
                node=curr.createChild(name, sb.toString(), null, null);
            if(node == null)
                return null;
            else
                curr=node;
        }

        if(tok.countTokens() > 0 && child_name != null)
            child_name.setValue(tok.nextToken());
        return curr;
    }


    /**
     * Returns the node at fqn. This method should not be used by clients (therefore it is package-private):
     * it is only used internally (for navigation). C++ 'friend' would come in handy here...