/*
 * 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.transport;

import java.text.DecimalFormat;

import org.apache.catalina.tribes.ChannelMessage;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.MessageListener;
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.NioReceiver;

public class SocketNioReceive {
    static int count = 0;
    static int accept = 0;
    static long start = 0;
    static double mb = 0;
    static int len = 0;
    static DecimalFormat df = new DecimalFormat("##.00");
    static double seconds = 0;

    protected static Object mutex = new Object();
    public static void main(String[] args) throws Exception {
        Member mbr = new MemberImpl("localhost", 9999, 0);
        ChannelData data = new ChannelData();
        data.setAddress(mbr);
        byte[] buf = new byte[8192 * 4];
        data.setMessage(new XByteBuffer(buf, false));
        buf = XByteBuffer.createDataPackage(data);
        len = buf.length;
        NioReceiver receiver = new NioReceiver();
        receiver.setPort(9999);
        receiver.setHost("localhost");
        MyList list = new MyList();
        receiver.setMessageListener(list);
        receiver.start();
        System.out.println("Listening on 9999");
        while (true) {
            try {
                synchronized (mutex) {
                    mutex.wait(5000);
                    if ( start != 0 ) {
                        System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, messages "+count+" accepts "+accept+", total "+mb+" MB.");
                    }
                }
            }catch (Throwable x) {
                x.printStackTrace();
            }
        }
    }
    
    public static class MyList implements MessageListener {
        boolean first = true;
        
        
        public void messageReceived(ChannelMessage msg) {
            if (first) {
                first = false;
                start = System.currentTimeMillis();
            }
            mb += ( (double) len) / 1024 / 1024;
            synchronized (this) {count++;}
            if ( ( (count) % 10000) == 0) {
                long time = System.currentTimeMillis();
                seconds = ( (double) (time - start)) / 1000;
                System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, messages "+count+", total "+mb+" MB.");
            }
        }        

        public boolean accept(ChannelMessage msg) {
            synchronized (this) {accept++;}
            return true;
        }

    }
}