flushtest.java

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

package org.jgroups.tests;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;

import org.jgroups.Address;
import org.jgroups.BlockEvent;
import org.jgroups.Channel;
import org.jgroups.ChannelException;
import org.jgroups.Event;
import org.jgroups.ExtendedReceiverAdapter;
import org.jgroups.GetStateEvent;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.SetStateEvent;
import org.jgroups.UnblockEvent;
import org.jgroups.View;
import org.jgroups.stack.Protocol;
import org.jgroups.util.Util;


/**
 * Tests the FLUSH protocol, requires flush-udp.xml in ./conf to be present and configured to use FLUSH
 * @author Bela Ban
 * @version $Id: FlushTest.java,v 1.16 2006/10/13 23:21:00 vlada Exp $
 */
public class FlushTest extends TestCase {
    Channel c1, c2,c3;
    static final String CONFIG="flush-udp.xml";


    public FlushTest(String name) {
        super(name);
    }
    
    public void setUp() throws Exception {
       super.setUp();       
   }

    public void tearDown() throws Exception {
        super.tearDown();

        if(c3 != null) {
           c3.close();
           assertFalse(c3.isOpen());
           assertFalse(c3.isConnected());
           c3=null;
        }
        
        if(c2 != null) {
            c2.close();
            assertFalse(c2.isOpen());
            assertFalse(c2.isConnected());
            c2=null;
        }               
        
        if(c1 != null) {
           c1.close();
           assertFalse(c1.isOpen());
           assertFalse(c1.isConnected());
           c1=null;
        }
        
        Util.sleep(1000);       
    }  
   

    public void testSingleChannel() throws ChannelException {
        c1=createChannel();
        MyReceiver receiver=new MyReceiver("c1");
        c1.setReceiver(receiver);
        c1.connect("bla");
        // Util.sleep(1000);
        checkEventSequence(receiver);
    }

    /**
     * Tests issue #1 in http://jira.jboss.com/jira/browse/JGRP-335
     */
    public void testJoinFollowedByUnicast() throws ChannelException {
        c1=createChannel();
        c1.setReceiver(new MySimpleReplier(c1, true));
        c1.connect("bla");

        Address target=c1.getLocalAddress();
        Message unicast_msg=new Message(target);

        c2=createChannel();
        c2.setReceiver(new MySimpleReplier(c2, false));
        c2.connect("bla");

        // now send unicast, this might block as described in the case
        c2.send(unicast_msg);
        // if we don't get here this means we'd time out
    }


    /**
     * Tests issue #2 in http://jira.jboss.com/jira/browse/JGRP-335
     */
    public void testStateTransferFollowedByUnicast() throws ChannelException {
        c1=createChannel();
        c1.setReceiver(new MySimpleReplier(c1, true));
        c1.connect("bla");

        Address target=c1.getLocalAddress();
        Message unicast_msg=new Message(target);

        c2=createChannel();
        c2.setReceiver(new MySimpleReplier(c2, false));
        c2.connect("bla");

        // Util.sleep(100);
        System.out.println("\n** Getting the state **");
        c2.getState(null, 10000);
        // now send unicast, this might block as described in the case
        c2.send(unicast_msg);
    }

    public void testTwoChannelsWithMessages() throws ChannelException {
        twoChannelsTestHelper(true);
    }

    public void testTwoChannelsNoMessages() throws ChannelException {
        twoChannelsTestHelper(false);
    }

    public void testThreeChannelsWithMessages() throws ChannelException {
        threeChannelsTestHelper(true);
    }

    public void testThreeChannelsNoMessages() throws ChannelException {
        threeChannelsTestHelper(false);
    }

    public void testStateTransferWithMessages() throws ChannelException {
        stateTransferTestHelper(true);
    }

    public void testStateTransferNoMessages() throws ChannelException {
        stateTransferTestHelper(false);
    }

    private void twoChannelsTestHelper(boolean sendMessages) throws ChannelException {
        c1=createChannel();
        MyReceiver receiver=new MyReceiver("c1");
        c1.setReceiver(receiver);
        c1.connect("bla");
        if(sendMessages){
            c1.send(new Message());
        }
        // Util.sleep(1000);

        c2=createChannel();
        MyReceiver receiver2=new MyReceiver("c2");
        c2.setReceiver(receiver2);
        c2.connect("bla");
        View view=c2.getView();
        assertEquals(2, view.size());
        // Util.sleep(1000);
        if(sendMessages){
            c1.send(new Message());
            c2.send(new Message());
        }

        checkEventSequence(receiver2);

        c2.close();
        // Util.sleep(500);
        if(sendMessages){
            c1.send(new Message());
        }
        // Util.sleep(1000);

        checkEventSequence(receiver);
    }


    private void threeChannelsTestHelper(boolean sendMessages) throws ChannelException {
        c1=createChannel();
        MyReceiver receiver=new MyReceiver("c1");
        c1.setReceiver(receiver);
        c1.connect("bla");
        if(sendMessages){
            c1.send(new Message());
        }

        // Util.sleep(1000);

        c2=createChannel();
        MyReceiver receiver2=new MyReceiver("c2");
        c2.setReceiver(receiver2);
        c2.connect("bla");
        View view=c2.getView();
        assertEquals(2, view.size());
        // Util.sleep(1000);

        if(sendMessages){
            c1.send(new Message());
            c2.send(new Message());
        }

        c3=createChannel();
        MyReceiver receiver3=new MyReceiver("c3");
        c3.setReceiver(receiver3);
        c3.connect("bla");
        view=c3.getView();
        assertEquals(3, view.size());
        // Util.sleep(1000);
        if(sendMessages){
            c1.send(new Message());
            c2.send(new Message());
            c3.send(new Message());
        }

        //close coordinator
        checkEventSequence(receiver);

        c1.close();
        if(sendMessages){
            c2.send(new Message());
            c2.send(new Message());
        }
        // Util.sleep(1000);

        //close coordinator one more time
        checkEventSequence(receiver2);
        c2.close();
        // Util.sleep(1000);

        checkEventSequence(receiver3);
    }



    private void stateTransferTestHelper(boolean sendMessages) throws ChannelException {
        c1=createChannel();
        MyReceiver receiver=new MyReceiver(c1,"c1");
        c1.setReceiver(receiver);
        c1.connect("bla");
        // Util.sleep(1000);
        if(sendMessages){
            c1.send(new Message());
            c1.send(new Message());
        }

        // Util.sleep(1000);

        c2=createChannel();
        MyReceiver receiver2=new MyReceiver(c2,"c2");
        c2.setReceiver(receiver2);
        c2.connect("bla");
        // Util.sleep(1000);


        c3=createChannel();
        MyReceiver receiver3=new MyReceiver(c3,"c3");
        c3.setReceiver(receiver3);
        c3.connect("bla");
        // Util.sleep(1000);
        if(sendMessages){
            c1.send(new Message());
            c2.send(new Message());
        }

        // Util.sleep(500);
        if(sendMessages){
            c1.send(new Message());
            c2.send(new Message());
            c3.send(new Message());
        }

        checkEventSequence(receiver);
        checkEventSequence(receiver2);
        checkEventSequence(receiver3);