package org.apache.jmeter.protocol.mqtt.client;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Listener;

public class ListenerforSubscribe implements Listener {

	@Override
	public void onConnected() {
		System.out.println("Subscriber is listening");

	}

	@Override
	public void onDisconnected() {
		System.out.println("Subscriber disabled listening");

	}

	@Override
	public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
//		String message = new String(body.getData());
//		System.out.println("Received: "+message);
		ack.run();

	}

	@Override
	public void onFailure(Throwable value) {
		System.out.println("Subscriber couldn't set up listener");
		System.out.println(value);
	}

}

@@ -42,16 +42,67 @@
 lient;%0A%0A
+import java.util.concurrent.atomic.AtomicInteger;%0A%0A
 import o
@@ -123,32 +123,32 @@
 hawtbuf.Buffer;%0A
-
 import org.fuses
@@ -275,16 +275,76 @@
 ener %7B%0A%0A
+%09private static AtomicInteger count= new AtomicInteger(0); %0A
 %09@Overri
@@ -662,16 +662,16 @@
 ata());%0A
-
 //%09%09Syst
@@ -708,16 +708,105 @@
 ssage);%0A
+%09%09count.getAndIncrement();%0A%09%09System.out.println(%22Received %22 + count.get() +%22 messages%22);%0A
 %09%09ack.ru

update callback