Issue #530 - Migrating from old maxInFlight Option to using the server assigned receive maximum.

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

Author: jpwsutton

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttConnectionOptions.java

@@ -65,7 +65,6 @@ enum UriType {
 	private int automaticReconnectMaxDelay = 120; // Max time to wait for automatic reconnection attempts in seconds.
 	private boolean useSubscriptionIdentifiers = true; // Whether to automatically assign subscription identifiers.
 	private int keepAliveInterval = 60; // Keep Alive Interval
-	private int maxInflight = 10; // Max inflight messages
 	private int connectionTimeout = 30; // Connection timeout in seconds
 	private boolean httpsHostnameVerificationEnabled = true;
 	private int maxReconnectDelay = 128000;
@@ -281,34 +280,6 @@ public void setKeepAliveInterval(int keepAliveInterval){
 		this.keepAliveInterval = keepAliveInterval;
 	}
 
-	/**
-	 * Returns the "max inflight". The max inflight limits to how many messages we
-	 * can send without receiving acknowledgments.
-	 * 
-	 * @see #setMaxInflight(int)
-	 * @return the max inflight
-	 */
-	public int getMaxInflight() {
-		return maxInflight;
-	}
-
-	/**
-	 * Sets the "max inflight". please increase this value in a high traffic
-	 * environment.
-	 * <p>
-	 * The default value is 10
-	 * </p>
-	 * 
-	 * @param maxInflight
-	 *            the number of maxInfligt messages
-	 */
-	public void setMaxInflight(int maxInflight) {
-		if (maxInflight < 0) {
-			throw new IllegalArgumentException();
-		}
-		this.maxInflight = maxInflight;
-	}
-
 	/**
 	 * Returns the connection timeout value.
 	 * 

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttConnectionOptionsBuilder.java

@@ -36,11 +36,6 @@ public MqttConnectionOptionsBuilder keepAliveInterval(int keepAlive) {
 		mqttConnectionOptions.setKeepAliveInterval(keepAlive);
 		return this;
 	}
-
-	public MqttConnectionOptionsBuilder maxInFlight(int maxInflight) {
-		mqttConnectionOptions.setMaxInflight(maxInflight);
-		return this;
-	}
 
 	public MqttConnectionOptionsBuilder connectionTimeout(int connectionTimeout) {
 		mqttConnectionOptions.setConnectionTimeout(connectionTimeout);

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ClientComms.java

@@ -341,7 +341,6 @@ public void connect(MqttConnectionOptions options, MqttToken token) throws MqttE
 
 				this.clientState.setKeepAliveSecs(conOptions.getKeepAliveInterval());
 				this.clientState.setCleanSession(conOptions.isCleanSession());
-				this.clientState.setMaxInflight(conOptions.getMaxInflight());
 
 				tokenStore.open();
 				ConnectBG conbg = new ConnectBG(this, token, connect, executorService);
@@ -946,7 +945,7 @@ class ReconnectDisconnectedBufferCallback implements IDisconnectedBufferCallback
 
 		public void publishBufferedMessage(BufferedMessage bufferedMessage) throws MqttException {
 			if (isConnected()) {
-				while (clientState.getActualInFlight() >= (clientState.getMaxInFlight() - 1)) {
+				while (clientState.getActualInFlight() >= (mqttSession.getReceiveMaximum() - 1)) {
 					// We need to Yield to the other threads to allow the in flight messages to
 					// clear
 					Thread.yield();

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ClientState.java

@@ -125,8 +125,7 @@ public class ClientState implements MqttState {
 	private long keepAlive;
 	private boolean cleanSession;
 	private MqttClientPersistence persistence;
-
-	private int maxInflight = 0;
+	
 	private int actualInFlight = 0;
 	private int inFlightPubRels = 0;
 
@@ -185,11 +184,6 @@ protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenSt
 		restoreState();
 	}
 
-	protected void setMaxInflight(int maxInflight) {
-		this.maxInflight = maxInflight;
-		pendingMessages = new Vector<MqttWireMessage>(this.maxInflight);
-	}
-
 	protected void setKeepAliveSecs(long keepAliveSecs) {
 		this.keepAlive = keepAliveSecs * 1000;
 	}
@@ -469,7 +463,7 @@ protected void restoreState() throws MqttException {
 
 	private void restoreInflightMessages() {
 		final String methodName = "restoreInflightMessages";
-		pendingMessages = new Vector<MqttWireMessage>(this.maxInflight);
+		pendingMessages = new Vector<MqttWireMessage>(this.mqttSession.getReceiveMaximum());
 		pendingFlows = new Vector<MqttWireMessage>();
 
 		Enumeration<Integer> keys = outboundQoS2.keys();
@@ -560,7 +554,7 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 
 		if (message instanceof MqttPublish) {
 			synchronized (queueLock) {
-				if (actualInFlight >= this.maxInflight) {
+				if (actualInFlight >= this.mqttSession.getReceiveMaximum()) {
 					// @TRACE 613= sending {0} msgs at max inflight window
 					log.fine(CLASS_NAME, methodName, "613", new Object[] { Integer.valueOf(actualInFlight) });
 
@@ -846,7 +840,7 @@ protected MqttWireMessage get() throws MqttException {
 				// freed.
 				// In both cases queueLock will be notified.
 				if ((pendingMessages.isEmpty() && pendingFlows.isEmpty())
-						|| (pendingFlows.isEmpty() && actualInFlight >= this.maxInflight)) {
+						|| (pendingFlows.isEmpty() && actualInFlight >= this.mqttSession.getReceiveMaximum())) {
 					try {
 						// @TRACE 644=wait for new work or for space in the inflight window
 						log.fine(CLASS_NAME, methodName, "644");
@@ -892,7 +886,7 @@ protected MqttWireMessage get() throws MqttException {
 
 					// If the inflight window is full then messages are not
 					// processed until the inflight window has space.
-					if (actualInFlight < this.maxInflight) {
+					if (actualInFlight < this.mqttSession.getReceiveMaximum()) {
 						// The in flight window is not full so process the
 						// first message in the queue
 						result = (MqttWireMessage) pendingMessages.elementAt(0);
@@ -1526,16 +1520,6 @@ public int getActualInFlight() {
 		return actualInFlight;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see org.eclipse.paho.mqttv5.client.internal.MqttState#getMaxInFlight()
-	 */
-	@Override
-	public int getMaxInFlight() {
-		return maxInflight;
-	}
-
 	/**
 	 * Tidy up - ensure that tokens are released as they are maintained over a
 	 * disconnect / connect cycle.
@@ -1576,7 +1560,7 @@ public Properties getDebug() {
 		props.put("In use msgids", inUseMsgIds);
 		props.put("pendingMessages", pendingMessages);
 		props.put("pendingFlows", pendingFlows);
-		props.put("maxInflight", Integer.valueOf(maxInflight));
+		props.put("serverReceiveMaximum", Integer.valueOf(this.mqttSession.getReceiveMaximum()));
 		props.put("nextMsgID", Integer.valueOf(nextMsgId));
 		props.put("actualInFlight", Integer.valueOf(actualInFlight));
 		props.put("inFlightPubRels", Integer.valueOf(inFlightPubRels));

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/MqttState.java

@@ -107,8 +107,6 @@ public interface MqttState {
 
 	int getActualInFlight();
 
-	int getMaxInFlight();
-
 	Properties getDebug();
 
 }