totaltokendemo.java

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

//$Id: TotalTokenDemo.java,v 1.10 2005/05/30 16:14:40 belaban Exp $

package org.jgroups.demos;

import org.jgroups.*;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.ProtocolStack;
import org.jgroups.util.Util;

import javax.swing.*;
import java.awt.*;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.io.Serializable;
import java.util.Iterator;
import java.util.Random;
import java.util.Vector;


/**
 *<p>
 * Demonstration of TOTAL_TOKEN protocol stack implementing total
 * order. TotalTokenDemo could however be used by any other
 * stack configuration which does not neccesarily have to satisfy
 * total order. If using stack configuration other than TOTAL_TOKEN
 * an appropriate xml configuration file should be used. See -help for
 * details.
 * <p>
 * TotalTokenDemo will verify :
 * <p>
 * total ordering of messages - by computing a color to be displayed
 * in a gui window.
 * <p>
 * virtual synchrony - by displaying number of messages transmitted
 * in last view.
 * <p>
 *
 *@author Vladimir Blagojevic [e-mail]
 *@author Ivan Bilenjkij  [e-mail]
 *@version $Revision: 1.10 $
 *
 *@see org.jgroups.protocols.TOTAL_TOKEN
 *
 **/


public class TotalTokenDemo extends JFrame implements Runnable
{
    private JChannel channel;
    //main tabbed pane

    final JTabbedPane tabbedPane;

    private ReceiverThread receiverThread;

    private ColorPanel colorPanel;
    private final ControlPanel control;
    private int mSize = 1024;
    private volatile boolean transmitting = false;
    private final String channelProperties;
    private Dimension preffered;

    public TotalTokenDemo(String props)
    {
        super();
        tabbedPane = new JTabbedPane();
        control = new ControlPanel();
        channelProperties = props;

        try
        {
            channel = new JChannel(channelProperties);
        }
        catch (ChannelException e)
        {
            System.err.println("Could not create channel, exiting ....");
            e.printStackTrace(System.err);
        }

        addPanel("Control", control);
        getContentPane().add(tabbedPane);
        connect();

    }
    public void addPanel(String name, JPanel panel)
    {


        if(tabbedPane.getTabCount() == 0)
        {
            preffered = panel.getPreferredSize();
        }


        panel.setPreferredSize(preffered);
        tabbedPane.add(name,panel);
    }

    public JChannel getChannel()
    {
        return channel;
    }


    public void connect()
    {
        try
        {
            channel.connect("TOTAL_TOKEN_DEMO_GROUP");
        }
        catch (ChannelException e)
        {
            e.printStackTrace();
        }
        receiverThread = new ReceiverThread();
        receiverThread.start();
        Address a = channel.getLocalAddress();
        if(a != null)   setTitle(a.toString());
        else setTitle("Not connected");
        control.connected();
    }
    public void run()
    {
        Random r = new Random();

        while (true)
        {
            Util.sleep(10);
            try
            {
                if (transmitting)
                {
                    channel.send(new Message(null, null, new TotalPayload(r.nextInt(255))));
                }
                else
                {
                    Util.sleep(200);
                }

            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }

    public void disconnect()
    {
        transmitting = false;
        receiverThread.shutDown();
        channel.disconnect();
        control.disconnected();
        setTitle("Not connected");
    }

    private class ReceiverThread extends Thread
    {
        volatile boolean running = true;
        Thread nullifier = null;
        private long startTimeThroughput = System.currentTimeMillis();
        private final long oneSecond = 1000;
        private long throughput = 1;

        public ReceiverThread()
        {
            nullifier = new Thread(new Runnable()
            {
                public void run()
                {
                    //nullifies throughput display
                    while (running)
                    {
                        Util.sleep(2000);
                        if ((System.currentTimeMillis() - startTimeThroughput) > 2000)
                        {
                            control.throughput.setText("0 KB/sec");
                        }
                    }
                }
            });
            nullifier.start();
        }

        public void shutDown()
        {
            running = false;
        }

        private void measureThrougput(long size)
        {
            if ((System.currentTimeMillis() - startTimeThroughput) > oneSecond)
            {
                control.throughput.setText("" + (throughput / 1024) + " KB/sec");
                startTimeThroughput = System.currentTimeMillis();
                throughput = 0;
            }
            else
            {
                throughput += size;
            }
        }


        public void run()
        {
            Object tmp;
            Message msg = null;
            int counter = 0;
            Vector v = new Vector();
            while (running)
            {
                Util.sleep(10);
                try
                {

                    tmp = channel.receive(0);
                    if (tmp == null) continue;

                    if (tmp instanceof View)
                    {
                        View vw = (View) tmp;
                        control.viewNumber.setText("" + vw.getVid().getId());
                        control.numMessagesInLastView.setText("" + counter);
                        counter = 0;
                        v.clear();
                        continue;
                    }

                    if (tmp instanceof ExitEvent)
                    {
                        System.out.println("received EXIT");
                        break;
                    }

                    if (!(tmp instanceof Message))
                        continue;

                    msg = (Message) tmp;

                    measureThrougput(msg.size());
                    TotalPayload p =null;

                    p=(TotalPayload)msg.getObject();
                    v.addElement(new Integer(p.getRandomSequence()));
                    int size = v.size();
                    if (size % 50 == 0)
                    {
                        int value = 0;
                        int i = 0;
                        Iterator iter = v.iterator();
                        while (iter.hasNext())
                        {
                            i++;
                            int seq = ((Integer) iter.next()).intValue();
                            if (i % 2 == 0)
                            {
                                value *= seq;
                            }
                            else if (i % 3 == 0)
                            {
                                value -= seq;
                            }
                            else
                                value += seq;
                        }
                        v.clear();
                        value = Math.abs(value);
                        int r = value % 85;
                        int g = value % 170;
                        int b = value % 255;
                        colorPanel.setSeq(r, g, b);
                    }
                    counter++;
                }
                catch (ChannelNotConnectedException e)
                {
                    e.printStackTrace();
                }
                catch (ChannelClosedException e)
                {
                    e.printStackTrace();
                }