Issue 216 (#266)

Adding subscribeWithResponse call to MqttClient

Signed-off-by: James Sutton <james.sutton@uk.ibm.com>

Author: jpwsutton

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttClient.java

@@ -435,8 +435,277 @@ public interface IMqttClient { //extends IMqttAsyncClient {
 	 * @throws MqttException if there was an error registering the subscription.
 	 * @throws IllegalArgumentException if the two supplied arrays are not the same size.
 	 */
-public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException;
+	public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException;
 
+	/**
+	 * Subscribe to a topic, which may include wildcards using a QoS of 1.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilter the topic to subscribe to, which can include wildcards.
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter) throws MqttException;
+	
+	/**
+	 * Subscribe to a topic, which may include wildcards using a QoS of 1.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilter the topic to subscribe to, which can include wildcards.
+	 * @param messageListener a callback to handle incoming messages
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, IMqttMessageListener messageListener) throws MqttException;
+
+	
+	/**
+	 * Subscribe to a topic, which may include wildcards.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilter the topic to subscribe to, which can include wildcards.
+	 * @param qos the maximum quality of service at which to subscribe. Messages
+	 * published at a lower quality of service will be received at the published
+	 * QoS.  Messages published at a higher quality of service will be received using
+	 * the QoS specified on the subscribe.
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, int qos) throws MqttException;
+	
+	/**
+	 * Subscribe to a topic, which may include wildcards.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilter the topic to subscribe to, which can include wildcards.
+	 * @param qos the maximum quality of service at which to subscribe. Messages
+	 * published at a lower quality of service will be received at the published
+	 * QoS.  Messages published at a higher quality of service will be received using
+	 * the QoS specified on the subscribe.
+	 * @param messageListener a callback to handle incoming messages
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException;
+	
+	/**
+	 * Subscribes to a one or more topics, which may include wildcards using a QoS of 1.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilters the topic to subscribe to, which can include wildcards.
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters) throws MqttException;
+	
+	/**
+	 * Subscribes to a one or more topics, which may include wildcards using a QoS of 1.
+	 *
+	 * @see #subscribeWithResponse(String[], int[])
+	 *
+	 * @param topicFilters the topic to subscribe to, which can include wildcards.
+	 * @param messageListeners one or more callbacks to handle incoming messages
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws MqttException if there was an error registering the subscription.
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, IMqttMessageListener[] messageListeners) throws MqttException;
+	
+	/**
+	 * Subscribes to multiple topics, each of which may include wildcards.
+	 * <p>The {@link #setCallback(MqttCallback)} method
+	 * should be called before this method, otherwise any received messages
+	 * will be discarded.
+	 * </p>
+	 * <p>
+	 * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to true
+	 * when when connecting to the server then the subscription remains in place
+	 * until either:
+	 * <ul>
+	 * <li>The client disconnects</li>
+	 * <li>An unsubscribe method is called to un-subscribe the topic</li>
+	 * </li>
+	 * </p>
+	 * <p>
+	 * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to false
+	 * when when connecting to the server then the subscription remains in place
+	 * until either:
+	 * <ul>
+	 * <li>An unsubscribe method is called to unsubscribe the topic</li>
+	 * <li>The client connects with cleanSession set to true</ul>
+	 * </li>
+	 * With cleanSession set to false the MQTT server will store messages on
+	 * behalf of the client when the client is not connected. The next time the
+	 * client connects with the <bold>same client ID</bold> the server will
+	 * deliver the stored messages to the client.
+	 * </p>
+	 *
+	 * <p>The "topic filter" string used when subscribing
+	 * may contain special characters, which allow you to subscribe to multiple topics
+	 * at once.</p>
+	 * <p>The topic level separator is used to introduce structure into the topic, and
+	 * can therefore be specified within the topic for that purpose.  The multi-level
+	 * wildcard and single-level wildcard can be used for subscriptions, but they
+	 * cannot be used within a topic by the publisher of a message.
+	 * <dl>
+	 * 	<dt>Topic level separator</dt>
+	 * 	<dd>The forward slash (/) is used to separate each level within
+	 * 	a topic tree and provide a hierarchical structure to the topic space. The
+	 * 	use of the topic level separator is significant when the two wildcard characters
+	 * 	are encountered in topics specified by subscribers.</dd>
+	 *
+	 * 	<dt>Multi-level wildcard</dt>
+	 * 	<dd><p>The number sign (#) is a wildcard character that matches
+	 * 	any number of levels within a topic. For example, if you subscribe to
+	 *  <span><span class="filepath">finance/stock/ibm/#</span></span>, you receive
+	 * 	messages on these topics:
+	 *  <pre>   finance/stock/ibm<br />   finance/stock/ibm/closingprice<br />   finance/stock/ibm/currentprice</pre>
+	 *  </p>
+	 *  <p>The multi-level wildcard
+	 *  can represent zero or more levels. Therefore, <em>finance/#</em> can also match
+	 * 	the singular <em>finance</em>, where <em>#</em> represents zero levels. The topic
+	 * 	level separator is meaningless in this context, because there are no levels
+	 * 	to separate.</p>
+	 *
+	 * 	<p>The <span>multi-level</span> wildcard can
+	 * 	be specified only on its own or next to the topic level separator character.
+	 * 	Therefore, <em>#</em> and <em>finance/#</em> are both valid, but <em>finance#</em> is
+	 * 	not valid. <span>The multi-level wildcard must be the last character
+	 *  used within the topic tree. For example, <em>finance/#</em> is valid but
+	 *  <em>finance/#/closingprice</em> is 	not valid.</span></p></dd>
+	 *
+	 * 	<dt>Single-level wildcard</dt>
+	 * 	<dd><p>The plus sign (+) is a wildcard character that matches only one topic
+	 * 	level. For example, <em>finance/stock/+</em> matches
+	 * <em>finance/stock/ibm</em> and <em>finance/stock/xyz</em>,
+	 * 	but not <em>finance/stock/ibm/closingprice</em>. Also, because the single-level
+	 * 	wildcard matches only a single level, <em>finance/+</em> does not match <em>finance</em>.</p>
+	 *
+	 * 	<p>Use
+	 * 	the single-level wildcard at any level in the topic tree, and in conjunction
+	 * 	with the multilevel wildcard. Specify the single-level wildcard next to the
+	 * 	topic level separator, except when it is specified on its own. Therefore,
+	 *  <em>+</em> and <em>finance/+</em> are both valid, but <em>finance+</em> is
+	 *  not valid. <span>The single-level wildcard can be used at the end of the
+	 *  topic tree or within the topic tree.
+	 * 	For example, <em>finance/+</em> and <em>finance/+/ibm</em> are both valid.</span></p>
+	 * 	</dd>
+	 * </dl>
+	 * </p>
+	 *
+	 * <p>This is a blocking method that returns once subscribe completes</p>
+	 *
+	 * @param topicFilters one or more topics to subscribe to, which can include wildcards.
+	 * @param qos the maximum quality of service to subscribe each topic at.Messages
+	 * published at a lower quality of service will be received at the published
+	 * QoS.  Messages published at a higher quality of service will be received using
+	 * the QoS specified on the subscribe.
+	 * @throws MqttException if there was an error registering the subscription.
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws IllegalArgumentException if the two supplied arrays are not the same size.
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos) throws MqttException;
+
+	/**
+	 * Subscribes to multiple topics, each of which may include wildcards.
+	 * <p>The {@link #setCallback(MqttCallback)} method
+	 * should be called before this method, otherwise any received messages
+	 * will be discarded.
+	 * </p>
+	 * <p>
+	 * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to true
+	 * when when connecting to the server then the subscription remains in place
+	 * until either:
+	 * <ul>
+	 * <li>The client disconnects</li>
+	 * <li>An unsubscribe method is called to un-subscribe the topic</li>
+	 * </li>
+	 * </p>
+	 * <p>
+	 * If (@link MqttConnectOptions#setCleanSession(boolean)} was set to false
+	 * when when connecting to the server then the subscription remains in place
+	 * until either:
+	 * <ul>
+	 * <li>An unsubscribe method is called to unsubscribe the topic</li>
+	 * <li>The client connects with cleanSession set to true</ul>
+	 * </li>
+	 * With cleanSession set to false the MQTT server will store messages on
+	 * behalf of the client when the client is not connected. The next time the
+	 * client connects with the <bold>same client ID</bold> the server will
+	 * deliver the stored messages to the client.
+	 * </p>
+	 *
+	 * <p>The "topic filter" string used when subscribing
+	 * may contain special characters, which allow you to subscribe to multiple topics
+	 * at once.</p>
+	 * <p>The topic level separator is used to introduce structure into the topic, and
+	 * can therefore be specified within the topic for that purpose.  The multi-level
+	 * wildcard and single-level wildcard can be used for subscriptions, but they
+	 * cannot be used within a topic by the publisher of a message.
+	 * <dl>
+	 * 	<dt>Topic level separator</dt>
+	 * 	<dd>The forward slash (/) is used to separate each level within
+	 * 	a topic tree and provide a hierarchical structure to the topic space. The
+	 * 	use of the topic level separator is significant when the two wildcard characters
+	 * 	are encountered in topics specified by subscribers.</dd>
+	 *
+	 * 	<dt>Multi-level wildcard</dt>
+	 * 	<dd><p>The number sign (#) is a wildcard character that matches
+	 * 	any number of levels within a topic. For example, if you subscribe to
+	 *  <span><span class="filepath">finance/stock/ibm/#</span></span>, you receive
+	 * 	messages on these topics:
+	 *  <pre>   finance/stock/ibm<br />   finance/stock/ibm/closingprice<br />   finance/stock/ibm/currentprice</pre>
+	 *  </p>
+	 *  <p>The multi-level wildcard
+	 *  can represent zero or more levels. Therefore, <em>finance/#</em> can also match
+	 * 	the singular <em>finance</em>, where <em>#</em> represents zero levels. The topic
+	 * 	level separator is meaningless in this context, because there are no levels
+	 * 	to separate.</p>
+	 *
+	 * 	<p>The <span>multi-level</span> wildcard can
+	 * 	be specified only on its own or next to the topic level separator character.
+	 * 	Therefore, <em>#</em> and <em>finance/#</em> are both valid, but <em>finance#</em> is
+	 * 	not valid. <span>The multi-level wildcard must be the last character
+	 *  used within the topic tree. For example, <em>finance/#</em> is valid but
+	 *  <em>finance/#/closingprice</em> is 	not valid.</span></p></dd>
+	 *
+	 * 	<dt>Single-level wildcard</dt>
+	 * 	<dd><p>The plus sign (+) is a wildcard character that matches only one topic
+	 * 	level. For example, <em>finance/stock/+</em> matches
+	 * <em>finance/stock/ibm</em> and <em>finance/stock/xyz</em>,
+	 * 	but not <em>finance/stock/ibm/closingprice</em>. Also, because the single-level
+	 * 	wildcard matches only a single level, <em>finance/+</em> does not match <em>finance</em>.</p>
+	 *
+	 * 	<p>Use
+	 * 	the single-level wildcard at any level in the topic tree, and in conjunction
+	 * 	with the multilevel wildcard. Specify the single-level wildcard next to the
+	 * 	topic level separator, except when it is specified on its own. Therefore,
+	 *  <em>+</em> and <em>finance/+</em> are both valid, but <em>finance+</em> is
+	 *  not valid. <span>The single-level wildcard can be used at the end of the
+	 *  topic tree or within the topic tree.
+	 * 	For example, <em>finance/+</em> and <em>finance/+/ibm</em> are both valid.</span></p>
+	 * 	</dd>
+	 * </dl>
+	 * </p>
+	 *
+	 * <p>This is a blocking method that returns once subscribe completes</p>
+	 *
+	 * @param topicFilters one or more topics to subscribe to, which can include wildcards.
+	 * @param qos the maximum quality of service to subscribe each topic at.Messages
+	 * published at a lower quality of service will be received at the published
+	 * QoS.  Messages published at a higher quality of service will be received using
+	 * the QoS specified on the subscribe.
+	 * @param messageListeners one or more callbacks to handle incoming messages
+	 * @throws MqttException if there was an error registering the subscription.
+	 * @return token used to track the subscribe after it has completed.
+	 * @throws IllegalArgumentException if the two supplied arrays are not the same size.
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException;
+	
 	/**
 	 * Requests the server unsubscribe the client from a topic.
 	 *

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttClient.java

@@ -368,6 +368,81 @@ public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] m
 			aClient.comms.setMessageListener(topicFilters[i], messageListeners[i]);
 		}
 	}
+	
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String)
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter) throws MqttException {
+		return this.subscribeWithResponse(new String[] {topicFilter}, new int[] {1});
+	}
+	
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String, IMqttMessageListener)
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, IMqttMessageListener messageListener) throws MqttException {
+		return this.subscribeWithResponse(new String[] {topicFilter}, new int[] {1}, new IMqttMessageListener[] {messageListener});
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String, int)
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, int qos) throws MqttException {
+		return this.subscribeWithResponse(new String[] {topicFilter}, new int[] {qos});
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String, int, IMqttMessageListener)
+	 */
+	public IMqttToken subscribeWithResponse(String topicFilter, int qos, IMqttMessageListener messageListener)
+			throws MqttException {
+		return this.subscribeWithResponse(new String[] {topicFilter}, new int[] {qos}, new IMqttMessageListener[] {messageListener});
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String[])
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters) throws MqttException {
+		int[] qos = new int[topicFilters.length];
+		for (int i=0; i<qos.length; i++) {
+			qos[i] = 1;
+		}
+		return this.subscribeWithResponse(topicFilters, qos);
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String[], IMqttMessageListener[])
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, IMqttMessageListener[] messageListeners)
+			throws MqttException {
+		int[] qos = new int[topicFilters.length];
+		for (int i=0; i<qos.length; i++) {
+			qos[i] = 1;
+		}
+		return this.subscribeWithResponse(topicFilters, qos, messageListeners);
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String[], int[])
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos) throws MqttException {
+		IMqttToken tok = aClient.subscribe(topicFilters, qos, null, null);
+		tok.waitForCompletion(getTimeToWait());
+		return tok;
+	}
+
+	/*
+	 * @see IMqttClient#subscribeWithResponse(String[], int[], IMqttMessageListener[])
+	 */
+	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners)
+			throws MqttException {
+		IMqttToken tok = this.subscribeWithResponse(topicFilters, qos);
+		
+		// add message handlers to the list for this client
+		for (int i = 0; i < topicFilters.length; ++i) {
+			aClient.comms.setMessageListener(topicFilters[i], messageListeners[i]);
+		}
+		return tok;
+	}
 
 	/*
 	 * @see IMqttClient#unsubscribe(String)