1 17 package org.apache.activemq.transport.reliable; 18 19 import org.apache.activemq.openwire.OpenWireFormat; 20 import org.apache.activemq.transport.udp.ByteBufferPool; 21 import org.apache.activemq.transport.udp.CommandDatagramChannel; 22 import org.apache.activemq.transport.udp.DatagramHeaderMarshaller; 23 import org.apache.activemq.transport.udp.UdpTransport; 24 import org.apache.commons.logging.Log; 25 import org.apache.commons.logging.LogFactory; 26 27 import java.io.IOException ; 28 import java.net.SocketAddress ; 29 import java.nio.ByteBuffer ; 30 import java.nio.channels.DatagramChannel ; 31 32 36 public class UnreliableCommandDatagramChannel extends CommandDatagramChannel { 37 38 private static final Log log = LogFactory.getLog(UnreliableCommandDatagramChannel.class); 39 40 private DropCommandStrategy dropCommandStrategy; 41 42 public UnreliableCommandDatagramChannel(UdpTransport transport, OpenWireFormat wireFormat, int datagramSize, 43 SocketAddress targetAddress, DatagramHeaderMarshaller headerMarshaller, ReplayBuffer replayBuffer, DatagramChannel channel, 44 ByteBufferPool bufferPool, DropCommandStrategy strategy) { 45 super(transport, wireFormat, datagramSize, targetAddress, headerMarshaller, channel, bufferPool); 46 this.dropCommandStrategy = strategy; 47 } 48 49 protected void sendWriteBuffer(int commandId, SocketAddress address, ByteBuffer writeBuffer, boolean redelivery) throws IOException { 50 if (dropCommandStrategy.shouldDropCommand(commandId, address, redelivery)) { 51 writeBuffer.flip(); 52 log.info("Dropping datagram with command: " + commandId); 53 54 getReplayBuffer().addBuffer(commandId, writeBuffer); 56 } 57 else { 58 super.sendWriteBuffer(commandId, address, writeBuffer, redelivery); 59 } 60 } 61 } 62 | Popular Tags |