distributedqueue.java

来自「JGRoups源码」· Java 代码 · 共 733 行 · 第 1/2 页

                }

                if (retval == null)
                {
                    try
                    {
                        synchronized (mutex)
                        {
                            mutex.wait();
                        }
                    }
                     catch (InterruptedException e)
                    {
                    }
                }
            }
        }
        else
        {
            while (((System.currentTimeMillis() - start) < timeout) && !stopped && (retval == null))
            {
                RspList rsp = disp.callRemoteMethods(null, "_remove", null, remove_signature, GroupRequest.GET_ALL, internal_timeout);
                Vector results = rsp.getResults();

                if (results.size() > 0)
                {
                    retval = results.elementAt(0);

                    if (logger.isDebugEnabled())
                    {
                        checkResult(rsp, retval);
                    }
                }

                if (retval == null)
                {
                    try
                    {
                        long delay = timeout - (System.currentTimeMillis() - start);

                        synchronized (mutex)
                        {
                            if (delay > 0)
                            {
                                mutex.wait(delay);
                            }
                        }
                    }
                     catch (InterruptedException e)
                    {
                    }
                }
            }
        }

        return retval;
    }

    public String toString()
    {
        return internalQueue.toString();
    }

    /*------------------------ Callbacks -----------------------*/
    public void _add(Object value)
    {
        if (logger.isDebugEnabled())
        {
            logger.debug(groupname + '@' + getLocalAddress() + " _add(" + value + ')');
        }

        /*lock the queue from other threads*/
        synchronized (mutex)
        {
            internalQueue.add(value);

            /*wake up all the threads that are waiting for the lock to be released*/
            mutex.notifyAll();
        }

        for (int i = 0; i < notifs.size(); i++)
            ((Notification)notifs.elementAt(i)).entryAdd(value);
    }

    public void _addAtHead(Object value)
    {
        /*lock the queue from other threads*/
        synchronized (mutex)
        {
            internalQueue.addFirst(value);

            /*wake up all the threads that are waiting for the lock to be released*/
            mutex.notifyAll();
        }

        for (int i = 0; i < notifs.size(); i++)
            ((Notification)notifs.elementAt(i)).entryAdd(value);
    }

    public void _reset()
    {
        if (logger.isDebugEnabled())
        {
            logger.debug(groupname + '@' + getLocalAddress() + " _reset()");
        }

        _private_reset();

        for (int i = 0; i < notifs.size(); i++)
            ((Notification)notifs.elementAt(i)).contentsCleared();
    }

    protected void _private_reset()
    {
        /*lock the queue from other threads*/
        synchronized (mutex)
        {
            internalQueue.clear();

            /*wake up all the threads that are waiting for the lock to be released*/
            mutex.notifyAll();
        }
    }

    public Object _remove()
    {
        Object retval = null;

        try
        {
            /*lock the queue from other threads*/
            synchronized (mutex)
            {
                retval = internalQueue.removeFirst();

                /*wake up all the threads that are waiting for the lock to be released*/
                mutex.notifyAll();
            }

            if (logger.isDebugEnabled())
            {
                logger.debug(groupname + '@' + getLocalAddress() + "_remove(" + retval + ')');
            }

            for (int i = 0; i < notifs.size(); i++)
                ((Notification)notifs.elementAt(i)).entryRemoved(retval);
        }
         catch (NoSuchElementException e)
        {
            logger.debug(groupname + '@' + getLocalAddress() + "_remove(): nothing to remove");
        }

        return retval;
    }

    public void _addAll(Collection c)
    {
        if (logger.isDebugEnabled())
        {
            logger.debug(groupname + '@' + getLocalAddress() + " _addAll(" + c + ')');
        }

        /*lock the queue from other threads*/
        synchronized (mutex)
        {
            internalQueue.addAll(c);

            /*wake up all the threads that are waiting for the lock to be released*/
            mutex.notifyAll();
        }

        for (int i = 0; i < notifs.size(); i++)
            ((Notification)notifs.elementAt(i)).contentsSet(c);
    }

    /*----------------------------------------------------------*/
    /*-------------------- State Exchange ----------------------*/
    public void receive(Message msg)
    {
    }

    public byte[] getState()
    {
        Vector copy = (Vector)getContents().clone();

        try
        {
            return Util.objectToByteBuffer(copy);
        }
         catch (Throwable ex)
        {
            logger.error("DistributedQueue.getState(): exception marshalling state.", ex);

            return null;
        }
    }

    public void setState(byte[] new_state)
    {
        Vector new_copy;

        try
        {
            new_copy = (Vector)Util.objectFromByteBuffer(new_state);

            if (new_copy == null)
            {
                return;
            }
        }
         catch (Throwable ex)
        {
            logger.error("DistributedQueue.setState(): exception unmarshalling state.", ex);

            return;
        }

        _private_reset(); // remove all elements      
        _addAll(new_copy);
    }

    /*------------------- Membership Changes ----------------------*/
    public void viewAccepted(View new_view)
    {
        Vector new_mbrs = new_view.getMembers();

        if (new_mbrs != null)
        {
            sendViewChangeNotifications(new_mbrs, members); // notifies observers (joined, left)
            members.removeAllElements();

            for (int i = 0; i < new_mbrs.size(); i++)
                members.addElement(new_mbrs.elementAt(i));
        }
    }

    /** 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()
    {
    }

    void sendViewChangeNotifications(Vector new_mbrs, Vector old_mbrs)
    {
        Vector joined;
        Vector left;
        Object mbr;
        Notification n;

        if ((notifs.size() == 0) || (old_mbrs == null) || (new_mbrs == null) || (old_mbrs.size() == 0) ||
                (new_mbrs.size() == 0))
        {
            return;
        }

        // 1. Compute set of members that joined: all that are in new_mbrs, but not in old_mbrs
        joined = new Vector();

        for (int i = 0; i < new_mbrs.size(); i++)
        {
            mbr = new_mbrs.elementAt(i);

            if (!old_mbrs.contains(mbr))
            {
                joined.addElement(mbr);
            }
        }

        // 2. Compute set of members that left: all that were in old_mbrs, but not in new_mbrs
        left = new Vector();

        for (int i = 0; i < old_mbrs.size(); i++)
        {
            mbr = old_mbrs.elementAt(i);

            if (!new_mbrs.contains(mbr))
            {
                left.addElement(mbr);
            }
        }

        for (int i = 0; i < notifs.size(); i++)
        {
            n = (Notification)notifs.elementAt(i);
            n.viewChange(joined, left);
        }
    }

    final void initSignatures()
    {
        try
        {
            if (add_signature == null)
            {
                add_signature = new Class[] { Object.class };
            }

            if (addAtHead_signature == null)
            {
                addAtHead_signature = new Class[] { Object.class };
            }

            if (addAll_signature == null)
            {
                addAll_signature = new Class[] { Collection.class };
            }

            if (reset_signature == null)
            {
                reset_signature = new Class[0];
            }

            if (remove_signature == null)
            {
                remove_signature = new Class[0];
            }
        }
         catch (Throwable ex)
        {
            logger.error("DistributedQueue.initMethods()", ex);
        }
    }

    public static void main(String[] args)
    {
        try
        {
            // The setup here is kind of weird:
            // 1. Create a channel
            // 2. Create a DistributedQueue (on the channel)
            // 3. Connect the channel (so the HT gets a VIEW_CHANGE)
            // 4. Start the HT
            //
            // A simpler setup is
            // DistributedQueue ht = new DistributedQueue("demo", null, 
            //         "file://c:/JGroups-2.0/conf/total-token.xml", 5000);
            JChannel c = new JChannel("file:/c:/JGroups-2.0/conf/conf/total-token.xml");

            DistributedQueue ht = new DistributedQueue(c);
            c.connect("demo");
            ht.start(5000);

            ht.add("name");
            ht.add("Michelle Ban");

            Object old_key = ht.remove();
            System.out.println("old key was " + old_key);
            old_key = ht.remove();
            System.out.println("old value was " + old_key);

            ht.add("name 'Michelle Ban'");

            System.out.println("queue is " + ht);
        }
         catch (Throwable t)
        {
            t.printStackTrace();
        }
    }
}