How to intercept and print publish messages? #0.15

[saeedrahmo]

How to intercept publish message and print them in the console?

I attached the piece of code that I wrote and the error that I got.

Also I appreciate it if anybody could provide me with the documentation of Moquette (http://andsel.github.io/moquette/ not available!).

The code:

public class Main {
    static class PublisherListener extends AbstractInterceptHandler {
        @Override
        public void onPublish(InterceptPublishMessage message) {
            System.out.println("moquette mqtt broker message intercepted, topic: " + message.getTopicName()
                    + ", content: " + new String(message.getPayload().array()));
        }

        @Override
        public String getID() {
            return null;
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException {
        IResourceLoader classpathLoader = new ClasspathResourceLoader();
        final IConfig classPathConfig = new ResourceLoaderConfig(classpathLoader);


        final Server mqttBroker = new Server();
        final List<? extends InterceptHandler> userHandlers = Arrays.asList(new PublisherListener());
        mqttBroker.startServer(classPathConfig, userHandlers);

        System.out.println("moquette mqtt broker started, press ctrl-c to shutdown..");
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                System.out.println("stopping moquette mqtt broker..");
                mqttBroker.stopServer();
                System.out.println("moquette mqtt broker stopped");
            }
        });
    }
}

The error:

log4j:WARN No appenders could be found for logger (io.moquette.broker.config.ResourceLoaderConfig).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Exception in thread "pool-2-thread-1" java.lang.UnsupportedOperationException: direct buffer
	at io.netty.buffer.PooledUnsafeDirectByteBuf.array(PooledUnsafeDirectByteBuf.java:226)
	at io.netty.buffer.AbstractPooledDerivedByteBuf.array(AbstractPooledDerivedByteBuf.java:117)
	at Main$PublisherListener.onPublish(Main.java:26)
	at io.moquette.interception.BrokerInterceptor.lambda$notifyTopicPublished$3(BrokerInterceptor.java:134)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[hylkevds]

The answer is right there in the stacktrace. And in the javadoc of the method you're calling.

public abstract byte[] array()
Returns the backing byte array of this buffer.
Throws:
UnsupportedOperationException - if there no accessible backing byte array

Have you tried using one of the toString() methods of ByteBuf if you want to turn the buffer into a String?

[saeedrahmo]

It was the problem but the toString() also raised an exception. I fixed it with the following code:

ByteBuf payload = msg.getPayload();
byte[] payloadContent = readBytesAndRewind(payload);
System.out.println("moquette mqtt broker message intercepted, topic: " + msg.getTopicName()
                    + ", content: " + new String(payloadContent, UTF_8));

[hylkevds]

That's odd, which error did you get with the toString()?
You should not have to rewind the buffer before using it. Though that could be a bug that was fixed in the meantime in 0.16-SNAPSHOT.

[saeedrahmo]

I made a mistake, you are completely right. The toString() works fine and it has no problem.

 System.out.println("moquette mqtt broker message intercepted, topic: " + msg.getTopicName()
                    + ", content: " + msg.getPayload().toString(UTF_8));