[jboss-jira] [JBoss JIRA] Created: (JGRP-991) ReceiveLocalMessages does not work on MuxChannel

[Stefan Anzinger (JIRA)]

ReceiveLocalMessages does not work on MuxChannel
------------------------------------------------

                 Key: JGRP-991
                 URL: https://jira.jboss.org/jira/browse/JGRP-991
             Project: JGroups
          Issue Type: Bug
    Affects Versions: 2.6.10
         Environment: RHEL4 Linux/Windows XP
            Reporter: Stefan Anzinger
            Assignee: Bela Ban


Setting receive local messages to false does not work on Multiplex Channels. Reason for this is, that the local_addr is not set on multiplex channels - therefore the mux channel cannot decide whether this message comes from local or not.

See attached a small testcase. The testcase just gets the local_addr from the "physical channel" and puts it into the MuxChannel local_addr via reflection.
Here is the testcase:

import java.io.ByteArrayInputStream;
import java.lang.reflect.Field;
import org.w3c.dom.*;
import javax.xml.parsers.*;
import junit.framework.TestCase;
import org.jgroups.*;
import org.jgroups.mux.Multiplexer;

public class TestJGroups extends TestCase {
    /**
     * Tests the receiveLocalMessages feature on multiplex
     * channels.
     * 
     * <h2>Course of action</h2>
     * <ol>
     *   <li>Creates tow physical channels</li>
     *   <li>For each physical channel a mux channel is created</li>
     *   <li>Receive Local messages is turned off on the mux channels</li>
     *   <li>Local address has to be set manually after connecting the physical channels (This may be done automatically)</li>
     *   <li>Message is sent from the first mux to the second mux channel</li>
     * </ol>
     * 
     * @throws Exception
     */
    public void testMultiplex() throws Exception {
        JChannelFactory factory = new JChannelFactory(getConfig());
        JChannel c1 = newChannel("c1", (JChannel)factory.createChannel());
        JChannel c2 = newChannel("c2", (JChannel)factory.createChannel());

        JChannel mc1 = newMuxChannel(c1, "mc1");
        JChannel mc2 = newMuxChannel(c2, "mc2");

        mc1.setOpt(Channel.LOCAL, Boolean.FALSE);
        mc2.setOpt(Channel.LOCAL, Boolean.FALSE);
        mc1.connect("X");
        mc2.connect("Y");

        // By commenting this, the messages are also received locally
        setLocalAddress(c1, mc1);   
        setLocalAddress(c2, mc2);
        
        Thread.sleep(2000);
        
        mc1.send(new Message(null, null, "HELLO WORLD"));
        
        Thread.sleep(500);
        
        c2.disconnect();
        c1.disconnect();
    }
    
    private JChannel newMuxChannel(JChannel channel, String name) throws Exception {
        Multiplexer mux = new Multiplexer(channel);
        JChannel mc = (JChannel)mux.createMuxChannel("hello1", "ASDF2");
        
        mc.setOpt(Channel.LOCAL, Boolean.FALSE);
        return newChannel(name, mc);
    }
    
    private void setLocalAddress(JChannel parent, JChannel mux) throws Exception {
        // local_addr Needs to be set, unless the "Ignore local messages" feature would not work.
        Field f = org.jgroups.JChannel.class.getDeclaredField("local_addr");
        f.setAccessible(true);
        f.set(mux, parent.getLocalAddress());
    }
    
    private JChannel newChannel(final String id, final JChannel c1) throws Exception
    {
        c1.setReceiver(new ReceiverAdapter() {
            @Override
            public void receive(Message msg) {
                super.receive(msg);
                System.out.println(id + " Received message: " + msg + " msg: " + msg.getObject());
            }
            @Override
            public void viewAccepted(View new_view) {
//                try {
//                    Thread.sleep((int)(Math.random()*1000.0));
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
//                System.out.println(id + " New View (new_view): " + new_view);
//                System.out.println(id + " New View (Channel):  " + c1.getView());
            }
        });
        return c1;
    }
    
    private Element getConfig() throws Exception {
        String config = 
            "<config>                                                                           " +
            "  <UDP mcast_recv_buf_size=\"80000\" mcast_send_buf_size=\"150000\"                " +
            "      mcast_port=\"32233\" mcast_addr=\"228.0.0.1\"                                " +
            "      ip_ttl=\"32\"/>                                                              " +
            "  <PING/>                                                                          " +
            "  <FD_SOCK/>                                                                       " +
            "  <VERIFY_SUSPECT timeout=\"15000\"/>                                               " +
            "  <pbcast.NAKACK gc_lag=\"50\" retransmit_timeout=\"300,600,1200,2400,48000\"/>     " +
            "  <UNICAST timeout=\"600,1200,2000,25000\"/>                                        " +
            "  <FRAG frag_size=\"8096\"/>                                                       " +
            "  <pbcast.STABLE desired_avg_gossip=\"20000\" max_bytes=\"500000\"/>               " +
            "  <pbcast.GMS print_local_addr=\"true\" join_timeout=\"60000\" shun=\"true\" />     " +
            "</config>                                                                          ";
        
        DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
        DocumentBuilder db = dbf.newDocumentBuilder();
        Document d = db.parse(new ByteArrayInputStream(config.getBytes()));
        return d.getDocumentElement();
    }
}


-- 
This message is automatically generated by JIRA.
-
If you think it was sent incorrectly contact one of the administrators: https://jira.jboss.org/jira/secure/Administrators.jspa
-
For more information on JIRA, see: http://www.atlassian.com/software/jira