multiplexertest.java

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

        c1_repl=new Cache(ch1_repl, "cache-1-repl");
        boolean rc=ch1_repl.getState(null, 5000);
        System.out.println("state transfer: " + rc);

        ch2_repl.connect("bla");
        c2_repl=new Cache(ch2_repl, "cache-2-repl");
        rc=ch2_repl.getState(null, 5000);
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("Caches after state transfers:");
        System.out.println("c1: " + c1);
        System.out.println("c1_repl: " + c1_repl);
        System.out.println("c2: " + c2);
        System.out.println("c2_repl: " + c2_repl);

        assertEquals(1, c1.size());
        assertEquals(1, c1_repl.size());

        assertEquals(1, c2.size());
        assertEquals(1, c2_repl.size());

        assertEquals("cache-1", c1.get("name"));
        assertEquals("cache-1", c1_repl.get("name"));

        assertEquals("cache-2", c2.get("name"));
        assertEquals("cache-2", c2_repl.get("name"));
    }


    public void testStateTransferWithRegistration() throws Exception {
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        ch1.connect("bla");
        c1=new Cache(ch1, "cache-1");
        assertEquals("cache has to be empty initially", 0, c1.size());

        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        ch2.connect("bla");
        c2=new Cache(ch2, "cache-2");
        assertEquals("cache has to be empty initially", 0, c2.size());
        c1.put("name", "cache-1");
        c2.put("name", "cache-2");

        ch1_repl=factory2.createMultiplexerChannel(STACK_NAME, "c1", true, null); // register for state transfer
        ch2_repl=factory2.createMultiplexerChannel(STACK_NAME, "c2", true, null); // register for state transfer

        ch1_repl.connect("bla");
        c1_repl=new Cache(ch1_repl, "cache-1-repl");
        boolean rc=ch1_repl.getState(null, 5000); // this will *not* trigger the state transfer protocol
        System.out.println("state transfer: " + rc);

        ch2_repl.connect("bla");
        c2_repl=new Cache(ch2_repl, "cache-2-repl");
        rc=ch2_repl.getState(null, 5000); // only *this* will trigger the state transfer
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("Caches after state transfers:");
        System.out.println("c1: " + c1);
        System.out.println("c1_repl: " + c1_repl);
        System.out.println("c2: " + c2);
        System.out.println("c2_repl: " + c2_repl);

        assertEquals(1, c1.size());
        assertEquals(1, c1_repl.size());

        assertEquals(1, c2.size());
        assertEquals(1, c2_repl.size());

        assertEquals("cache-1", c1.get("name"));
        assertEquals("cache-1", c1_repl.get("name"));

        assertEquals("cache-2", c2.get("name"));
        assertEquals("cache-2", c2_repl.get("name"));
        c1.clear();
        c1_repl.clear();
        c2.clear();
        c2_repl.clear();
    }


    private void setCorrectPortRange(Channel ch) {
        ProtocolStack stack=((MuxChannel)ch).getProtocolStack();
        Protocol tcpping=stack.findProtocol("TCPPING");
        if(tcpping == null)
            return;

        Properties props=tcpping.getProperties();
        String port_range=props.getProperty("port_range");
        if(port_range != null) {
            System.out.println("port_range in TCPPING: " + port_range + ", setting it to 2");
            port_range="2";
            Properties p=new Properties();
            // p.putAll(props);
            p.setProperty("port_range", port_range);
            tcpping.setProperties(p);
        }
    }


    public void testStateTransferWithReconnect() throws Exception {
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        setCorrectPortRange(ch1);

        assertTrue(ch1.isOpen());
        assertFalse(ch1.isConnected());
        ch1.connect("bla");
        assertTrue(ch1.isOpen());
        assertTrue(ch1.isConnected());
        assertServiceAndClusterView(ch1, 1, 1);

        c1=new Cache(ch1, "cache-1");
        assertEquals("cache has to be empty initially", 0, c1.size());

        ch1_repl=factory2.createMultiplexerChannel(STACK_NAME, "c1");
        setCorrectPortRange(ch1_repl);
        assertTrue(ch1_repl.isOpen());
        assertFalse(ch1_repl.isConnected());

        c1.put("name", "Bela");
        c1.put("id", new Long(322649));
        c1.put("hobbies", "biking");
        c1.put("bike", "Centurion");

        ch1_repl.connect("bla");
        assertTrue(ch1_repl.isOpen());
        assertTrue(ch1_repl.isConnected());
        assertServiceAndClusterView(ch1_repl, 2, 2);
        Util.sleep(500);
        assertServiceAndClusterView(ch1, 2, 2);

        c1_repl=new Cache(ch1_repl, "cache-1-repl");
        boolean rc=ch1_repl.getState(null, 5000);
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("c1_repl: " + c1_repl);
        assertEquals("initial state should have been transferred", 4, c1_repl.size());
        assertEquals(new Long(322649), c1.get("id"));
        assertEquals(new Long(322649), c1_repl.get("id"));

        assertEquals("biking", c1.get("hobbies"));
        assertEquals("biking", c1_repl.get("hobbies"));

        assertEquals("Centurion", c1.get("bike"));
        assertEquals("Centurion", c1_repl.get("bike"));

        ch1_repl.disconnect();
        assertTrue(ch1_repl.isOpen());
        assertFalse(ch1_repl.isConnected());
        Util.sleep(1000);
        assertServiceAndClusterView(ch1, 1, 1);

        c1_repl.clear();

        ch1_repl.connect("bla");
        assertTrue(ch1_repl.isOpen());
        assertTrue(ch1_repl.isConnected());
        assertServiceAndClusterView(ch1_repl, 2, 2);
        Util.sleep(300);
        assertServiceAndClusterView(ch1, 2, 2);

        assertEquals("cache has to be empty initially", 0, c1_repl.size());

        rc=ch1_repl.getState(null, 5000);
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("c1_repl: " + c1_repl);
        assertEquals("initial state should have been transferred", 4, c1_repl.size());

        assertEquals(new Long(322649), c1.get("id"));
        assertEquals(new Long(322649), c1_repl.get("id"));

        assertEquals("biking", c1.get("hobbies"));
        assertEquals("biking", c1_repl.get("hobbies"));

        assertEquals("Centurion", c1.get("bike"));
        assertEquals("Centurion", c1_repl.get("bike"));

        // Now see what happens if we reconnect the first channel
        // But first, add another MuxChannel on that JChannel
        // just so it remains coordinator (test that it doesn't
        // ask for state from itself)
        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        setCorrectPortRange(ch2);
        assertTrue(ch2.isOpen());
        assertFalse(ch2.isConnected());
        assertServiceAndClusterView(ch1, 2, 2);
        assertServiceAndClusterView(ch1_repl, 2, 2);


        ch1.disconnect();
        //sleep a bit and thus let asynch VIEW to propagate to other channel
        Util.sleep(500);
        assertTrue(ch1.isOpen());
        assertFalse(ch1.isConnected());
        assertServiceAndClusterView(ch1_repl, 1, 1);
        assertTrue(ch2.isOpen());
        assertFalse(ch2.isConnected());

        c1.clear();

        ch1.connect("bla");
        assertTrue(ch1.isOpen());
        assertTrue(ch1.isConnected());
        assertServiceAndClusterView(ch1, 2, 2);
        Util.sleep(500);
        assertServiceAndClusterView(ch1_repl, 2, 2);
        assertTrue(ch2.isOpen());
        assertFalse(ch2.isConnected());

        assertEquals("cache has to be empty initially", 0, c1.size());

        rc=ch1.getState(null, 5000);
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("c1: " + c1);
        assertEquals("initial state should have been transferred", 4, c1.size());

        assertEquals(new Long(322649), c1.get("id"));
        assertEquals(new Long(322649), c1_repl.get("id"));

        assertEquals("biking", c1.get("hobbies"));
        assertEquals("biking", c1_repl.get("hobbies"));

        assertEquals("Centurion", c1.get("bike"));
        assertEquals("Centurion", c1_repl.get("bike"));
    }


    private void assertServiceAndClusterView(Channel ch, int num_service_view_mbrs, int num_cluster_view_mbrs) {
        View service_view, cluster_view;
        service_view=ch.getView();
        cluster_view=((MuxChannel)ch).getClusterView();

        String msg="cluster view=" + cluster_view + ", service view=" + service_view;

        assertNotNull(service_view);
        assertNotNull(cluster_view);

        assertEquals(msg, num_service_view_mbrs, service_view.size());
        assertEquals(msg, num_cluster_view_mbrs, cluster_view.size());
    }


    public void testStateTransferFromSelfWithRegularChannel() throws Exception {
        JChannel ch=new JChannel();
        ch.connect("X");
        try {
            boolean rc=ch.getState(null, 2000);
            assertFalse("getState() on singleton should return false", rc);
        }
        finally {
            ch.close();
        }
    }

    public void testStateTransferFromSelf() throws Exception {
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        ch1.connect("bla");
        boolean rc=ch1.getState(null, 2000);
        assertFalse("getState() on singleton should return false", rc);
        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        ch2.connect("foo");
        rc=ch2.getState(null, 2000);
        assertFalse("getState() on singleton should return false", rc);
    }


    public void testAdditionalData() throws Exception {
        byte[] additional_data=new byte[]{'b', 'e', 'l', 'a'};
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        Map m=new HashMap(1);
        m.put("additional_data", additional_data);
        ch1.down(new Event(Event.CONFIG, m));
        ch1.connect("bla");
        IpAddress local_addr=(IpAddress)ch1.getLocalAddress();
        assertNotNull(local_addr);
        byte[] tmp=local_addr.getAdditionalData();
        assertNotNull(tmp);
        assertEquals(tmp, additional_data);

        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        ch2.connect("foo");
        local_addr=(IpAddress)ch2.getLocalAddress();
        assertNotNull(local_addr);
        tmp=local_addr.getAdditionalData();
        assertNotNull(tmp);
        assertEquals(tmp, additional_data);
    }

    public void testAdditionalData2() throws Exception {
        byte[] additional_data=new byte[]{'b', 'e', 'l', 'a'};
        byte[] additional_data2=new byte[]{'m', 'i', 'c', 'h', 'i'};
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        ch1.connect("bla");
        IpAddress local_addr=(IpAddress)ch1.getLocalAddress();
        assertNotNull(local_addr);
        byte[] tmp=local_addr.getAdditionalData();
        assertNull(tmp);

        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        Map m=new HashMap(1);
        m.put("additional_data", additional_data);
        ch2.down(new Event(Event.CONFIG, m));
        ch2.connect("foo");
        local_addr=(IpAddress)ch2.getLocalAddress();
        assertNotNull(local_addr);
        tmp=local_addr.getAdditionalData();
        assertNotNull(tmp);
        assertEquals(tmp, additional_data);

        local_addr=(IpAddress)ch1.getLocalAddress();
        assertNotNull(local_addr);
        tmp=local_addr.getAdditionalData();
        assertNotNull(tmp);
        assertEquals(tmp, additional_data);

        m.clear();
        m.put("additional_data", additional_data2);
        ch2.down(new Event(Event.CONFIG, m));
        local_addr=(IpAddress)ch2.getLocalAddress();
        assertNotNull(local_addr);
        tmp=local_addr.getAdditionalData();
        assertNotNull(tmp);
        assertEquals(tmp, additional_data2);
        assertFalse(Arrays.equals(tmp, additional_data));
    }

    public void testGetSubstates() throws Exception {
        ch1=factory.createMultiplexerChannel(STACK_NAME, "c1");
        ch1.connect("bla");
        c1=new ExtendedCache(ch1, "cache-1");
        assertEquals("cache has to be empty initially", 0, c1.size());

        ch2=factory.createMultiplexerChannel(STACK_NAME, "c2");
        ch2.connect("bla");
        c2=new ExtendedCache(ch2, "cache-2");
        assertEquals("cache has to be empty initially", 0, c2.size());

        for(int i=0; i < 10; i++) {
            c1.put(new Integer(i), new Integer(i));
            c2.put(new Integer(i), new Integer(i));
        }

        ch1_repl=factory2.createMultiplexerChannel(STACK_NAME, "c1");
        ch2_repl=factory2.createMultiplexerChannel(STACK_NAME, "c2");
        ch1_repl.connect("bla");
        c1_repl=new ExtendedCache(ch1_repl, "cache-1-repl");
        boolean rc=ch1_repl.getState(null, "odd", 5000);
        System.out.println("state transfer: " + rc);

        ch2_repl.connect("bla");
        c2_repl=new ExtendedCache(ch2_repl, "cache-2-repl");
        rc=ch2_repl.getState(null, "even", 5000);
        System.out.println("state transfer: " + rc);
        Util.sleep(500);

        System.out.println("Caches after state transfers:");
        System.out.println("c1: " + c1);