Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MQTT: Possible NPE

Browse Source 
Examining the code revealed a possible (but improbable) NPE.

Checking the connection was disjoint from the publish; encapsulate the check within publish.
Synchronize the connectionLost method so it can't null the client while it is being checked.

Add `@SuppressWarnings("deprecation")` on the deprecated `connectIfNeeded()` usage.
This commit is contained in:
Gary Russell
2015-11-08 14:02:17 -05:00
committed by Artem Bilan
parent 90a485e6a9
commit f5fa979681
2 changed files with 28 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -133,8 +133,9 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
}
@Override
@SuppressWarnings("deprecation")
protected void handleMessageInternal(Message<?> message) throws Exception {
this.connectIfNeeded();
connectIfNeeded();
String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);
Object mqttMessage = this.converter.fromMessage(message, Object.class);
if (topic == null && this.defaultTopic == null) {
@@ -144,7 +145,14 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
}
protected abstract void connectIfNeeded();
/**
* Invoked before {@link #publish(String, Object, Message)}.
* @deprecated subclasses should check the connection in
* {@link #publish(String, Object, Message)}.
*/
@Deprecated
protected void connectIfNeeded() {
}
protected abstract void publish(String topic, Object mqttMessage, Message<?> message) throws Exception;

38
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -154,41 +154,37 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
}
}
private synchronized void doConnect() throws MqttException {
private synchronized MqttAsyncClient checkConnection() throws MqttException {
if (this.client != null && !this.client.isConnected()) {
this.client.close();
this.client = null;
}
if (this.client == null) {
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
incrementClientInstance();
this.client.setCallback(this);
this.client.connect(connectionOptions).waitForCompletion(this.completionTimeout);
if (logger.isDebugEnabled()) {
logger.debug("Client connected");
}
}
}
@Override
protected void connectIfNeeded() {
if (this.client == null || !this.client.isConnected()) {
try {
this.doConnect();
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
MqttAsyncClient client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
incrementClientInstance();
client.setCallback(this);
client.connect(connectionOptions).waitForCompletion(this.completionTimeout);
this.client = client;
if (logger.isDebugEnabled()) {
logger.debug("Client connected");
}
}
catch (MqttException e) {
throw new MessagingException("Failed to connect", e);
}
}
return this.client;
}
@Override
protected void publish(String topic, Object mqttMessage, Message<?> message) throws Exception {
Assert.isInstanceOf(MqttMessage.class, mqttMessage);
IMqttDeliveryToken token = this.client.publish(topic, (MqttMessage) mqttMessage);
MqttAsyncClient client = checkConnection();
IMqttDeliveryToken token = client.publish(topic, (MqttMessage) mqttMessage);
if (!this.async) {
token.waitForCompletion(this.completionTimeout);
}
@@ -208,7 +204,7 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
}
@Override
public void connectionLost(Throwable cause) {
public synchronized void connectionLost(Throwable cause) {
logger.error("Lost connection; will attempt reconnect on next request");
this.client = null;
}