Cleaned up UDP transceivers, fixes for bug 4223 and 4320

9 files changed, 269 insertions, 109 deletions

diff --git a/java/src/Ice/ConnectionI.java b/java/src/Ice/ConnectionI.java
index 7fb693abf3c..5f107e2da6c 100644
--- a/java/src/Ice/ConnectionI.java
+++ b/java/src/Ice/ConnectionI.java
@@ -291,6 +291,12 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
         assert(_state > StateNotValidated);
         assert(_state < StateClosing);
 
+        //
+        // Ensure the message isn't bigger than what we can send with the
+        // transport.
+        //
+        _transceiver.checkSendSize(os.getBuffer(), _instance.messageSizeMax());
+
         if(response)
         {
             //
@@ -358,6 +364,12 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
         assert(_state > StateNotValidated);
         assert(_state < StateClosing);
 
+        //
+        // Ensure the message isn't bigger than what we can send with the
+        // transport.
+        //
+        _transceiver.checkSendSize(os.getBuffer(), _instance.messageSizeMax());
+
         if(response)
         {
             //
@@ -880,13 +892,14 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
                 }
                 if((current.operation & IceInternal.SocketOperation.Read) != 0 && !_readStream.isEmpty())
                 {
-                    if(_readStream.size() == IceInternal.Protocol.headerSize) // Read header.
+                    if(_readHeader) // Read header if necessary.
                     {
                         if(!_transceiver.read(_readStream.getBuffer(), _hasMoreData))
                         {
                             return;
                         }
                         assert(!_readStream.getBuffer().b.hasRemaining());
+                        _readHeader = false;
 
                         int pos = _readStream.pos();
                         if(pos < IceInternal.Protocol.headerSize)
@@ -903,8 +916,8 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
                         m[1] = _readStream.readByte();
                         m[2] = _readStream.readByte();
                         m[3] = _readStream.readByte();
-                        if(m[0] != IceInternal.Protocol.magic[0] || m[1] != IceInternal.Protocol.magic[1]
-                           || m[2] != IceInternal.Protocol.magic[2] || m[3] != IceInternal.Protocol.magic[3])
+                        if(m[0] != IceInternal.Protocol.magic[0] || m[1] != IceInternal.Protocol.magic[1] ||
+                           m[2] != IceInternal.Protocol.magic[2] || m[3] != IceInternal.Protocol.magic[3])
                         {
                             Ice.BadMagicException ex = new Ice.BadMagicException();
                             ex.badMagic = m;
@@ -957,12 +970,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
                     {
                         if(_endpoint.datagram())
                         {
-                            if(_warnUdp)
-                            {
-                                _logger.warning("DatagramLimitException: maximum size of " + _readStream.pos() +
-                                                " exceeded");
-                            }
-                            throw new Ice.DatagramLimitException();
+                            throw new Ice.DatagramLimitException(); // The message was truncated.
                         }
                         else
                         {
@@ -1015,8 +1023,13 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
             }
             catch(DatagramLimitException ex) // Expected.
             {
+                if(_warnUdp)
+                {
+                    _logger.warning("maximum datagram size of " + _readStream.pos() + " exceeded");
+                }
                 _readStream.resize(IceInternal.Protocol.headerSize, true);
                 _readStream.pos(0);
+                _readHeader = true;
                 return;
             }
             catch(SocketException ex)
@@ -1035,6 +1048,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
                     }
                     _readStream.resize(IceInternal.Protocol.headerSize, true);
                     _readStream.pos(0);
+                    _readHeader = true;
                 }
                 else
                 {
@@ -1284,6 +1298,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
         _batchRequestCompress = false;
         _batchMarker = 0;
         _readStream = new IceInternal.BasicStream(instance);
+        _readHeader = false;
         _writeStream = new IceInternal.BasicStream(instance);
         _dispatchCount = 0;
         _state = StateNotInitialized;
@@ -1732,6 +1747,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
         _writeStream.pos(0);
 
         _readStream.resize(IceInternal.Protocol.headerSize, true);
+        _readHeader = true;
         _readStream.pos(0);
 
         return true;
@@ -1950,6 +1966,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
         _readStream.swap(info.stream);
         _readStream.resize(IceInternal.Protocol.headerSize, true);
         _readStream.pos(0);
+        _readHeader = true;
 
         assert(info.stream.pos() == info.stream.size());
 
@@ -2490,6 +2507,7 @@ public final class ConnectionI extends IceInternal.EventHandler implements Conne
     private java.util.LinkedList<OutgoingMessage> _sendStreams = new java.util.LinkedList<OutgoingMessage>();
 
     private IceInternal.BasicStream _readStream;
+    private boolean _readHeader;
     private IceInternal.BasicStream _writeStream;
 
     private int _dispatchCount;
diff --git a/java/src/IceInternal/ReferenceFactory.java b/java/src/IceInternal/ReferenceFactory.java
index d329b6249a0..30a236394ec 100644
--- a/java/src/IceInternal/ReferenceFactory.java
+++ b/java/src/IceInternal/ReferenceFactory.java
@@ -44,13 +44,14 @@ public final class ReferenceFactory
         //
         // Create new reference
         //
-        FixedReference ref = new FixedReference(_instance, 
-                                                _communicator, 
-                                                ident, 
-                                                "", // Facet
-                                                Reference.ModeTwoway,
-                                                false,
-                                                fixedConnection);
+        FixedReference ref = new FixedReference(
+            _instance, 
+            _communicator, 
+            ident, 
+            "", // Facet
+            fixedConnection.endpoint().datagram() ? Reference.ModeDatagram : Reference.ModeTwoway,
+            fixedConnection.endpoint().secure(),
+            fixedConnection);
         return updateCache(ref);
     }
 
diff --git a/java/src/IceInternal/UdpTransceiver.java b/java/src/IceInternal/UdpTransceiver.java
index bd18ad77678..2001f8e0e67 100644
--- a/java/src/IceInternal/UdpTransceiver.java
+++ b/java/src/IceInternal/UdpTransceiver.java
@@ -32,7 +32,7 @@ final class UdpTransceiver implements Transceiver
     {
         assert(_fd != null);
         
-        if(!_connect && _traceLevels.network >= 1)
+        if(_state >= StateConnected && _traceLevels.network >= 1)
         {
             String s = "closing udp connection\n" + toString();
             _logger.trace(_traceLevels.networkCat, s);
@@ -52,40 +52,28 @@ final class UdpTransceiver implements Transceiver
     write(Buffer buf)
     {
         assert(buf.b.position() == 0);
-        final int packetSize = java.lang.Math.min(_maxPacketSize, _sndSize - _udpOverhead);
-        if(packetSize < buf.size())
-        {
-            //
-            // We don't log a warning here because the client gets an exception anyway.
-            //
-            throw new Ice.DatagramLimitException();
-        }
+        assert(_fd != null && _state >= StateConnected);
+                
+        // The caller is supposed to check the send size before by calling checkSendSize
+        assert(java.lang.Math.min(_maxPacketSize, _sndSize - _udpOverhead) >= buf.size());
 
-        while(buf.b.hasRemaining())
+        int ret = 0;
+        while(true)
         {
             try
             {
-                assert(_fd != null);
-
-                int ret = _fd.write(buf.b);
-                
-                if(ret == 0)
+                if(_state == StateConnected)
                 {
-                    return false;
-                }
-
-                if(_traceLevels.network >= 3)
-                {
-                    String s = "sent " + ret + " bytes via udp\n" + toString();
-                    _logger.trace(_traceLevels.networkCat, s);
+                    ret = _fd.write(buf.b);
                 }
-
-                if(_stats != null)
+                else
                 {
-                    _stats.bytesSent(type(), ret);
+                    if(_peerAddr == null)
+                    {
+                        throw new Ice.SocketException(); // No peer has sent a datagram yet.
+                    }
+                    ret = _fd.send(buf.b, _peerAddr);
                 }
-
-                assert(ret == buf.b.limit());
                 break;
             }
             catch(java.nio.channels.AsynchronousCloseException ex)
@@ -112,6 +100,23 @@ final class UdpTransceiver implements Transceiver
             }
         }
 
+        if(ret == 0)
+        {
+            return false;
+        }
+
+        if(_traceLevels.network >= 3)
+        {
+            String s = "sent " + ret + " bytes via udp\n" + toString();
+            _logger.trace(_traceLevels.networkCat, s);
+        }
+
+        if(_stats != null)
+        {
+            _stats.bytesSent(type(), ret);
+        }
+
+        assert(ret == buf.b.limit());
         return true;
     }
 
@@ -122,18 +127,6 @@ final class UdpTransceiver implements Transceiver
         moreData.value = false;
 
         final int packetSize = java.lang.Math.min(_maxPacketSize, _rcvSize - _udpOverhead);
-        if(packetSize < buf.size())
-        {
-            //
-            // We log a warning here because this is the server side -- without the
-            // the warning, there would only be silence.
-            //
-            if(_warn)
-            {
-                _logger.warning("DatagramLimitException: maximum size of " + packetSize + " exceeded");
-            }
-            throw new Ice.DatagramLimitException();
-        }
         buf.resize(packetSize, true);
         buf.b.position(0);
 
@@ -142,31 +135,14 @@ final class UdpTransceiver implements Transceiver
         {
             try
             {
-                java.net.InetSocketAddress sender = (java.net.InetSocketAddress)_fd.receive(buf.b);
-
-                if(sender == null || buf.b.position() == 0)
+                java.net.SocketAddress peerAddr = _fd.receive(buf.b);
+                if(peerAddr == null || buf.b.position() == 0)
                 {
                     return false;
                 }
 
+                _peerAddr = (java.net.InetSocketAddress)peerAddr;
                 ret = buf.b.position();
-
-                if(_connect)
-                {
-                    //
-                    // If we must connect, then we connect to the first peer that
-                    // sends us a packet.
-                    //
-                    Network.doConnect(_fd, sender);
-                    _connect = false; // We're connected now
-
-                    if(_traceLevels.network >= 1)
-                    {
-                        String s = "connected udp socket\n" + toString();
-                        _logger.trace(_traceLevels.networkCat, s);
-                    }
-                }
-
                 break;
             }
             catch(java.nio.channels.AsynchronousCloseException ex)
@@ -193,6 +169,21 @@ final class UdpTransceiver implements Transceiver
             }
         }
 
+        if(_state == StateNeedConnect)
+        {
+            //
+            // If we must connect, we connect to the first peer that sends us a packet.
+            //
+            Network.doConnect(_fd, _peerAddr);
+            _state = StateConnected;
+
+            if(_traceLevels.network >= 1)
+            {
+                String s = "connected udp socket\n" + toString();
+                _logger.trace(_traceLevels.networkCat, s);
+            }
+        }
+
         if(_traceLevels.network >= 3)
         {
             String s = "received " + ret + " bytes via udp\n" + toString();
@@ -219,14 +210,31 @@ final class UdpTransceiver implements Transceiver
     public String
     toString()
     {
-        if(_mcastAddr != null && _fd != null)
+        if(_fd == null)
         {
-            return Network.fdToString(_fd) + "\nmulticast address = " + Network.addrToString(_mcastAddr);
+            return "<closed>";
+        }
+
+        String s;
+        if(_state == StateNotConnected)
+        {
+            java.net.DatagramSocket socket = ((java.nio.channels.DatagramChannel)_fd).socket();
+            s = "local address = " + Network.addrToString((java.net.InetSocketAddress)socket.getLocalSocketAddress());
+            if(_peerAddr != null)
+            {
+                s += "\nremote address = " + Network.addrToString(_peerAddr);
+            }
         }
         else
         {
-            return Network.fdToString(_fd);
+            s = Network.fdToString(_fd);
+        }
+
+        if(_mcastAddr != null)
+        {
+            s += "\nmulticast address = " + Network.addrToString(_mcastAddr);
         }
+        return s;
     }
 
     public Ice.ConnectionInfo
@@ -238,15 +246,31 @@ final class UdpTransceiver implements Transceiver
         java.net.DatagramSocket socket = _fd.socket();
         info.localAddress = socket.getLocalAddress().getHostAddress();
         info.localPort = socket.getLocalPort();
-        if(socket.getInetAddress() != null)
+        if(_state == StateNotConnected)
         {
-            info.remoteAddress = socket.getInetAddress().getHostAddress();
-            info.remotePort = socket.getPort();
+            if(_peerAddr != null)
+            {
+                info.remoteAddress = _peerAddr.getAddress().getHostAddress();
+                info.remotePort = _peerAddr.getPort();
+            }
+            else
+            {
+                info.remoteAddress = "";
+                info.remotePort = -1;
+            }
         }
         else
         {
-            info.remoteAddress = "";
-            info.remotePort = -1;
+            if(socket.getInetAddress() != null)
+            {
+                info.remoteAddress = socket.getInetAddress().getHostAddress();
+                info.remotePort = socket.getPort();
+            }
+            else
+            {
+                info.remoteAddress = "";
+                info.remotePort = -1;
+            }
         }
         if(_mcastAddr != null)
         {
@@ -268,6 +292,11 @@ final class UdpTransceiver implements Transceiver
         {
             Ex.throwMemoryLimitException(buf.size(), messageSizeMax);
         }
+        
+        //
+        // The maximum packetSize is either the maximum allowable UDP packet size, or 
+        // the UDP send buffer size (which ever is smaller).
+        //
         final int packetSize = java.lang.Math.min(_maxPacketSize, _sndSize - _udpOverhead);
         if(packetSize < buf.size())
         {
@@ -289,8 +318,7 @@ final class UdpTransceiver implements Transceiver
         _traceLevels = instance.traceLevels();
         _logger = instance.initializationData().logger;
         _stats = instance.initializationData().stats;
-        _connect = true;
-        _warn = instance.initializationData().properties.getPropertyAsInt("Ice.Warn.Datagrams") > 0;
+        _state = StateNeedConnect;
         _addr = addr;
 
         try
@@ -299,7 +327,7 @@ final class UdpTransceiver implements Transceiver
             setBufSize(instance);
             Network.setBlock(_fd, false);
             Network.doConnect(_fd, _addr);
-            _connect = false; // We're connected now
+            _state = StateConnected; // We're connected now
             if(_addr.getAddress().isMulticastAddress())
             {
                 configureMulticast(null, mcastInterface, mcastTtl);
@@ -326,8 +354,7 @@ final class UdpTransceiver implements Transceiver
         _traceLevels = instance.traceLevels();
         _logger = instance.initializationData().logger;
         _stats = instance.initializationData().stats;
-        _connect = connect;
-        _warn = instance.initializationData().properties.getPropertyAsInt("Ice.Warn.Datagrams") > 0;
+        _state = connect ? StateNeedConnect : StateNotConnected;
 
         try
         {
@@ -344,7 +371,20 @@ final class UdpTransceiver implements Transceiver
             {
                 Network.setReuseAddress(_fd, true);
                 _mcastAddr = _addr;
-                _addr = Network.doBind(_fd, Network.getAddress("0.0.0.0", port, Network.EnableIPv4));
+                if(System.getProperty("os.name").startsWith("Windows"))
+                {
+                    //
+                    // Windows does not allow binding to the mcast address itself
+                    // so we bind to INADDR_ANY (0.0.0.0) instead. As a result,
+                    // bi-directional connection won't work because the source 
+                    // address won't be the multicast address and the client will
+                    // therefore reject the datagram.
+                    //
+                    int protocol = 
+                        _mcastAddr.getAddress().getAddress().length == 4 ? Network.EnableIPv4 : Network.EnableIPv6;
+                    _addr = Network.getAddressForServer("", port, protocol);
+                }
+                _addr = Network.doBind(_fd, _addr);
                 if(port == 0)
                 {
                     _mcastAddr = new java.net.InetSocketAddress(_mcastAddr.getAddress(), _addr.getPort());
@@ -425,7 +465,7 @@ final class UdpTransceiver implements Transceiver
             // Get property for buffer size and check for sanity.
             //
             int sizeRequested = instance.initializationData().properties.getPropertyAsIntWithDefault(prop, dfltSize);
-            if(sizeRequested < _udpOverhead)
+            if(sizeRequested < (_udpOverhead + IceInternal.Protocol.headerSize))
             {
                 _logger.warning("Invalid " + prop + " value of " + sizeRequested + " adjusted to " + dfltSize);
                 sizeRequested = dfltSize;
@@ -570,13 +610,13 @@ final class UdpTransceiver implements Transceiver
     private TraceLevels _traceLevels;
     private Ice.Logger _logger;
     private Ice.Stats _stats;
-    private boolean _connect;
-    private final boolean _warn;
+    private int _state;
     private int _rcvSize;
     private int _sndSize;
     private java.nio.channels.DatagramChannel _fd;
     private java.net.InetSocketAddress _addr;
     private java.net.InetSocketAddress _mcastAddr = null;
+    private java.net.InetSocketAddress _peerAddr = null;
 
     //
     // The maximum IP datagram size is 65535. Subtract 20 bytes for the IP header and 8 bytes for the UDP header
@@ -584,4 +624,8 @@ final class UdpTransceiver implements Transceiver
     //
     private final static int _udpOverhead = 20 + 8;
     private final static int _maxPacketSize = 65535 - _udpOverhead;
+
+    private static final int StateNeedConnect = 0;
+    private static final int StateConnected = 1;
+    private static final int StateNotConnected = 2;
 }
diff --git a/java/test/Ice/info/AllTests.java b/java/test/Ice/info/AllTests.java
index e5750cf3b62..9dd3650f2a8 100644
--- a/java/test/Ice/info/AllTests.java
+++ b/java/test/Ice/info/AllTests.java
@@ -144,6 +144,7 @@ public class AllTests
             Ice.IPConnectionInfo info = (Ice.IPConnectionInfo)base.ice_getConnection().getInfo();
             test(!info.incoming);
             test(info.adapterName.length() == 0);
+            test(info.localPort > 0);
             test(info.remotePort == 12010);
             test(info.remoteAddress.equals(defaultHost));
             test(info.localAddress.equals(defaultHost));
@@ -155,6 +156,14 @@ public class AllTests
             test(ctx.get("localAddress").equals(info.remoteAddress));
             test(ctx.get("remotePort").equals(Integer.toString(info.localPort)));
             test(ctx.get("localPort").equals(Integer.toString(info.remotePort)));
+
+            info = (Ice.IPConnectionInfo)base.ice_datagram().ice_getConnection().getInfo();
+            test(!info.incoming);
+            test(info.adapterName.length() == 0);
+            test(info.localPort > 0);
+            test(info.remotePort == 12010);
+            test(info.remoteAddress.equals(defaultHost));
+            test(info.localAddress.equals(defaultHost));
         }
         out.println("ok");
 
diff --git a/java/test/Ice/udp/AllTests.java b/java/test/Ice/udp/AllTests.java
index 725003dbe31..22550b6f370 100644
--- a/java/test/Ice/udp/AllTests.java
+++ b/java/test/Ice/udp/AllTests.java
@@ -78,7 +78,7 @@ public class AllTests
 
         System.out.print("testing udp... ");
         System.out.flush();
-        Ice.ObjectPrx base = communicator.stringToProxy("test:udp -p 12010").ice_datagram();
+        Ice.ObjectPrx base = communicator.stringToProxy("test -d:udp -p 12010");
         TestIntfPrx obj = TestIntfPrxHelper.uncheckedCast(base);
 
         int nRetry = 5;
@@ -124,7 +124,7 @@ public class AllTests
             {
                 test(seq.length > 16384);
             }
-            
+            obj.ice_getConnection().close(false);
             communicator.getProperties().setProperty("Ice.UDP.SndSize", "64000");
             seq = new byte[50000];
             try
@@ -153,28 +153,85 @@ public class AllTests
         {
             host = "239.255.1.1";
         }
-        base = communicator.stringToProxy("test:udp -h " + host + " -p 12020").ice_datagram();
-        obj = TestIntfPrxHelper.uncheckedCast(base);
+        base = communicator.stringToProxy("test -d:udp -h " + host + " -p 12020");
+        TestIntfPrx objMcast = TestIntfPrxHelper.uncheckedCast(base);
 
-        replyI.reset();
-        obj.ping(reply);
-        if(!replyI.waitReply(5, 2000))
+        nRetry = 5;
+        while(nRetry-- > 0)
         {
-            System.out.println("failed (is a firewall enabled?)");
-            return obj;
+            replyI.reset();
+            objMcast.ping(reply);
+            ret = replyI.waitReply(5, 2000);
+            if(ret)
+            {
+                break; // Success
+            }
+            replyI = new PingReplyI();
+            reply = (PingReplyPrx)PingReplyPrxHelper.uncheckedCast(adapter.addWithUUID(replyI)).ice_datagram();
         }
-
-        replyI.reset();
-        obj.ping(reply);
-        ret = replyI.waitReply(5, 2000);
-        if(!replyI.waitReply(5, 2000))
+        if(!ret)
         {
             System.out.println("failed (is a firewall enabled?)");
-            return obj;
+        }
+        else
+        {
+            System.out.println("ok");
         }
 
-        System.out.println("ok");
+        System.out.print("testing udp bi-dir connection... ");
+        System.out.flush();
+        obj.ice_getConnection().setAdapter(adapter);
+        objMcast.ice_getConnection().setAdapter(adapter);
+        nRetry = 5;
+        while(nRetry-- > 0)
+        {
+            replyI.reset();
+            obj.pingBiDir(reply.ice_getIdentity());
+            obj.pingBiDir(reply.ice_getIdentity());
+            obj.pingBiDir(reply.ice_getIdentity());
+            ret = replyI.waitReply(3, 2000);
+            if(ret)
+            {
+                break; // Success
+            }
+            replyI = new PingReplyI();
+            reply = (PingReplyPrx)PingReplyPrxHelper.uncheckedCast(adapter.addWithUUID(replyI)).ice_datagram();
+        }
+        test(ret);
+        if(!System.getProperty("os.name").startsWith("Windows"))
+        {
+            //
+            // Windows doesn't support sending replies back on the multicast UDP connection,
+            // see UdpTransceiver constructor for the details.
+            // 
+            nRetry = 5;
+            while(nRetry-- > 0)
+            {
+                replyI.reset();
+                objMcast.pingBiDir(reply.ice_getIdentity());
+                ret = replyI.waitReply(5, 2000);
+                if(ret)
+                {
+                    break; // Success
+                }
+                replyI = new PingReplyI();
+                reply = (PingReplyPrx)PingReplyPrxHelper.uncheckedCast(adapter.addWithUUID(replyI)).ice_datagram();
+            }
 
-        return obj;
+            if(!ret)
+            {
+                System.out.println("failed (is a firewall enabled?)");
+            }
+            else
+            {
+                System.out.println("ok");
+            }
+        }
+        else
+        {
+            System.out.println("ok");
+        }
+        
+        return objMcast;
     }
 }
diff --git a/java/test/Ice/udp/Client.java b/java/test/Ice/udp/Client.java
index 3cab9099e6e..85611b9b3e1 100644
--- a/java/test/Ice/udp/Client.java
+++ b/java/test/Ice/udp/Client.java
@@ -26,6 +26,7 @@ public class Client extends test.Util.Application
         initData.properties = Ice.Util.createProperties(argsH);
         initData.properties.setProperty("Ice.Package.Test", "test.Ice.udp");
         initData.properties.setProperty("Ice.Warn.Connections", "0");
+        initData.properties.setProperty("Ice.UDP.RcvSize", "16384");
         initData.properties.setProperty("Ice.UDP.SndSize", "16384");
         return initData;
     }
diff --git a/java/test/Ice/udp/Server.java b/java/test/Ice/udp/Server.java
index 9ce5a7869c8..c0549751288 100644
--- a/java/test/Ice/udp/Server.java
+++ b/java/test/Ice/udp/Server.java
@@ -36,6 +36,7 @@ public class Server extends test.Util.Application
         initData.properties.setProperty("Ice.Package.Test", "test.Ice.udp");
         initData.properties.setProperty("Ice.Warn.Connections", "0");
         initData.properties.setProperty("Ice.UDP.RcvSize", "16384");
+        initData.properties.setProperty("Ice.UDP.SndSize", "16384");
 
         String host;
         if(initData.properties.getProperty("Ice.IPv6") == "1")
diff --git a/java/test/Ice/udp/Test.ice b/java/test/Ice/udp/Test.ice
index 4db62e371c3..aa60b97402c 100644
--- a/java/test/Ice/udp/Test.ice
+++ b/java/test/Ice/udp/Test.ice
@@ -10,6 +10,8 @@
 #ifndef TEST_ICE
 #define TEST_ICE
 
+#include <Ice/Identity.ice>
+
 [["java:package:test.Ice.udp"]]
 module Test
 {
@@ -25,6 +27,7 @@ interface TestIntf
 {
     void ping(PingReply* reply);
     void sendByteSeq(ByteSeq seq, PingReply* reply);
+    void pingBiDir(Ice::Identity id);
     void shutdown();
 };
 
diff --git a/java/test/Ice/udp/TestIntfI.java b/java/test/Ice/udp/TestIntfI.java
index 7865321300b..dfb0c231042 100644
--- a/java/test/Ice/udp/TestIntfI.java
+++ b/java/test/Ice/udp/TestIntfI.java
@@ -37,6 +37,32 @@ public final class TestIntfI extends _TestIntfDisp
         }
     }
 
+    public void pingBiDir(Ice.Identity id, Ice.Current current)
+    {
+        try
+        {
+            //
+            // Ensure sending too much data doesn't cause the UDP connection
+            // to be closed.
+            //
+            try
+            {
+                byte[] seq = new byte[32 * 1024];
+                TestIntfPrxHelper.uncheckedCast(current.con.createProxy(id)).sendByteSeq(seq, null);
+            }
+            catch(Ice.DatagramLimitException ex)
+            {
+                // Expected.
+            }
+
+            PingReplyPrxHelper.uncheckedCast(current.con.createProxy(id)).reply();
+        }
+        catch(Ice.LocalException ex)
+        {
+            assert(false);
+        }
+    }
+
     public void shutdown(Ice.Current current)
     {
         current.adapter.getCommunicator().shutdown();