mina-core/src/test/java/testcase/MinaRegressionTest.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 testcase;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.filter.executor.OrderedThreadPoolExecutor;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.SocketConnector;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * TODO : Add documentation
 * @author <a href="http://mina.apache.org">Apache MINA Project</a>
 *
 */
public class MinaRegressionTest extends IoHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(MinaRegressionTest.class);

    public static final int MSG_SIZE = 5000;

    public static final int MSG_COUNT = 10;

    private static final int PORT = 23234;

    private static final int BUFFER_SIZE = 8192;

    private static final int TIMEOUT = 10000;

    public static final String OPEN = "open";

    public SocketAcceptor acceptor;

    public SocketConnector connector;

    private final Object LOCK = new Object();

    private static final ThreadFactory THREAD_FACTORY = new ThreadFactory() {
        public Thread newThread(final Runnable r) {
            return new Thread(null, r, "MinaThread", 64 * 1024);
        }
    };

    private OrderedThreadPoolExecutor executor;

    public static AtomicInteger sent = new AtomicInteger(0);

    public MinaRegressionTest() throws IOException {
        executor = new OrderedThreadPoolExecutor(0, 1000, 60, TimeUnit.SECONDS, THREAD_FACTORY);

        acceptor = new NioSocketAcceptor(Runtime.getRuntime().availableProcessors() + 1);
        acceptor.setReuseAddress(true);
        acceptor.getSessionConfig().setReceiveBufferSize(BUFFER_SIZE);

        acceptor.getFilterChain().addLast("threadPool", new ExecutorFilter(executor));
        acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MyProtocolCodecFactory()));

        connector = new NioSocketConnector(Runtime.getRuntime().availableProcessors() + 1);

        connector.setConnectTimeoutMillis(TIMEOUT);
        connector.getSessionConfig().setSendBufferSize(BUFFER_SIZE);
        connector.getSessionConfig().setReuseAddress(true);
    }

    public void connect() throws Exception {
        final InetSocketAddress socketAddress = new InetSocketAddress("0.0.0.0", PORT);

        acceptor.setHandler(new MyIoHandler(LOCK));

        acceptor.bind(socketAddress);
        connector.setHandler(this);

        final IoFutureListener<ConnectFuture> listener = new IoFutureListener<ConnectFuture>() {
            public void operationComplete(ConnectFuture future) {
                try {
                    logger.info("Write message to session " + future.getSession().getId());
                    final IoSession s = future.getSession();
                    IoBuffer wb = IoBuffer.allocate(MSG_SIZE);
                    wb.put(new byte[MSG_SIZE]);
                    wb.flip();
                    s.write(wb);
                } catch (Exception e) {
                    logger.error("Can't send message: {}", e.getMessage());
                }
            }
        };

        for (int i = 0; i < MSG_COUNT; i++) {
            ConnectFuture future = connector.connect(socketAddress);
            future.addListener(listener);
        }

        synchronized (LOCK) {
            LOCK.wait(50000);
        }

        connector.dispose();
        acceptor.unbind();
        acceptor.dispose();
        executor.shutdownNow();

        logger.info("Received: " + MyIoHandler.received.intValue());
        logger.info("Sent: " + sent.intValue());
        logger.info("FINISH");
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) {
        if (!(cause instanceof IOException)) {
            logger.error("Exception: ", cause);
        } else {
            logger.info("I/O error: " + cause.getMessage());
        }
        session.close(true);
    }

    @Override
    public void messageSent(IoSession session, Object message) throws Exception {
        sent.incrementAndGet();
    }

    public static void main(String[] args) throws Exception {
        logger.info("START");
        new MinaRegressionTest().connect();
    }
}