test/org/apache/catalina/tribes/test/NioSenderTest.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.catalina.tribes.test;

import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;

import org.apache.catalina.tribes.Channel;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.io.ChannelData;
import org.apache.catalina.tribes.io.XByteBuffer;
import org.apache.catalina.tribes.membership.MemberImpl;
import org.apache.catalina.tribes.transport.nio.NioSender;

public class NioSenderTest {
    private Selector selector = null;
    private int counter = 0;
    Member mbr;
    private static int testOptions = Channel.SEND_OPTIONS_DEFAULT;
    public NioSenderTest()  {
        // Default constructor
    }

    public synchronized int inc() {
        return ++counter;
    }

    public synchronized ChannelData getMessage(Member mbr) {
        String msg = "Thread-"+Thread.currentThread().getName()+" Message:"+inc();
        ChannelData data = new ChannelData(true);
        data.setMessage(new XByteBuffer(msg.getBytes(),false));
        data.setAddress(mbr);

        return data;
    }

    public void init() throws Exception {
        selector = Selector.open();
        mbr = new MemberImpl("localhost",4444,0);
        NioSender sender = new NioSender();
        sender.setDestination(mbr);
        sender.setDirectBuffer(true);
        sender.setSelector(selector);
        sender.setMessage(XByteBuffer.createDataPackage(getMessage(mbr)));
        sender.connect();
    }

    public void run() {
        while (true) {

            int selectedKeys = 0;
            try {
                selectedKeys = selector.select(100);
                //               if ( selectedKeys == 0 ) {
                //                   System.out.println("No registered interests. Sleeping for a second.");
                //                   Thread.sleep(100);
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }

            if (selectedKeys == 0) {
                continue;
            }

            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey sk = it.next();
                it.remove();
                try {
                    int readyOps = sk.readyOps();
                    sk.interestOps(sk.interestOps() & ~readyOps);
                    NioSender sender = (NioSender) sk.attachment();
                    if ( sender.process(sk, (testOptions&Channel.SEND_OPTIONS_USE_ACK)==Channel.SEND_OPTIONS_USE_ACK) ) {
                        System.out.println("Message completed for handler:"+sender);
                        Thread.sleep(2000);
                        sender.reset();
                        sender.setMessage(XByteBuffer.createDataPackage(getMessage(mbr)));
                    }


                } catch (Throwable t) {
                    t.printStackTrace();
                    return;
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        NioSenderTest sender = new NioSenderTest();
        sender.init();
        sender.run();
    }
}