diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java new file mode 100644 index 00000000..fdde9316 --- /dev/null +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java @@ -0,0 +1,248 @@ +/* + * Copyright (c) 2012-2018 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ + +package io.moquette.server; + +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import io.moquette.spi.impl.security.AcceptAllAuthenticator; +import io.moquette.spi.impl.subscriptions.Topic; +import io.moquette.spi.security.IAuthorizator; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.mqtt.MqttMessageBuilders; +import io.netty.handler.codec.mqtt.MqttPublishMessage; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Properties; + +import static io.moquette.spi.impl.ProtocolProcessorTest.EMPTY_OBSERVERS; +import static io.moquette.spi.impl.ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.Assert.*; + +public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { + + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.class); + + static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + + Server m_server; + IMqttClient m_client; + IMqttClient m_publisher; + MessageCollector m_messagesCollector; + IConfig m_config; + + @BeforeClass + public static void beforeTests() { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_dataStore = new MqttDefaultFilePersistence(tmpDir); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); + } + + protected void startServer() throws IOException { + m_server = new Server(); + final Properties configProps = IntegrationUtils.prepareTestProperties(); + m_config = new MemoryConfig(configProps); + + final IAuthorizator authorizator = new IAuthorizator() { + @Override + public boolean canWrite(Topic topic, String user, String client) { + return true; + } + + @Override + public boolean canRead(Topic topic, String user, String client) { + return false; + } + }; + + m_server.startServer(m_config, EMPTY_OBSERVERS, null, new AcceptAllAuthenticator(), authorizator); + } + + @Before + public void setUp() throws Exception { + startServer(); + + m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); + m_messagesCollector = new MessageCollector(); + m_client.setCallback(m_messagesCollector); + + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); + } + + @After + public void tearDown() throws Exception { + if (m_client != null && m_client.isConnected()) { + m_client.disconnect(); + } + + if (m_publisher != null && m_publisher.isConnected()) { + m_publisher.disconnect(); + } + + stopServer(); + } + + private void stopServer() { + m_server.stopServer(); + } + + @Test + public void shouldNotInternalPublishOnReadBlockedSubscriptionTopic() throws Exception { + LOG.info("*** shouldNotInternalPublishOnReadBlockedSubscriptionTopic ***"); + + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + m_client.connect(options); + try { + m_client.subscribe("/topic", 0); + fail(); + } catch (MqttException mex) { + // it's OK, the subscribed should fail with error code 128 + } + + // Exercise + MqttPublishMessage message = MqttMessageBuilders.publish() + .topicName("/topic") + .retained(true) + .qos(MqttQoS.AT_MOST_ONCE) +// .qQos(MqttQoS.AT_LEAST_ONCE) +// .qos(MqttQoS.EXACTLY_ONCE) + .payload(Unpooled.copiedBuffer("Hello World!!".getBytes(UTF_8))) + .build(); + + m_server.internalPublish(message, "INTRLPUB"); + + // TODO verify the message is not published + final MqttMessage mqttMessage = m_messagesCollector.waitMessage(1); + assertNull(mqttMessage); + } + + +// @Ignore("This test hasn't any meaning using in memory storage service") +// @Test +// public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { +// LOG.info("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("/topic", 0); +// m_client.disconnect(); +// +// m_server.stopServer(); +// +// m_server.startServer(IntegrationUtils.prepareTestProperties()); +// +// // reconnect and publish +// m_client.connect(options); +// m_client.publish("/topic", "Test my payload".getBytes(UTF_8), 0, false); +// +// assertEquals("/topic", m_messagesCollector.getTopic()); +// } +// +// /** +// * subscriber A connect and subscribe on "a/b" QoS 1 subscriber B connect and subscribe on "a/+" +// * BUT with QoS 2 publisher connects and send a message "hello" on "a/b" subscriber A must +// * receive a notification with QoS1 subscriber B must receive a notification with QoS2 +// */ +// @Test +// public void checkSubscribersGetCorrectQosNotifications() throws Exception { +// LOG.info("*** checkSubscribersGetCorrectQosNotifications ***"); +// String tmpDir = System.getProperty("java.io.tmpdir"); +// +// MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberA"); +// +// MqttClient subscriberA = new MqttClient("tcp://localhost:1883", "SubscriberA", dsSubscriberA); +// MessageCollector cbSubscriberA = new MessageCollector(); +// subscriberA.setCallback(cbSubscriberA); +// subscriberA.connect(); +// subscriberA.subscribe("a/b", 1); +// +// MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberB"); +// +// MqttClient subscriberB = new MqttClient("tcp://localhost:1883", "SubscriberB", dsSubscriberB); +// MessageCollector cbSubscriberB = new MessageCollector(); +// subscriberB.setCallback(cbSubscriberB); +// subscriberB.connect(); +// subscriberB.subscribe("a/+", 2); +// +// m_client.connect(); +// m_client.publish("a/b", "Hello world MQTT!!".getBytes(UTF_8), 2, false); +// +// MqttMessage messageOnA = cbSubscriberA.waitMessage(1); +// assertEquals("Hello world MQTT!!", new String(messageOnA.getPayload(), UTF_8)); +// assertEquals(1, messageOnA.getQos()); +// subscriberA.disconnect(); +// +// MqttMessage messageOnB = cbSubscriberB.waitMessage(1); +// assertEquals("Hello world MQTT!!", new String(messageOnB.getPayload(), UTF_8)); +// assertEquals(2, messageOnB.getQos()); +// subscriberB.disconnect(); +// } +// +// @Test +// public void testSubcriptionDoesntStayActiveAfterARestart() throws Exception { +// LOG.info("*** testSubcriptionDoesntStayActiveAfterARestart ***"); +// String tmpDir = System.getProperty("java.io.tmpdir"); +// // clientForSubscribe1 connect and subscribe to /topic QoS2 +// MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence( +// tmpDir + File.separator + "clientForSubscribe1"); +// +// MqttClient clientForSubscribe1 = new MqttClient("tcp://localhost:1883", "clientForSubscribe1", dsSubscriberA); +// MessageCollector cbSubscriber1 = new MessageCollector(); +// clientForSubscribe1.setCallback(cbSubscriber1); +// clientForSubscribe1.connect(); +// clientForSubscribe1.subscribe("topic", 0); +// +// // server stop +// m_server.stopServer(); +// System.out.println("\n\n SEVER REBOOTING \n\n"); +// // server start +// startServer(); +// +// // clientForSubscribe2 connect and subscribe to /topic QoS2 +// MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence( +// tmpDir + File.separator + "clientForSubscribe2"); +// MqttClient clientForSubscribe2 = new MqttClient("tcp://localhost:1883", "clientForSubscribe2", dsSubscriberB); +// MessageCollector cbSubscriber2 = new MessageCollector(); +// clientForSubscribe2.setCallback(cbSubscriber2); +// clientForSubscribe2.connect(); +// clientForSubscribe2.subscribe("topic", 0); +// +// // clientForPublish publish on /topic with QoS2 a message +// MqttClientPersistence dsSubscriberPUB = new MqttDefaultFilePersistence( +// tmpDir + File.separator + "clientForPublish"); +// MqttClient clientForPublish = new MqttClient("tcp://localhost:1883", "clientForPublish", dsSubscriberPUB); +// clientForPublish.connect(); +// clientForPublish.publish("topic", "Hello".getBytes(UTF_8), 2, true); +// +// // verify clientForSubscribe1 doesn't receive a notification but clientForSubscribe2 yes +// LOG.info("Before waiting to receive 1 sec from {}", clientForSubscribe1.getClientId()); +// assertFalse(clientForSubscribe1.isConnected()); +// assertTrue(clientForSubscribe2.isConnected()); +// LOG.info("Waiting to receive 1 sec from {}", clientForSubscribe2.getClientId()); +// MqttMessage messageOnB = cbSubscriber2.waitMessage(1); +// assertEquals("Hello", new String(messageOnB.getPayload(), UTF_8)); +// } +}