001: /*
002: * Licensed to the Apache Software Foundation (ASF) under one or more
003: * contributor license agreements. See the NOTICE file distributed with
004: * this work for additional information regarding copyright ownership.
005: * The ASF licenses this file to You under the Apache License, Version 2.0
006: * (the "License"); you may not use this file except in compliance with
007: * the License. You may obtain a copy of the License at
008: *
009: * http://www.apache.org/licenses/LICENSE-2.0
010: *
011: * Unless required by applicable law or agreed to in writing, software
012: * distributed under the License is distributed on an "AS IS" BASIS,
013: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014: * See the License for the specific language governing permissions and
015: * limitations under the License.
016: */
017: package org.apache.catalina.tribes.test.transport;
018:
019: import java.io.OutputStream;
020: import java.net.Socket;
021: import java.text.DecimalFormat;
022: import org.apache.catalina.tribes.transport.nio.NioSender;
023: import org.apache.catalina.tribes.membership.MemberImpl;
024: import java.nio.channels.Selector;
025: import org.apache.catalina.tribes.io.XByteBuffer;
026: import org.apache.catalina.tribes.Member;
027: import java.nio.channels.SelectionKey;
028: import java.util.Iterator;
029: import org.apache.catalina.tribes.Channel;
030: import org.apache.catalina.tribes.io.ChannelData;
031: import java.math.BigDecimal;
032: import java.util.Arrays;
033:
034: public class SocketNioValidateSend {
035:
036: public static void main(String[] args) throws Exception {
037: Selector selector = Selector.open();
038: Member mbr = new MemberImpl("localhost", 9999, 0);
039: byte seq = 0;
040: byte[] buf = new byte[50000];
041: Arrays.fill(buf, seq);
042: int len = buf.length;
043: BigDecimal total = new BigDecimal((double) 0);
044: BigDecimal bytes = new BigDecimal((double) len);
045: NioSender sender = new NioSender();
046: sender.setDestination(mbr);
047: sender.setDirectBuffer(true);
048: sender.setSelector(selector);
049: sender.connect();
050: sender.setMessage(buf);
051: System.out.println("Writing to 9999");
052: long start = 0;
053: double mb = 0;
054: boolean first = true;
055: int count = 0;
056:
057: DecimalFormat df = new DecimalFormat("##.00");
058: while (count < 100000) {
059: if (first) {
060: first = false;
061: start = System.currentTimeMillis();
062: }
063: sender.setMessage(buf);
064: int selectedKeys = 0;
065: try {
066: selectedKeys = selector.select(0);
067: } catch (Exception e) {
068: e.printStackTrace();
069: continue;
070: }
071:
072: if (selectedKeys == 0) {
073: continue;
074: }
075:
076: Iterator it = selector.selectedKeys().iterator();
077: while (it.hasNext()) {
078: SelectionKey sk = (SelectionKey) it.next();
079: it.remove();
080: try {
081: int readyOps = sk.readyOps();
082: sk.interestOps(sk.interestOps() & ~readyOps);
083: if (sender.process(sk, false)) {
084: total = total.add(bytes);
085: sender.reset();
086: seq++;
087: Arrays.fill(buf, seq);
088: sender.setMessage(buf);
089: mb += ((double) len) / 1024 / 1024;
090: if (((++count) % 10000) == 0) {
091: long time = System.currentTimeMillis();
092: double seconds = ((double) (time - start)) / 1000;
093: System.out
094: .println("Throughput "
095: + df.format(mb / seconds)
096: + " MB/seconds, total "
097: + mb + " MB, total "
098: + total + " bytes.");
099: }
100: }
101:
102: } catch (Throwable t) {
103: t.printStackTrace();
104: return;
105: }
106: }
107: }
108: System.out.println("Complete, sleeping 15 seconds");
109: Thread.sleep(15000);
110: }
111: }
|