private void checkForActivity() throws MqttException {
		final String methodName = "checkForActivity";
		if(this.isMyControll && connected && this.keepAlive > 0) { //
			pingOutstanding = true;
			lastPing =  System.currentTimeMillis();;
			MqttToken token = new MqttToken(clientComms.getClient().getClientId());
			tokenStore.saveToken(token, pingCommand);
			pendingFlows.insertElementAt(pingCommand, 0);
			this.isMyControll=false;
			return;
		}
		if (connected && this.keepAlive > 0) {
			long time = System.currentTimeMillis();
		
			if (!pingOutstanding) {
				// Is a ping required? 
				if (time - lastOutboundActivity >= this.keepAlive ||
					time - lastInboundActivity >= this.keepAlive) {

					//@TRACE 620=ping needed. keepAlive={0} lastOutboundActivity={1} lastInboundActivity={2}
					log.fine(className,methodName,"620", new Object[]{new Long(this.keepAlive),new Long(lastOutboundActivity),new Long(lastInboundActivity)});
					
					pingOutstanding = true;
					lastPing = time;
					MqttToken token = new MqttToken(clientComms.getClient().getClientId());
					tokenStore.saveToken(token, pingCommand);
					pendingFlows.insertElementAt(pingCommand, 0);
				}
			} else if (time - lastPing >= this.keepAlive) {
				// A ping is outstanding but no packet has been received in KA so connection is deemed broken
				//@TRACE 619=Timed out as no activity, keepAlive={0} lastOutboundActivity={1} lastInboundActivity={2}
				log.severe(className,methodName,"619", new Object[]{new Long(this.keepAlive),new Long(lastOutboundActivity),new Long(lastInboundActivity)});
				
				// A ping has already been sent. At this point, assume that the
				// broker has hung and the TCP layer hasn't noticed.
				throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_TIMEOUT);
			}		
		}
	}