activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jstrac...@apache.org
Subject svn commit: r384603 - in /incubator/activemq/trunk/activemq-core: ./ src/main/java/org/apache/activemq/openwire/ src/main/java/org/apache/activemq/transport/ src/main/java/org/apache/activemq/transport/udp/ src/main/java/org/apache/activemq/transport/u...
Date Thu, 09 Mar 2006 20:26:50 GMT
Author: jstrachan
Date: Thu Mar  9 12:26:47 2006
New Revision: 384603

URL: http://svn.apache.org/viewcvs?rev=384603&view=rev
Log:
added a working UDP server with test cases

Added:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
  (with props)
Modified:
    incubator/activemq/trunk/activemq-core/project.xml
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandReadBuffer.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/DatagramReplayStrategy.java
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/ExceptionIfDroppedPacketStrategy.java
    incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
    incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java

Modified: incubator/activemq/trunk/activemq-core/project.xml
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/project.xml?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/project.xml (original)
+++ incubator/activemq/trunk/activemq-core/project.xml Thu Mar  9 12:26:47 2006
@@ -360,10 +360,6 @@
                 <exclude>**/MultipleTestsWithSpringFactoryBeanTest.*</exclude>
                 <exclude>**/MultipleTestsWithXBeanFactoryBeanTest.*</exclude>
                 <exclude>**/MultipleTestsWithSpringXBeanFactoryBeanTest.*</exclude>
-                
-                <!-- TODO FIXME -->
-                <exclude>**/UdpTransportUsingServerTest.*</exclude>
-                <exclude>**/UdpSendReceiveWithTwoConnectionsTest.*</exclude>
             </excludes>
         </unitTest>
         <resources>

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/openwire/OpenWireFormat.java
Thu Mar  9 12:26:47 2006
@@ -81,11 +81,13 @@
     
     public OpenWireFormat copy() {
         OpenWireFormat answer = new OpenWireFormat();
+        answer.version = version;
         answer.stackTraceEnabled = stackTraceEnabled;
         answer.tcpNoDelayEnabled = tcpNoDelayEnabled;
         answer.cacheEnabled = cacheEnabled;
         answer.tightEncodingEnabled = tightEncodingEnabled;
         answer.sizePrefixDisabled = sizePrefixDisabled;
+        answer.preferedWireFormatInfo = preferedWireFormatInfo;
         return answer;
     }
     
@@ -104,8 +106,8 @@
     static IdGenerator g = new IdGenerator();
     String id = g.generateId();
     public String toString() {
-        //return "OpenWireFormat{version="+version+", cacheEnabled="+cacheEnabled+", stackTraceEnabled="+stackTraceEnabled+",
tightEncodingEnabled="+tightEncodingEnabled+", sizePrefixDisabled="+sizePrefixDisabled+"}";
-        return "OpenWireFormat{id="+id+", tightEncodingEnabled="+tightEncodingEnabled+"}";
+        return "OpenWireFormat{version="+version+", cacheEnabled="+cacheEnabled+", stackTraceEnabled="+stackTraceEnabled+",
tightEncodingEnabled="+tightEncodingEnabled+", sizePrefixDisabled="+sizePrefixDisabled+"}";
+        //return "OpenWireFormat{id="+id+", tightEncodingEnabled="+tightEncodingEnabled+"}";
     }
     
     public int getVersion() {

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
Thu Mar  9 12:26:47 2006
@@ -94,7 +94,14 @@
                     onException(new IOException("Remote wire format ("+info.getVersion()+")
is lower the minimum version required ("+minimumVersion+")"));
                 }
                 
+                if (log.isDebugEnabled()) {
+                    log.debug(this + " before negotiation: " + wireFormat);
+                }
                 wireFormat.renegociatWireFormat(info);
+                
+                if (log.isDebugEnabled()) {
+                    log.debug(this + " after negotiation: " + wireFormat);
+                }
 	
             } catch (IOException e) {
                 onException(e);

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandChannel.java
Thu Mar  9 12:26:47 2006
@@ -43,6 +43,7 @@
 
     private static final Log log = LogFactory.getLog(CommandChannel.class);
 
+    private final String name;
     private DatagramChannel channel;
     private OpenWireFormat wireFormat;
     private ByteBufferPool bufferPool;
@@ -50,11 +51,12 @@
     private DatagramReplayStrategy replayStrategy;
     private SocketAddress targetAddress;
     private DatagramHeaderMarshaller headerMarshaller = new DatagramHeaderMarshaller();
+    private final boolean checkSequenceNumbers;
 
     // reading
     private Object readLock = new Object();
     private ByteBuffer readBuffer;
-    private CommandReadBuffer readStack;
+    private DatagramReadBuffer readStack;
     private SocketAddress lastReadDatagramAddress;
 
     // writing
@@ -64,14 +66,20 @@
     private int largeMessageBufferSize = 128 * 1024;
     private DatagramHeader header = new DatagramHeader();
 
-    public CommandChannel(DatagramChannel channel, OpenWireFormat wireFormat, ByteBufferPool
bufferPool, int datagramSize,
-            DatagramReplayStrategy replayStrategy, SocketAddress targetAddress) {
+    public CommandChannel(String name, DatagramChannel channel, OpenWireFormat wireFormat,
ByteBufferPool bufferPool, int datagramSize,
+            DatagramReplayStrategy replayStrategy, SocketAddress targetAddress, boolean checkSequenceNumbers)
{
+        this.name = name;
         this.channel = channel;
         this.wireFormat = wireFormat;
         this.bufferPool = bufferPool;
         this.datagramSize = datagramSize;
         this.replayStrategy = replayStrategy;
         this.targetAddress = targetAddress;
+        this.checkSequenceNumbers = checkSequenceNumbers;
+    }
+
+    public String toString() {
+        return "CommandChannel#" + name;
     }
 
     public void start() throws Exception {
@@ -79,7 +87,9 @@
         wireFormat.setCacheEnabled(false);
         wireFormat.setTightEncodingEnabled(true);
 
-        readStack = new CommandReadBuffer(wireFormat, replayStrategy);
+        if (checkSequenceNumbers) {
+            readStack = new CommandReadBuffer(name, wireFormat, replayStrategy);
+        }
         bufferPool.setDefaultSize(datagramSize);
         bufferPool.start();
         readBuffer = bufferPool.borrowBuffer();
@@ -98,26 +108,21 @@
             readBuffer.clear();
             lastReadDatagramAddress = channel.receive(readBuffer);
             readBuffer.flip();
-            
-            if (log.isDebugEnabled()) {
-                log.debug("Read a datagram from: " + lastReadDatagramAddress);
-            }
+
             header = headerMarshaller.readHeader(readBuffer);
             header.setFromAddress(lastReadDatagramAddress);
 
             if (log.isDebugEnabled()) {
-                log.debug("Received datagram from: " + lastReadDatagramAddress + " header:
" + header);
+                log.debug("Received datagram on: " + name + " from: " + lastReadDatagramAddress
+ " header: " + header);
             }
             int remaining = readBuffer.remaining();
             int size = header.getDataSize();
             /*
-            if (size > remaining) {
-                throw new IOException("Invalid command size: " + size + " when there are
only: " + remaining + " byte(s) remaining");
-            }
-            else if (size < remaining) {
-                log.warn("Extra bytes in buffer. Expecting: " + size + " but has: " + remaining);
-            }
-            */
+             * if (size > remaining) { throw new IOException("Invalid command
+             * size: " + size + " when there are only: " + remaining + " byte(s)
+             * remaining"); } else if (size < remaining) { log.warn("Extra bytes
+             * in buffer. Expecting: " + size + " but has: " + remaining); }
+             */
             if (size != remaining) {
                 log.warn("Expecting: " + size + " but has: " + remaining);
             }
@@ -133,23 +138,39 @@
                 // TODO use a DataInput implementation that talks direct to the
                 // ByteBuffer
                 DataInputStream dataIn = new DataInputStream(new ByteArrayInputStream(data));
-                Command command = (Command) wireFormat.doUnmarshal(dataIn);
+                Command command = (Command) wireFormat.unmarshal(dataIn);
+                // Command command = (Command) wireFormat.doUnmarshal(dataIn);
                 header.setCommand(command);
             }
 
-            answer = readStack.read(header);
+            if (readStack != null) {
+                answer = readStack.read(header);
+            }
+            else {
+                answer = header.getCommand();
+            }
         }
         if (answer != null) {
+            if (log.isDebugEnabled()) {
+                log.debug("Channel: " + name + " about to process: " + answer);
+            }
             processor.process(answer, header);
         }
     }
 
     /**
-     * Called if a packet is received on a different channel from a remote client
-     * @throws IOException 
+     * Called if a packet is received on a different channel from a remote
+     * client
+     * 
+     * @throws IOException
      */
     public Command onDatagramReceived(DatagramHeader header) throws IOException {
-        return readStack.read(header);
+        if (readStack != null) {
+            return readStack.read(header);
+        }
+        else {
+            return header.getCommand();
+        }
     }
 
     public void write(Command command) throws IOException {
@@ -159,10 +180,12 @@
     public void write(Command command, SocketAddress address) throws IOException {
         synchronized (writeLock) {
             header.incrementCounter();
-            bs = new BooleanStream();
-            // TODO
-            //bs.clear();
-            int size = wireFormat.tightMarshal1(command, bs);
+
+            ByteArrayOutputStream largeBuffer = new ByteArrayOutputStream(largeMessageBufferSize);
+            wireFormat.marshal(command, new DataOutputStream(largeBuffer));
+            byte[] data = largeBuffer.toByteArray();
+            int size = data.length;
+
             if (size < datagramSize) {
                 header.setPartial(false);
                 header.setComplete(true);
@@ -170,13 +193,6 @@
                 writeBuffer.clear();
                 headerMarshaller.writeHeader(header, writeBuffer);
 
-                // TODO use a DataOutput implementation that talks direct to the
-                // ByteBuffer
-                ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-                DataOutputStream dataOut = new DataOutputStream(buffer);
-                wireFormat.tightMarshal2(command, dataOut, bs);
-                dataOut.close();
-                byte[] data = buffer.toByteArray();
                 writeBuffer.put(data);
 
                 sendWriteBuffer(address);
@@ -186,10 +202,7 @@
                 header.setComplete(false);
 
                 // lets split the command up into chunks
-                ByteArrayOutputStream largeBuffer = new ByteArrayOutputStream(largeMessageBufferSize);
-                wireFormat.marshal(command, new DataOutputStream(largeBuffer));
 
-                byte[] data = largeBuffer.toByteArray();
                 int offset = 0;
                 boolean lastFragment = false;
                 for (int fragment = 0, length = data.length; !lastFragment; fragment++) {
@@ -248,15 +261,14 @@
             return lastReadDatagramAddress;
         }
     }
-    
 
     // Implementation methods
     // -------------------------------------------------------------------------
     protected void sendWriteBuffer(SocketAddress address) throws IOException {
         writeBuffer.flip();
-        
+
         if (log.isDebugEnabled()) {
-            log.debug("Sending datagram to: " + address + " header: " + header);
+            log.debug("Channel: " + name + " sending datagram to: " + address + " header:
" + header);
         }
         channel.send(writeBuffer, address);
     }

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandReadBuffer.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandReadBuffer.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandReadBuffer.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/CommandReadBuffer.java
Thu Mar  9 12:26:47 2006
@@ -35,7 +35,7 @@
  * 
  * @version $Revision$
  */
-public class CommandReadBuffer {
+public class CommandReadBuffer implements DatagramReadBuffer {
     private static final Log log = LogFactory.getLog(CommandReadBuffer.class);
 
     private OpenWireFormat wireFormat;
@@ -43,8 +43,10 @@
     private SortedSet headers = new TreeSet();
     private long expectedCounter = 1;
     private ByteArrayOutputStream out = new ByteArrayOutputStream();
+    private final String name;
 
-    public CommandReadBuffer(OpenWireFormat wireFormat, DatagramReplayStrategy replayStrategy)
{
+    public CommandReadBuffer(String name, OpenWireFormat wireFormat, DatagramReplayStrategy
replayStrategy) {
+        this.name = name;
         this.wireFormat = wireFormat;
         this.replayStrategy = replayStrategy;
     }
@@ -57,13 +59,16 @@
                 log.warn("Ignoring out of step packet: " + header);
             }
             else {
-                replayStrategy.onDroppedPackets(expectedCounter, actualCounter);
+                replayStrategy.onDroppedPackets(name, expectedCounter, actualCounter);
                 
                 // lets add it to the list for later on
                 headers.add(header);
             }
 
             // lets see if the first item in the set is the next header
+            if (headers.isEmpty()) {
+                return null;
+            }
             header = (DatagramHeader) headers.first();
             if (expectedCounter != header.getCounter()) {
                 return null;
@@ -71,7 +76,7 @@
         }
 
         // we've got a valid header so increment counter
-        replayStrategy.onReceivedPacket(expectedCounter);
+        replayStrategy.onReceivedPacket(name, expectedCounter);
         expectedCounter++;
 
         Command answer = null;

Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java?rev=384603&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
(added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
Thu Mar  9 12:26:47 2006
@@ -0,0 +1,33 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed 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.activemq.transport.udp;
+
+import org.apache.activemq.command.Command;
+
+import java.io.IOException;
+
+/**
+ * Represents an inbound buffer of datagrams for dealing with out of order
+ * or fragmented commands.
+ * 
+ * @version $Revision$
+ */
+public interface DatagramReadBuffer {
+
+    Command read(DatagramHeader header) throws IOException;
+
+}
\ No newline at end of file

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/DatagramReadBuffer.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
Thu Mar  9 12:26:47 2006
@@ -56,6 +56,7 @@
     private DatagramChannel channel;
     private boolean trace = false;
     private boolean useLocalHost = true;
+    private boolean checkSequenceNumbers = true;
     private int port;
     private int minmumWireFormatVersion;
     private String description = null;
@@ -112,7 +113,8 @@
         commandChannel.write(command, address);
     }
 
-    public void doConsume(Command command, DatagramHeader header) throws IOException {
+
+    public void receivedHeader(DatagramHeader header) {
         wireFormatHeader = header;
     }
 
@@ -253,6 +255,14 @@
     public OpenWireFormat getWireFormat() {
         return wireFormat;
     }
+    
+    public boolean isCheckSequenceNumbers() {
+        return checkSequenceNumbers;
+    }
+
+    public void setCheckSequenceNumbers(boolean checkSequenceNumbers) {
+        this.checkSequenceNumbers = checkSequenceNumbers;
+    }
 
     // Implementation methods
     // -------------------------------------------------------------------------
@@ -303,7 +313,7 @@
         if (bufferPool == null) {
             bufferPool = new DefaultBufferPool();
         }
-        commandChannel = new CommandChannel(channel, wireFormat, bufferPool, datagramSize,
replayStrategy, targetAddress);
+        commandChannel = new CommandChannel(toString(), channel, wireFormat, bufferPool,
datagramSize, replayStrategy, targetAddress, isCheckSequenceNumbers());
         commandChannel.start();
 
         // lets pass the header & address into the channel so it avoids a
@@ -336,5 +346,6 @@
             targetAddress = lastAddress;
         }
     }
+
 
 }

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportFactory.java
Thu Mar  9 12:26:47 2006
@@ -85,7 +85,6 @@
 
     protected Transport createTransport(URI location, WireFormat wf) throws UnknownHostException,
IOException {
         OpenWireFormat wireFormat = asOpenWireFormat(wf);
-        wireFormat.setSizePrefixDisabled(true);
         return new UdpTransport(wireFormat, location);
     }
 
@@ -113,7 +112,7 @@
         transport = new WireFormatNegotiator(transport, asOpenWireFormat(format), udpTransport.getMinmumWireFormatVersion())
{
             protected void onWireFormatNegotiated(WireFormatInfo info) {
                 // lets switch to the targetAddress that the last packet was
-                // received as
+                // received as so that all future requests go to the newly created UDP channel
                 udpTransport.useLastInboundDatagramAsNewTarget();
             }
         };
@@ -122,8 +121,6 @@
 
     protected OpenWireFormat asOpenWireFormat(WireFormat wf) {
         OpenWireFormat answer = (OpenWireFormat) wf;
-        answer.setSizePrefixDisabled(true);
-        answer.setCacheEnabled(false);
         return answer;
     }
 }

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransportServer.java
Thu Mar  9 12:26:47 2006
@@ -24,7 +24,6 @@
 import org.apache.activemq.transport.ResponseCorrelator;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportListener;
-import org.apache.activemq.transport.TransportLogger;
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.TransportServerSupport;
 import org.apache.activemq.transport.WireFormatNegotiator;
@@ -55,6 +54,10 @@
         super(connectURI);
         this.serverTransport = serverTransport;
         this.configuredTransport = configuredTransport;
+
+        // lets disable the incremental checking of the sequence numbers
+        // as we are getting messages from many different clients
+        serverTransport.setCheckSequenceNumbers(false);
     }
 
     public String toString() {
@@ -96,12 +99,16 @@
 
     public void process(Command command, DatagramHeader header) throws IOException {
         SocketAddress address = header.getFromAddress();
-        System.out.println(toString() + " received command: " + command + " from address:
" + address);
+        if (log.isDebugEnabled()) {
+            log.debug("Received command on: " + this + " from address: " + address + " command:
" + command);
+        }
         Transport transport = null;
         synchronized (transports) {
             transport = (Transport) transports.get(address);
             if (transport == null) {
-                System.out.println("###Êcreating new server connector");
+                if (log.isDebugEnabled()) {
+                    log.debug("Creating a new UDP server connection");
+                }
                 transport = createTransport(command, header);
                 transport = configureTransport(transport);
                 transports.put(address, transport);
@@ -114,23 +121,30 @@
 
     protected Transport configureTransport(Transport transport) {
         transport = new ResponseCorrelator(transport);
-        
-        // TODO
-        //transport = new InactivityMonitor(transport, serverTransport.getMaxInactivityDuration());
 
+        if (serverTransport.getMaxInactivityDuration() > 0) {
+            transport = new InactivityMonitor(transport, serverTransport.getMaxInactivityDuration());
+        }
+        
         getAcceptListener().onAccept(transport);
         return transport;
     }
 
-    protected Transport createTransport(Command command, DatagramHeader header) throws IOException
{
+    protected Transport createTransport(final Command command, DatagramHeader header) throws
IOException {
         final SocketAddress address = header.getFromAddress();
-        // TODO lets copy the wireformat...
-        final UdpTransport transport = new UdpTransport(serverTransport.getWireFormat(),
address);
-        
-        // lets send the packet into the transport so it can track packets
-        transport.doConsume(command, header);
+        final OpenWireFormat connectionWireFormat = serverTransport.getWireFormat().copy();
+        final UdpTransport transport = new UdpTransport(connectionWireFormat, address);
+
+        transport.receivedHeader(header);
+
+        return new WireFormatNegotiator(transport, transport.getWireFormat(), serverTransport.getMinmumWireFormatVersion())
{
 
-        return new WireFormatNegotiator(transport, serverTransport.getWireFormat(), serverTransport.getMinmumWireFormatVersion())
{
+            public void start() throws Exception {
+                super.start();
+
+                // process the inbound wireformat
+                onCommand(command);
+            }
 
             // lets use the specific addressing of wire format
             protected void sendWireFormat(WireFormatInfo info) throws IOException {

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/DatagramReplayStrategy.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/DatagramReplayStrategy.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/DatagramReplayStrategy.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/DatagramReplayStrategy.java
Thu Mar  9 12:26:47 2006
@@ -25,8 +25,9 @@
  */
 public interface DatagramReplayStrategy {
 
-    void onDroppedPackets(long expectedCounter, long actualCounter) throws IOException;
+    void onDroppedPackets(String name, long expectedCounter, long actualCounter) throws IOException;
 
-    void onReceivedPacket(long expectedCounter);
+    void onReceivedPacket(String name, long expectedCounter);
 
 }
+

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/ExceptionIfDroppedPacketStrategy.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/ExceptionIfDroppedPacketStrategy.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/ExceptionIfDroppedPacketStrategy.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/replay/ExceptionIfDroppedPacketStrategy.java
Thu Mar  9 12:26:47 2006
@@ -25,12 +25,12 @@
  */
 public class ExceptionIfDroppedPacketStrategy implements DatagramReplayStrategy {
 
-    public void onDroppedPackets(long expectedCounter, long actualCounter) throws IOException
{
+    public void onDroppedPackets(String name, long expectedCounter, long actualCounter) throws
IOException {
         long count = actualCounter - expectedCounter;
-        throw new IOException("" + count +  " packet(s) dropped. Expected: " + expectedCounter
+ " but was: " + actualCounter);
+        throw new IOException(name + count +  " packet(s) dropped. Expected: " + expectedCounter
+ " but was: " + actualCounter);
     }
 
-    public void onReceivedPacket(long expectedCounter) {
+    public void onReceivedPacket(String name, long expectedCounter) {
     }
 
 }

Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTestSupport.java
Thu Mar  9 12:26:47 2006
@@ -157,7 +157,7 @@
     protected Command assertCommandReceived() throws InterruptedException {
         Command answer = null;
         synchronized (lock) {
-            lock.wait(5000);
+            lock.wait(1000);
             answer = receivedCommand;
         }
 

Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java?rev=384603&r1=384602&r2=384603&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/udp/UdpTransportUsingServerTest.java
Thu Mar  9 12:26:47 2006
@@ -16,7 +16,6 @@
  */
 package org.apache.activemq.transport.udp;
 
-import org.apache.activemq.openwire.OpenWireFormat;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportFactory;
 import org.apache.activemq.transport.TransportServer;
@@ -35,7 +34,8 @@
 
     protected Transport createProducer() throws Exception {
         System.out.println("Producer using URI: " + producerURI);
-        return TransportFactory.connect(new URI(producerURI));
+        URI uri = new URI(producerURI);
+        return TransportFactory.connect(uri);
     }
 
     protected TransportServer createServer() throws Exception {
@@ -45,12 +45,4 @@
     protected Transport createConsumer() throws Exception {
         return null;
     }
-
-    protected OpenWireFormat createWireFormat() {
-        OpenWireFormat answer = new OpenWireFormat();
-        answer.setCacheEnabled(false);
-        answer.setSizePrefixDisabled(true);
-        return answer;
-    }
-
 }



Mime
View raw message