Merge pull request #1470 from vitalidze/master

UDP stops listening after failure

2 files changed, 131 insertions, 3 deletions

diff --git a/src/org/traccar/MainEventHandler.java b/src/org/traccar/MainEventHandler.java
index 8b94a68c9..81376724b 100644
--- a/src/org/traccar/MainEventHandler.java
+++ b/src/org/traccar/MainEventHandler.java
@@ -20,6 +20,7 @@ import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelStateEvent;
 import org.jboss.netty.channel.ExceptionEvent;
 import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.socket.DatagramChannel;
 import org.jboss.netty.handler.timeout.IdleStateAwareChannelHandler;
 import org.jboss.netty.handler.timeout.IdleStateEvent;
 import org.traccar.helper.Log;
@@ -61,7 +62,7 @@ public class MainEventHandler extends IdleStateAwareChannelHandler {
     @Override
     public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
         Log.info(formatChannel(e.getChannel()) + " disconnected");
-        e.getChannel().close();
+        closeChannel(e.getChannel());
 
         Context.getConnectionManager().removeActiveDevice(e.getChannel());
     }
@@ -69,13 +70,18 @@ public class MainEventHandler extends IdleStateAwareChannelHandler {
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
         Log.warning(formatChannel(e.getChannel()) + " error", e.getCause());
-        e.getChannel().close();
+        closeChannel(e.getChannel());
     }
 
     @Override
     public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) {
         Log.info(formatChannel(e.getChannel()) + " timed out");
-        e.getChannel().close();
+        closeChannel(e.getChannel());
     }
 
+    private void closeChannel(Channel channel) {
+        if (!(channel instanceof DatagramChannel)) {
+            channel.close();
+        }
+    }
 }
diff --git a/test/org/traccar/ChannelClosingTest.java b/test/org/traccar/ChannelClosingTest.java
new file mode 100644
index 000000000..0b138235f
--- /dev/null
+++ b/test/org/traccar/ChannelClosingTest.java
@@ -0,0 +1,122 @@
+package org.traccar;
+
+import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.*;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.traccar.database.IdentityManager;
+import org.traccar.helper.ChannelBufferTools;
+import org.traccar.model.Device;
+import org.traccar.protocol.GatorProtocol;
+import org.traccar.protocol.GatorProtocolDecoder;
+
+import java.net.*;
+import java.util.concurrent.CyclicBarrier;
+
+public class ChannelClosingTest {
+
+    @BeforeClass
+    public static void init() {
+        Context.init(new IdentityManager() {
+
+            private Device createDevice() {
+                return null;
+            }
+
+            @Override
+            public Device getDeviceById(long id) {
+                return createDevice();
+            }
+
+            @Override
+            public Device getDeviceByUniqueId(String imei) {
+                return createDevice();
+            }
+
+        });
+    }
+
+    static class ExceptionWaiter implements ChannelUpstreamHandler {
+        final CyclicBarrier barrier = new CyclicBarrier(2);
+        Channel channel;
+
+        @Override
+        public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
+            ctx.sendUpstream(e);
+            if (e instanceof ExceptionEvent) {
+                channel = e.getChannel();
+                barrier.await();
+            }
+        }
+
+        void waitFor() throws Exception {
+            barrier.await();
+        }
+
+        Channel channel() {
+            return channel;
+        }
+    }
+
+    @Test
+    public void testUDP() throws Exception {
+        final ExceptionWaiter exception = new ExceptionWaiter();
+
+        TrackerServer udpServer = new TrackerServer(new ConnectionlessBootstrap(), "gator") {
+            @Override
+            protected void addSpecificHandlers(ChannelPipeline pipeline) {
+                pipeline.addLast("objectDecoder", new GatorProtocolDecoder(new GatorProtocol()));
+                pipeline.addLast("exceptionWaiter", exception);
+            }
+        };
+        final int PORT = 50522;
+        udpServer.setPort(PORT);
+        try {
+            udpServer.start();
+            Assert.assertFalse(udpServer.getChannelGroup().isEmpty());
+            try (DatagramSocket socket = new DatagramSocket()) {
+                socket.connect(InetAddress.getLocalHost(), PORT);
+                byte[] data = ChannelBufferTools.convertHexString("242400");
+                socket.send(new DatagramPacket(data, data.length));
+            }
+            exception.waitFor();
+            Assert.assertFalse(udpServer.getChannelGroup().isEmpty());
+            Assert.assertTrue(exception.channel().isBound());
+            Assert.assertTrue(exception.channel().isOpen());
+        } finally {
+            udpServer.stop();
+        }
+    }
+
+    @Test
+    public void testTCP() throws Exception {
+        final ExceptionWaiter exception = new ExceptionWaiter();
+        TrackerServer tcpServer = new TrackerServer(new ServerBootstrap(), "gator") {
+            @Override
+            protected void addSpecificHandlers(ChannelPipeline pipeline) {
+                pipeline.addLast("objectDecoder", new GatorProtocolDecoder(new GatorProtocol()));
+                pipeline.addLast("exceptionWaiter", exception);
+            }
+        };
+        final int PORT = 50522;
+        tcpServer.setPort(PORT);
+        try {
+            tcpServer.start();
+            try (Socket socket = new Socket()) {
+                socket.connect(new InetSocketAddress(InetAddress.getLocalHost(), PORT));
+
+                byte[] data = ChannelBufferTools.convertHexString("242400");
+                socket.getOutputStream().write(data);
+
+                exception.waitFor();
+                Assert.assertFalse(exception.channel().isBound());
+                Assert.assertFalse(exception.channel().isOpen());
+                Assert.assertFalse(exception.channel().isConnected());
+            }
+        } finally {
+            tcpServer.stop();
+        }
+    }
+}