Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

This closes #4190

This commit is contained in:
Justin Bertram 2022-08-26 12:00:49 -05:00
parent a0f39a4b28 755b987ab1
commit 3fee038f85
No known key found for this signature in database
GPG Key ID: F41830B875BB8633
6 changed files with 111 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
View File

@ -195,7 +195,7 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
if (!state.isAttached() && sessionExpiryInterval > 0 && state.getDisconnectedTime() + (sessionExpiryInterval * 1000) < System.currentTimeMillis()) {
toRemove.add(entry.getKey());
}
if (state.isWill() && !state.isAttached() && state.isFailed() && !state.isWillSent() && state.getWillDelayInterval() > 0 && state.getDisconnectedTime() + (state.getWillDelayInterval() * 1000) < System.currentTimeMillis()) {
if (state.isWill() && !state.isAttached() && state.isFailed() && state.getWillDelayInterval() > 0 && state.getDisconnectedTime() + (state.getWillDelayInterval() * 1000) < System.currentTimeMillis()) {
state.getSession().sendWillMessage();
}
}
@ -203,7 +203,7 @@ public class MQTTProtocolManager extends AbstractProtocolManager<MqttMessage, MQ
for (String key : toRemove) {
logger.debugf("Removing state for session: %s", key);
MQTTSessionState state = removeSessionState(key);
if (state != null && state.isWill() && !state.isAttached() && state.isFailed() && !state.isWillSent()) {
if (state != null && state.isWill() && !state.isAttached() && state.isFailed()) {
state.getSession().sendWillMessage();
}
}

46
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java
View File

@ -256,32 +256,30 @@ public class MQTTSession {
}
public void sendWillMessage() {
try {
MqttProperties properties;
if (state.getWillUserProperties() == null) {
properties = MqttProperties.NO_PROPERTIES;
} else {
properties = new MqttProperties();
for (MqttProperties.MqttProperty userProperty : state.getWillUserProperties()) {
properties.add(userProperty);
if (state.getWillStatus() == MQTTSessionState.WillStatus.NOT_SENT) {
try {
state.setWillStatus(MQTTSessionState.WillStatus.SENDING);
MqttProperties properties;
if (state.getWillUserProperties() == null) {
properties = MqttProperties.NO_PROPERTIES;
} else {
properties = new MqttProperties();
for (MqttProperties.MqttProperty userProperty : state.getWillUserProperties()) {
properties.add(userProperty);
}
}
MqttPublishMessage publishMessage = MqttMessageBuilders.publish().messageId(0).qos(MqttQoS.valueOf(state.getWillQoSLevel())).retained(state.isWillRetain()).topicName(state.getWillTopic()).payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage()).properties(properties).build();
logger.debugf("%s sending will message: %s", this, publishMessage);
getMqttPublishManager().sendToQueue(publishMessage, true);
state.setWillStatus(MQTTSessionState.WillStatus.SENT);
state.setWillMessage(null);
} catch (ActiveMQSecurityException e) {
state.setWillStatus(MQTTSessionState.WillStatus.NOT_SENT);
MQTTLogger.LOGGER.authorizationFailureSendingWillMessage(e.getMessage());
} catch (Exception e) {
state.setWillStatus(MQTTSessionState.WillStatus.NOT_SENT);
MQTTLogger.LOGGER.errorSendingWillMessage(e);
}
MqttPublishMessage publishMessage = MqttMessageBuilders.publish()
.messageId(0)
.qos(MqttQoS.valueOf(state.getWillQoSLevel()))
.retained(state.isWillRetain())
.topicName(state.getWillTopic())
.payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage())
.properties(properties)
.build();
logger.debugf("%s sending will message: %s", this, publishMessage);
getMqttPublishManager().sendToQueue(publishMessage, true);
state.setWillSent(true);
state.setWillMessage(null);
} catch (ActiveMQSecurityException e) {
MQTTLogger.LOGGER.authorizationFailureSendingWillMessage(e.getMessage());
} catch (Exception e) {
MQTTLogger.LOGGER.errorSendingWillMessage(e);
}
}

42
artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
View File

@ -78,7 +78,7 @@ public class MQTTSessionState {
private List<? extends MqttProperties.MqttProperty> willUserProperties;
private boolean willSent = false;
private WillStatus willStatus = WillStatus.NOT_SENT;
private boolean failed = false;
@ -113,7 +113,7 @@ public class MQTTSessionState {
willMessage.clear();
willMessage = null;
}
willSent = false;
willStatus = WillStatus.NOT_SENT;
failed = false;
willDelayInterval = 0;
willRetain = false;
@ -282,12 +282,12 @@ public class MQTTSessionState {
return willUserProperties;
}
public boolean isWillSent() {
return willSent;
public WillStatus getWillStatus() {
return willStatus;
}
public void setWillSent(boolean willSent) {
this.willSent = willSent;
public void setWillStatus(WillStatus willStatus) {
this.willStatus = willStatus;
}
public boolean isFailed() {
@ -448,4 +448,34 @@ public class MQTTSessionState {
public String toString() {
return "MQTTSessionState[" + "session=" + session + ", clientId='" + clientId + "', subscriptions=" + subscriptions + ", messageRefStore=" + messageRefStore + ", addressMessageMap=" + addressMessageMap + ", pubRec=" + pubRec + ", attached=" + attached + ", outboundStore=" + outboundStore + ", disconnectedTime=" + disconnectedTime + ", sessionExpiryInterval=" + clientSessionExpiryInterval + ", isWill=" + isWill + ", willMessage=" + willMessage + ", willTopic='" + willTopic + "', willQoSLevel=" + willQoSLevel + ", willRetain=" + willRetain + ", willDelayInterval=" + willDelayInterval + ", failed=" + failed + ", maxPacketSize=" + clientMaxPacketSize + ']';
}
public enum WillStatus {
NOT_SENT, SENT, SENDING;
public byte getStatus() {
switch (this) {
case NOT_SENT:
return 0;
case SENT:
return 1;
case SENDING:
return 2;
default:
return -1;
}
}
public static WillStatus getStatus(byte status) {
switch (status) {
case 0:
return NOT_SENT;
case 1:
return SENT;
case 2:
return SENDING;
default:
return null;
}
}
}
}

7
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
View File

@ -251,6 +251,13 @@ public final class PagingManagerImpl implements PagingManager {
}
}
/*
* For tests only!
*/
protected void setDiskFull(boolean diskFull) {
this.diskFull = diskFull;
}
@Override
public boolean isDiskFull() {
return diskFull;

23
tests/integration-tests/src/test/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImplAccessor.java Normal file
View File

@ -0,0 +1,23 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.artemis.core.paging.impl;
public class PagingManagerImplAccessor {
public static void setDiskFull(PagingManagerImpl pagingManager, boolean diskFull) {
pagingManager.setDiskFull(diskFull);
}
}

21
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java
View File

@ -27,10 +27,14 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.paging.impl.PagingManagerImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingManagerImplAccessor;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTReasonCodes;
import org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.tests.util.RandomUtil;
import org.apache.activemq.artemis.utils.Wait;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
@ -225,4 +229,21 @@ public class MQTT5Test extends MQTT5TestSupport {
server.start();
org.apache.activemq.artemis.tests.util.Wait.assertTrue(() -> getSubscriptionQueue(topic, clientId) != null, 3000, 10);
}
@Test(timeout = DEFAULT_TIMEOUT)
public void testRecursiveWill() throws Exception {
AssertionLoggerHandler.startCapture(true);
try {
final String WILL_QUEUE = "will";
server.createQueue(new QueueConfiguration(WILL_QUEUE).setRoutingType(RoutingType.ANYCAST));
PagingManagerImplAccessor.setDiskFull((PagingManagerImpl) server.getPagingManager(), true);
MqttClient client = createPahoClient("willGenerator");
MqttConnectionOptions options = new MqttConnectionOptionsBuilder().will(WILL_QUEUE, new MqttMessage(RandomUtil.randomBytes())).build();
client.connect(options);
client.disconnectForcibly(0, 0, false);
Wait.assertTrue(() -> AssertionLoggerHandler.findText("AMQ229119"), 2000, 100);
} finally {
AssertionLoggerHandler.stopCapture();
}
}
}