package io.moquette.interception; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.core.ITopic; import io.moquette.interception.messages.InterceptPublishMessage; import io.moquette.server.Server; import io.netty.buffer.ByteBuf; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static io.moquette.spi.impl.Utils.readBytesAndRewind; /** * Created by mackristof on 28/05/2016. */ public class HazelcastInterceptHandler extends AbstractInterceptHandler { private static final Logger LOG = LoggerFactory.getLogger(HazelcastInterceptHandler.class); private final HazelcastInstance hz; public HazelcastInterceptHandler(Server server){ this.hz = server.getHazelcastInstance(); } @Override public String getID() { return HazelcastInterceptHandler.class.getName() + "@" + hz.getName(); } @Override public void onPublish(InterceptPublishMessage msg) { //TODO ugly, too much array copy ByteBuf payload = msg.getPayload(); byte[] payloadContent = readBytesAndRewind(payload); LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(payloadContent)); ITopic topic = hz.getTopic("moquette"); HazelcastMsg hazelcastMsg = new HazelcastMsg(msg); topic.publish(hazelcastMsg); } }