Java 类org.apache.catalina.tribes.io.ChannelData 实例源码

项目：tomcat7 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：tomcat7 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：apache-tomcat-7.0.73-with-comment 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：apache-tomcat-7.0.73-with-comment 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：lazycat 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
        throws ChannelException {
    if (access.addAndGet(1) == 1)
        txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    } catch (ChannelException x) {
        msgTxErr.addAndGet(1);
        if (access.get() == 1)
            access.addAndGet(-1);
        throw x;
    }
    mbTx += (bytes * destination.length) / (1024d * 1024d);
    mbAppTx += bytes / (1024d * 1024d);
    if (access.addAndGet(-1) == 0) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：lazycat 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl == null || (impl.startLevel & Channel.MBR_TX_SEQ) != Channel.MBR_TX_SEQ)
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData) message);
    if (data.length > McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length[" + data.length + "] exceeds max packet size of "
                + McastServiceImpl.MAX_PACKET_SIZE + " bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data, 0, data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：class-guard 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：class-guard 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：apache-tomcat-7.0.57 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：apache-tomcat-7.0.57 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：apache-tomcat-7.0.57 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：apache-tomcat-7.0.57 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：WBSAirback 文件：ThroughputInterceptor.java

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

项目：WBSAirback 文件：McastService.java

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

项目：tomcat7 文件：ThroughputInterceptor.java

@Override
public void messageReceived(ChannelMessage msg) {
    if ( rxStart == 0 ) rxStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    mbRx += bytes/(1024d*1024d);
    msgRxCnt.addAndGet(1);
    if ( msgRxCnt.get() % interval == 0 ) report(timeTx);
    super.messageReceived(msg);

}

项目：tomcat7 文件：NonBlockingCoordinator.java

public ChannelData createData(CoordinationMessage msg, MemberImpl local) {
    msg.write();
    ChannelData data = new ChannelData(true);
    data.setAddress(local);
    data.setMessage(msg.getBuffer());
    data.setOptions(Channel.SEND_OPTIONS_USE_ACK);
    data.setTimestamp(System.currentTimeMillis());
    return data;
}

项目：tomcat7 文件：TcpPingInterceptor.java

protected void sendPingMessage(Member[] members) {
    if ( members == null || members.length == 0 ) return;
    ChannelData data = new ChannelData(true);//generates a unique Id
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
    try {
        super.sendMessage(members, data, null);
    }catch (ChannelException x) {
        log.warn("Unable to send TCP ping.",x);
    }
}

项目：tomcat7 文件：StaticMembershipInterceptor.java

protected void sendMemberMessage(Member[] members, byte[] message) throws ChannelException {
    if ( members == null || members.length == 0 ) return;
    ChannelData data = new ChannelData(true);
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(message, false));
    super.sendMessage(members, data, null);
}

项目：tomcat7 文件：BioReplicationTask.java

protected void execute(ObjectReader reader) throws Exception{
    int pkgcnt = reader.count();

    if ( pkgcnt > 0 ) {
        ChannelMessage[] msgs = reader.execute();
        for ( int i=0; i<msgs.length; i++ ) {
            /**
             * Use send ack here if you want to ack the request to the remote 
             * server before completing the request
             * This is considered an asynchronized request
             */
            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            try {
                //process the message
                getCallback().messageDataReceived(msgs[i]);
                /**
                 * Use send ack here if you want the request to complete on this
                 * server before sending the ack to the remote server
                 * This is considered a synchronized request
                 */
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            }catch  ( Exception x ) {
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.FAIL_ACK_COMMAND);
                log.error("Error thrown from messageDataReceived.",x);
            }
            if ( getUseBufferPool() ) {
                BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                msgs[i].setMessage(null);
            }
        }                       
    }


}

项目：tomcat7 文件：SocketNioReceive.java

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();
        }
    }
}

项目：tomcat7 文件：NioSenderTest.java

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;
}

项目：apache-tomcat-7.0.73-with-comment 文件：ThroughputInterceptor.java

@Override
public void messageReceived(ChannelMessage msg) {
    if ( rxStart == 0 ) rxStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    mbRx += bytes/(1024d*1024d);
    msgRxCnt.addAndGet(1);
    if ( msgRxCnt.get() % interval == 0 ) report(timeTx);
    super.messageReceived(msg);

}

项目：apache-tomcat-7.0.73-with-comment 文件：NonBlockingCoordinator.java

public ChannelData createData(CoordinationMessage msg, MemberImpl local) {
    msg.write();
    ChannelData data = new ChannelData(true);
    data.setAddress(local);
    data.setMessage(msg.getBuffer());
    data.setOptions(Channel.SEND_OPTIONS_USE_ACK);
    data.setTimestamp(System.currentTimeMillis());
    return data;
}

项目：apache-tomcat-7.0.73-with-comment 文件：TcpPingInterceptor.java

protected void sendPingMessage(Member[] members) {
    if ( members == null || members.length == 0 ) return;
    ChannelData data = new ChannelData(true);//generates a unique Id
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
    try {
        super.sendMessage(members, data, null);
    }catch (ChannelException x) {
        log.warn("Unable to send TCP ping.",x);
    }
}

项目：apache-tomcat-7.0.73-with-comment 文件：StaticMembershipInterceptor.java

protected void sendMemberMessage(Member[] members, byte[] message) throws ChannelException {
    if ( members == null || members.length == 0 ) return;
    ChannelData data = new ChannelData(true);
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(message, false));
    super.sendMessage(members, data, null);
}

项目：apache-tomcat-7.0.73-with-comment 文件：BioReplicationTask.java

protected void execute(ObjectReader reader) throws Exception{
    int pkgcnt = reader.count();

    if ( pkgcnt > 0 ) {
        ChannelMessage[] msgs = reader.execute();
        for ( int i=0; i<msgs.length; i++ ) {
            /**
             * Use send ack here if you want to ack the request to the remote 
             * server before completing the request
             * This is considered an asynchronized request
             */
            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            try {
                //process the message
                getCallback().messageDataReceived(msgs[i]);
                /**
                 * Use send ack here if you want the request to complete on this
                 * server before sending the ack to the remote server
                 * This is considered a synchronized request
                 */
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            }catch  ( Exception x ) {
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.FAIL_ACK_COMMAND);
                log.error("Error thrown from messageDataReceived.",x);
            }
            if ( getUseBufferPool() ) {
                BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                msgs[i].setMessage(null);
            }
        }                       
    }


}

项目：apache-tomcat-7.0.73-with-comment 文件：SocketNioReceive.java

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();
        }
    }
}

项目：apache-tomcat-7.0.73-with-comment 文件：NioSenderTest.java

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;
}

项目：lazycat 文件：ThroughputInterceptor.java

@Override
public void messageReceived(ChannelMessage msg) {
    if (rxStart == 0)
        rxStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData) msg).getDataPackageLength());
    mbRx += bytes / (1024d * 1024d);
    msgRxCnt.addAndGet(1);
    if (msgRxCnt.get() % interval == 0)
        report(timeTx);
    super.messageReceived(msg);

}

项目：lazycat 文件：NonBlockingCoordinator.java

public ChannelData createData(CoordinationMessage msg, MemberImpl local) {
    msg.write();
    ChannelData data = new ChannelData(true);
    data.setAddress(local);
    data.setMessage(msg.getBuffer());
    data.setOptions(Channel.SEND_OPTIONS_USE_ACK);
    data.setTimestamp(System.currentTimeMillis());
    return data;
}

项目：lazycat 文件：TcpPingInterceptor.java

protected void sendPingMessage(Member[] members) {
    if (members == null || members.length == 0)
        return;
    ChannelData data = new ChannelData(true);// generates a unique Id
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
    try {
        super.sendMessage(members, data, null);
    } catch (ChannelException x) {
        log.warn("Unable to send TCP ping.", x);
    }
}

项目：lazycat 文件：StaticMembershipInterceptor.java

protected void sendMemberMessage(Member[] members, byte[] message) throws ChannelException {
    if (members == null || members.length == 0)
        return;
    ChannelData data = new ChannelData(true);
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(message, false));
    super.sendMessage(members, data, null);
}

项目：lazycat 文件：BioReplicationTask.java

protected void execute(ObjectReader reader) throws Exception {
    int pkgcnt = reader.count();

    if (pkgcnt > 0) {
        ChannelMessage[] msgs = reader.execute();
        for (int i = 0; i < msgs.length; i++) {
            /**
             * Use send ack here if you want to ack the request to the
             * remote server before completing the request This is
             * considered an asynchronous request
             */
            if (ChannelData.sendAckAsync(msgs[i].getOptions()))
                sendAck(Constants.ACK_COMMAND);
            try {
                // process the message
                getCallback().messageDataReceived(msgs[i]);
                /**
                 * Use send ack here if you want the request to complete on
                 * this server before sending the ack to the remote server
                 * This is considered a synchronized request
                 */
                if (ChannelData.sendAckSync(msgs[i].getOptions()))
                    sendAck(Constants.ACK_COMMAND);
            } catch (Exception x) {
                if (ChannelData.sendAckSync(msgs[i].getOptions()))
                    sendAck(Constants.FAIL_ACK_COMMAND);
                log.error("Error thrown from messageDataReceived.", x);
            }
            if (getUseBufferPool()) {
                BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                msgs[i].setMessage(null);
            }
        }
    }

}

项目：class-guard 文件：ThroughputInterceptor.java

@Override
public void messageReceived(ChannelMessage msg) {
    if ( rxStart == 0 ) rxStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    mbRx += bytes/(1024d*1024d);
    msgRxCnt.addAndGet(1);
    if ( msgRxCnt.get() % interval == 0 ) report(timeTx);
    super.messageReceived(msg);

}

项目：class-guard 文件：NonBlockingCoordinator.java

public ChannelData createData(CoordinationMessage msg, MemberImpl local) {
    msg.write();
    ChannelData data = new ChannelData(true);
    data.setAddress(local);
    data.setMessage(msg.getBuffer());
    data.setOptions(Channel.SEND_OPTIONS_USE_ACK);
    data.setTimestamp(System.currentTimeMillis());
    return data;
}

项目：class-guard 文件：TcpPingInterceptor.java

protected void sendPingMessage(Member[] members) {
    if ( members == null || members.length == 0 ) return;
    ChannelData data = new ChannelData(true);//generates a unique Id
    data.setAddress(getLocalMember(false));
    data.setTimestamp(System.currentTimeMillis());
    data.setOptions(getOptionFlag());
    data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
    try {
        super.sendMessage(members, data, null);
    }catch (ChannelException x) {
        log.warn("Unable to send TCP ping.",x);
    }
}

项目：class-guard 文件：MultipointBioSender.java

@Override
public synchronized void sendMessage(Member[] destination, ChannelMessage msg) throws ChannelException {
    byte[] data = XByteBuffer.createDataPackage((ChannelData)msg);
    BioSender[] senders = setupForSend(destination);
    ChannelException cx = null;
    for ( int i=0; i<senders.length; i++ ) {
        try {
            senders[i].sendMessage(data,(msg.getOptions()&Channel.SEND_OPTIONS_USE_ACK)==Channel.SEND_OPTIONS_USE_ACK);
        } catch (Exception x) {
            if (cx == null) cx = new ChannelException(x);
            cx.addFaultyMember(destination[i],x);
        }
    }
    if (cx!=null ) throw cx;
}

项目：class-guard 文件：BioReplicationTask.java

protected void execute(ObjectReader reader) throws Exception{
    int pkgcnt = reader.count();

    if ( pkgcnt > 0 ) {
        ChannelMessage[] msgs = reader.execute();
        for ( int i=0; i<msgs.length; i++ ) {
            /**
             * Use send ack here if you want to ack the request to the remote 
             * server before completing the request
             * This is considered an asynchronized request
             */
            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            try {
                //process the message
                getCallback().messageDataReceived(msgs[i]);
                /**
                 * Use send ack here if you want the request to complete on this
                 * server before sending the ack to the remote server
                 * This is considered a synchronized request
                 */
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            }catch  ( Exception x ) {
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.FAIL_ACK_COMMAND);
                log.error("Error thrown from messageDataReceived.",x);
            }
            if ( getUseBufferPool() ) {
                BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                msgs[i].setMessage(null);
            }
        }                       
    }


}

项目：class-guard 文件：SocketNioReceive.java

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();
        }
    }
}

项目：class-guard 文件：NioSenderTest.java

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;
}