@Test public void testNotifyTopicPublished() throws Exception { interceptor.notifyTopicPublished( MqttMessageBuilders.publish().qos(MqttQoS.AT_MOST_ONCE) .payload(Unpooled.copiedBuffer("Hello".getBytes(UTF_8))).build(), "cli1234", "cli1234"); interval(); assertEquals(60, n.get()); }
@Override public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { msg.retain(); executor.execute(() -> { try { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) { LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, " + "interceptorId={}", clientID, messageId, topic, handler.getID()); handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } } finally { ReferenceCountUtil.release(msg); } }); }
BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { msg.retain(); executor.execute(() -> { try { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) { LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, " + "interceptorId={}", clientID, messageId, topic, handler.getID()); handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } } finally { ReferenceCountUtil.release(msg); } }); } }
BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { msg.retain(); executor.execute(() -> { try { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) { LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, " + "interceptorId={}", clientID, messageId, topic, handler.getID()); handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } } finally { ReferenceCountUtil.release(msg); } }); } private  BrokerInterceptor(int poolSize, List<InterceptHandler> handlers);  BrokerInterceptor(List<InterceptHandler> handlers);  BrokerInterceptor(IConfig props, List<InterceptHandler> handlers); }
BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { msg.retain(); executor.execute(() -> { try { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) { LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, " + "interceptorId={}", clientID, messageId, topic, handler.getID()); handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } } finally { ReferenceCountUtil.release(msg); } }); } private  BrokerInterceptor(int poolSize, List<InterceptHandler> handlers);  BrokerInterceptor(List<InterceptHandler> handlers);  BrokerInterceptor(IConfig props, List<InterceptHandler> handlers); @Override void notifyClientConnected(final MqttConnectMessage msg); @Override void notifyClientDisconnected(final String clientID, final String username); @Override void notifyClientConnectionLost(final String clientID, final String username); @Override void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username); @Override void notifyTopicSubscribed(final Subscription sub, final String username); @Override void notifyTopicUnsubscribed(final String topic, final String clientID, final String username); @Override void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg); @Override void addInterceptHandler(InterceptHandler interceptHandler); @Override void removeInterceptHandler(InterceptHandler interceptHandler); }
BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) { msg.retain(); executor.execute(() -> { try { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) { LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, " + "interceptorId={}", clientID, messageId, topic, handler.getID()); handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); } } finally { ReferenceCountUtil.release(msg); } }); } private  BrokerInterceptor(int poolSize, List<InterceptHandler> handlers);  BrokerInterceptor(List<InterceptHandler> handlers);  BrokerInterceptor(IConfig props, List<InterceptHandler> handlers); @Override void notifyClientConnected(final MqttConnectMessage msg); @Override void notifyClientDisconnected(final String clientID, final String username); @Override void notifyClientConnectionLost(final String clientID, final String username); @Override void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username); @Override void notifyTopicSubscribed(final Subscription sub, final String username); @Override void notifyTopicUnsubscribed(final String topic, final String clientID, final String username); @Override void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg); @Override void addInterceptHandler(InterceptHandler interceptHandler); @Override void removeInterceptHandler(InterceptHandler interceptHandler);  }