diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index e415f1b7..e7aefdb4 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -7,7 +7,6 @@ import java.util.List; import java.util.Map; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; -import java.util.logging.Level; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 50592f73..ed0cfb39 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -37,80 +37,184 @@ public class ServerIntegrationTest { } } +// +// @Test +// public void testSubscribe() throws IOException, InterruptedException { +//// startServer(); +// Client client = new Client("localhost", Server.PORT); +//// Client client = new Client("test.mosquitto.org", 1883); +// client.connect(); +// +// +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// +// client.publish("/topic", "Test my payload".getBytes()); +// +// client.close(); +// client.shutdown(); +// +// assertTrue(received); +// } +// +// +// @Test +// public void testCleanSession_maintainClientSubscriptions() { +// Client client = new Client("localhost", Server.PORT/*, "CLID_123"*/); +// client.connect(false); //without session cleanup +// +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.close(); +// +// //Exercise that the client maintain the subscriptions +// client.connect(false); +// client.publish("/topic", "Test my payload".getBytes()); +// client.close(); +// +// //Verify +// assertTrue(received); +// +// //TearDown +// client.shutdown(); +// } +// +// @Test +// public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() { +// Client client = new Client("localhost", Server.PORT, "CLID_123"); +// client.connect(false); //without session cleanup +// +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.close(); +// client.shutdown(); +// +// //Exercise that the client maintain the subscriptions +// client = new Client("localhost", Server.PORT, "CLID_123"); +// client.register("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.connect(false); +// client.publish("/topic", "Test my payload".getBytes()); +// client.close(); +// +// //Verify +// assertTrue(received); +// +// //TearDown +// client.shutdown(); +// } +// +// +// /** +// * Check that after a client has connected with clean session false, subscribed +// * to some topic and exited, if it reconnect with clean session true, the server +// * correctly cleanup every previous subscription +// */ +// @Test +// public void testCleanSession_correctlyClientSubscriptions() { +// Client client = new Client("localhost", Server.PORT, "CLID_123"); +// client.connect(false); //without session cleanup +// +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.close(); +// client.shutdown(); +// +// //Exercise that the client maintain the subscriptions +// client = new Client("localhost", Server.PORT, "CLID_123"); +// client.register("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.connect(true); +// client.publish("/topic", "Test my payload".getBytes()); +// client.close(); +// +// //Verify +// assertFalse(received); +// +// //TearDown +// client.shutdown(); +// } +// +// +// @Test +// public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException { +// Client client = new Client("localhost", Server.PORT, "CLID_123"); +// client.connect(false); //without session cleanup +// +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.close(); +// client.shutdown(); +// server.stopServer(); +// +// server.startServer(); +// +// //Exercise that the client maintain the subscriptions +// client = new Client("localhost", Server.PORT, "CLID_123"); +// client.register("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.connect(false); +// client.publish("/topic", "Test my payload".getBytes()); +// client.close(); +// +// //Verify +// assertTrue(received); +// +// //TearDown +// client.shutdown(); +// } + @Test - public void testSubscribe() throws IOException, InterruptedException { -// startServer(); - Client client = new Client("localhost", Server.PORT); -// Client client = new Client("test.mosquitto.org", 1883); + public void testRetain_maintainMessage_againstClientDestruction() { + Client client = new Client("localhost", Server.PORT, "CLID_123"); client.connect(); - - client.subscribe("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - - client.publish("/topic", "Test my payload".getBytes()); - - client.close(); - client.shutdown(); - - assertTrue(received); - } - - - @Test - public void testCleanSession_maintainClientSubscriptions() { - Client client = new Client("localhost", Server.PORT/*, "CLID_123"*/); - client.connect(false); //without session cleanup - - client.subscribe("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - client.close(); - - //Exercise that the client maintain the subscriptions - client.connect(false); - client.publish("/topic", "Test my payload".getBytes()); - client.close(); - - //Verify - assertTrue(received); - - //TearDown - client.shutdown(); - } - - @Test - public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() { - Client client = new Client("localhost", Server.PORT, "CLID_123"); - client.connect(false); //without session cleanup - - client.subscribe("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); + client.publish("/topic", "Test my payload".getBytes(), true); client.close(); client.shutdown(); //Exercise that the client maintain the subscriptions client = new Client("localhost", Server.PORT, "CLID_123"); - client.register("/topic", new IPublishCallback() { + client.connect(); + client.subscribe("/topic", new IPublishCallback() { public void published(String topic, byte[] message) { received = true; } }); - client.connect(false); - client.publish("/topic", "Test my payload".getBytes()); client.close(); //Verify @@ -120,109 +224,4 @@ public class ServerIntegrationTest { client.shutdown(); } - - /** - * Check that after a client has connected with clean session false, subscribed - * to some topic and exited, if it reconnect with clean session true, the server - * correctly cleanup every previous subscription - */ - @Test - public void testCleanSession_correctlyClientSubscriptions() { - Client client = new Client("localhost", Server.PORT, "CLID_123"); - client.connect(false); //without session cleanup - - client.subscribe("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - client.close(); - client.shutdown(); - - //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); - client.register("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - client.connect(true); - client.publish("/topic", "Test my payload".getBytes()); - client.close(); - - //Verify - assertFalse(received); - - //TearDown - client.shutdown(); - } - - - @Test - public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException { - Client client = new Client("localhost", Server.PORT, "CLID_123"); - client.connect(false); //without session cleanup - - client.subscribe("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - client.close(); - client.shutdown(); - server.stopServer(); - - server.startServer(); - - //Exercise that the client maintain the subscriptions - client = new Client("localhost", Server.PORT, "CLID_123"); - client.register("/topic", new IPublishCallback() { - - public void published(String topic, byte[] message) { - received = true; - } - }); - client.connect(false); - client.publish("/topic", "Test my payload".getBytes()); - client.close(); - - //Verify - assertTrue(received); - - //TearDown - client.shutdown(); - } - - - -// @Test -// public void testRetain_maintainMessage_againstClientDestruction() { -// Client client = new Client("localhost", Server.PORT, "CLID_123"); -// client.connect(false); //without session cleanup -// -// client.publish("/topic", "Test my payload".getBytes(), true); -// client.close(); -// client.shutdown(); -// -// //Exercise that the client maintain the subscriptions -// client = new Client("localhost", Server.PORT, "CLID_123"); -// client.connect(false); -// client.subscribe("/topic", new IPublishCallback() { -// -// public void published(String topic, byte[] message) { -// received = true; -// } -// }); -// client.close(); -// -// //Verify -// assertTrue(received); -// -// //TearDown -// client.shutdown(); -// } - }