import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
/**
 * start simple mqtt server
 * @throws IOException throw at {@link io.moquette.server.netty.NettyAcceptor}
 */
public void start() {
    List<InterceptHandler> interceptHandlers = Collections.emptyList();

    boolean isVerbose = config.getConfiguration().getBoolean("verbose", false);
    if (isVerbose) {
        interceptHandlers = new ArrayList<>();
        interceptHandlers.add(new LogInterceptHandler());
    }

    try {
        mqttBroker.startServer(config, interceptHandlers);
    } catch (IOException e) {
        logger.error("Start mqtt broker.", e);
    }
    System.out.println("MQTT Broker started.");
    Runtime.getRuntime().addShutdownHook(new Thread() {
        @Override
        public void run() {
            logger.info("Stopping broker");
            mqttBroker.stopServer();
            logger.info("Broker stopped");
        }
    });
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@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);
            }
    });
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Test
public void testAddAndRemoveInterceptHandler() throws Exception {
    InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class);
    InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class);
    // add
    interceptor.addInterceptHandler(interceptHandlerMock1);
    interceptor.addInterceptHandler(interceptHandlerMock2);

    Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE);
    interceptor.notifyTopicSubscribed(subscription, "cli1234");
    interval();

    verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
    verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));

    // remove
    interceptor.removeInterceptHandler(interceptHandlerMock1);

    interceptor.notifyTopicSubscribed(subscription, "cli1235");
    interval();
    // removeInterceptHandler() performs another interaction
    // TODO: fix this
    // verifyNoMoreInteractions(interceptHandlerMock1);
    verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235")));
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, IOException {
        IResourceLoader classpathLoader = new ClasspathResourceLoader();
        final IConfig classPathConfig = new ResourceLoaderConfig(classpathLoader);

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

        System.out.println("Broker started press [CTRL+C] to stop");
        //Bind  a shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Stopping broker");
            mqttBroker.stopServer();
            System.out.println("Broker stopped");
        }));

        Thread.sleep(20000);
        System.out.println("Before self publish");
        MqttPublishMessage message = MqttMessageBuilders.publish()
            .topicName("/exit")
            .retained(true)
//        qos(MqttQoS.AT_MOST_ONCE);
//        qQos(MqttQoS.AT_LEAST_ONCE);
            .qos(MqttQoS.EXACTLY_ONCE)
            .payload(Unpooled.copiedBuffer("Hello World!!".getBytes()))
            .build();

        mqttBroker.internalPublish(message, "INTRLPUB");
        System.out.println("After self publish");
    }
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
public static <T extends InterceptHandler> Collection<String> getInterceptorIds(Collection<T> handlers) {
    Collection<String> result = new ArrayList<>(handlers.size());
    for (T handler : handlers) {
        result.add(handler.getID());
    }
    return result;
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
private BrokerInterceptor(int poolSize, List<InterceptHandler> handlers) {
    LOG.info("Initializing broker interceptor. InterceptorIds={}", getInterceptorIds(handlers));
    this.handlers = new HashMap<>();
    for (Class<?> messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
        this.handlers.put(messageType, new CopyOnWriteArrayList<InterceptHandler>());
    }
    for (InterceptHandler handler : handlers) {
        this.addInterceptHandler(handler);
    }
    executor = Executors.newFixedThreadPool(poolSize);
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyClientConnected(final MqttConnectMessage msg) {
    for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
        LOG.debug("Sending MQTT CONNECT message to interceptor. CId={}, interceptorId={}",
                msg.payload().clientIdentifier(), handler.getID());
        executor.execute(() -> handler.onConnect(new InterceptConnectMessage(msg)));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyClientDisconnected(final String clientID, final String username) {
    for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) {
        LOG.debug("Notifying MQTT client disconnection to interceptor. CId={}, username={}, interceptorId={}",
            clientID, username, handler.getID());
        executor.execute(() -> handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyClientConnectionLost(final String clientID, final String username) {
    for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) {
        LOG.debug("Notifying unexpected MQTT client disconnection to interceptor CId={}, username={}, " +
            "interceptorId={}", clientID, username, handler.getID());
        executor.execute(() -> handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyTopicSubscribed(final Subscription sub, final String username) {
    for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) {
        LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. CId={}, topicFilter={}, interceptorId={}",
            sub.getClientId(), sub.getTopicFilter(), handler.getID());
        executor.execute(() -> handler.onSubscribe(new InterceptSubscribeMessage(sub, username)));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) {
    for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) {
        LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. CId={}, topic={}, interceptorId={}", clientID,
            topic, handler.getID());
        executor.execute(() -> handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) {
    for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
        LOG.debug("Notifying MQTT ACK message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}",
            msg.getMsg().getClientID(), msg.getPacketID(), msg.getTopic(), handler.getID());
        executor.execute(() -> handler.onMessageAcknowledged(msg));
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void addInterceptHandler(InterceptHandler interceptHandler) {
    Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
    LOG.info("Adding MQTT message interceptor. InterceptorId={}, handledMessageTypes={}",
        interceptHandler.getID(), interceptedMessageTypes);
    for (Class<?> interceptMessageType : interceptedMessageTypes) {
        this.handlers.get(interceptMessageType).add(interceptHandler);
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
@Override
public void removeInterceptHandler(InterceptHandler interceptHandler) {
    Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
    LOG.info("Removing MQTT message interceptor. InterceptorId={}, handledMessageTypes={}",
        interceptHandler.getID(), interceptedMessageTypes);
    for (Class<?> interceptMessageType : interceptedMessageTypes) {
        this.handlers.get(interceptMessageType).remove(interceptHandler);
    }
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
private static Class<?>[] getInterceptedMessageTypes(InterceptHandler interceptHandler) {
    Class<?>[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes();
    if (interceptedMessageTypes == null) {
        return InterceptHandler.ALL_MESSAGE_TYPES;
    }
    return interceptedMessageTypes;
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
public void startServer(IConfig config, List<? extends InterceptHandler> handlers, ISslContextCreator sslCtxCreator,
        IAuthenticator authenticator, IAuthorizator authorizator) throws IOException {
    if (handlers == null) {
        handlers = Collections.emptyList();
    }
    LOG.info("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers));

    scheduler = Executors.newScheduledThreadPool(1);

    final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
    if (handlerProp != null) {
        config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
    }
    configureCluster(config);
    final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME);
    LOG.info("Configuring Using persistent store file, path={}", persistencePath);
    m_processorBootstrapper = new ProtocolProcessorBootstrapper();
    final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator,
        this);
    LOG.info("Initialized MQTT protocol processor");
    if (sslCtxCreator == null) {
        LOG.warn("Using default SSL context creator");
        sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
    }

    LOG.info("Binding server to the configured ports");
    m_acceptor = new NettyAcceptor();
    m_acceptor.initialize(processor, config, sslCtxCreator);
    m_processor = processor;

    LOG.info("Moquette server has been initialized successfully");
    m_initialized = true;
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
/**
 * SPI method used by Broker embedded applications to add intercept handlers.
 *
 * @param interceptHandler
 *            the handler to add.
 */
public void addInterceptHandler(InterceptHandler interceptHandler) {
    if (!m_initialized) {
        LOG.error("Moquette is not started, MQTT message interceptor cannot be added. InterceptorId={}",
            interceptHandler.getID());
        throw new IllegalStateException("Can't register interceptors on a server that is not yet started");
    }
    LOG.info("Adding MQTT message interceptor. InterceptorId={}", interceptHandler.getID());
    m_processor.addInterceptHandler(interceptHandler);
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
/**
 * SPI method used by Broker embedded applications to remove intercept handlers.
 *
 * @param interceptHandler
 *            the handler to remove.
 */
public void removeInterceptHandler(InterceptHandler interceptHandler) {
    if (!m_initialized) {
        LOG.error("Moquette is not started, MQTT message interceptor cannot be removed. InterceptorId={}",
            interceptHandler.getID());
        throw new IllegalStateException("Can't deregister interceptors from a server that is not yet started");
    }
    LOG.info("Removing MQTT message interceptor. InterceptorId={}", interceptHandler.getID());
    m_processor.removeInterceptHandler(interceptHandler);
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
public void addInterceptHandler(InterceptHandler interceptHandler) {
    this.m_interceptor.addInterceptHandler(interceptHandler);
}
 

import io.moquette.interception.InterceptHandler; //导入依赖的package包/类
public void removeInterceptHandler(InterceptHandler interceptHandler) {
    this.m_interceptor.removeInterceptHandler(interceptHandler);
}