<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/util/Validator.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.util;

import com.github.longkerdandy.mithqtt.http.MqttHttpConfiguration;
import org.apache.commons.lang3.StringUtils;

import java.util.regex.Pattern;

/**
 * Validator
 */
@SuppressWarnings("unused")
public class Validator {

    // MQTT client id validate regex pattern
    protected Pattern clientIdPattern;
    // MQTT topic name validate regex pattern
    protected Pattern topicNamePattern;
    // MQTT topic filter validate regex pattern
    protected Pattern topicFilterPattern;

    public Validator(MqttHttpConfiguration config) {
        if (StringUtils.isNotBlank(config.getClientIdValidator()))
            this.clientIdPattern = Pattern.compile(config.getClientIdValidator());
        if (StringUtils.isNotBlank(config.getTopicNameValidator()))
            this.topicNamePattern = Pattern.compile(config.getTopicNameValidator());
        if (StringUtils.isNotBlank(config.getTopicFilterValidator()))
            this.topicFilterPattern = Pattern.compile(config.getTopicFilterValidator());
    }

    /**
     * Is MQTT topic name (no wildcards) valid
     *
     * @param topicName Topic Name
     * @return True if valid
     */
    public boolean isTopicNameValid(String topicName) {
        return !StringUtils.isEmpty(topicName) &&
                !topicName.contains("+") &&
                !topicName.contains("#") &&
                (this.topicNamePattern == null || this.topicNamePattern.matcher(topicName).matches());
    }

    /**
     * Is MQTT topic filter (may contain wildcards) valid
     *
     * @param topicFilter Topic Filter
     * @return True if valid
     */
    public boolean isTopicFilterValid(String topicFilter) {
        return !StringUtils.isEmpty(topicFilter) &&
                (this.topicFilterPattern == null || this.topicFilterPattern.matcher(topicFilter).matches());
    }

    /**
     * Is MQTT client id valid
     *
     * @param clientId Client Id
     * @return True if valid
     */
    public boolean isClientIdValid(String clientId) {
        return this.clientIdPattern == null || this.clientIdPattern.matcher(clientId).matches();
    }
}


<|start_filename|>mithqtt-api/src/test/java/io/netty/handler/codec/mqtt/MqttVersionTest.java<|end_filename|>
package io.netty.handler.codec.mqtt;

import org.junit.Test;

/**
 * Mqtt Version Test
 */
public class MqttVersionTest {

    @Test
    public void test() {
        int l1 = 3;
        int l2 = 4;
        assert MqttVersion.fromProtocolNameAndLevel("MQIsdp", (byte) l1) == MqttVersion.MQTT_3_1;
        assert MqttVersion.fromProtocolNameAndLevel("MQTT", (byte) l2) == MqttVersion.MQTT_3_1_1;
    }
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/MqttGrantedQoS.java<|end_filename|>
/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.handler.codec.mqtt;

/**
 * Return Code of {@link io.netty.handler.codec.mqtt.MqttSubAckMessage}
 */
public enum MqttGrantedQoS {
    AT_MOST_ONCE(0),
    AT_LEAST_ONCE(1),
    EXACTLY_ONCE(2),
    NOT_GRANTED(0x80);

    private final int value;

    MqttGrantedQoS(int value) {
        this.value = value;
    }

    public static MqttGrantedQoS valueOf(int value) {
        for (MqttGrantedQoS r : values()) {
            if (r.value == value) {
                return r;
            }
        }
        throw new IllegalArgumentException("invalid granted QoS: " + value);
    }

    public int value() {
        return value;
    }
}


<|start_filename|>mithqtt-broker/src/main/java/com/github/longkerdandy/mithqtt/broker/cluster/BrokerClusterListenerImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.broker.cluster;

import com.github.longkerdandy.mithqtt.api.cluster.ClusterListener;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttAdditionalHeader;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.message.MqttSubscribePayloadGranted;
import com.github.longkerdandy.mithqtt.broker.session.SessionRegistry;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Broker Cluster Listener Implementation
 */
public class BrokerClusterListenerImpl implements ClusterListener {

    private static final Logger logger = LoggerFactory.getLogger(BrokerClusterListenerImpl.class);

    private final SessionRegistry registry;

    public BrokerClusterListenerImpl(SessionRegistry registry) {
        this.registry = registry;
    }

    @Override
    public void onConnect(Message<MqttConnectVariableHeader, MqttConnectPayload> msg) {
        // do nothing
    }

    @Override
    public void onSubscribe(Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> msg) {
        // do nothing
    }

    @Override
    public void onUnsubscribe(Message<MqttPacketIdVariableHeader, MqttUnsubscribePayload> msg) {
        // do nothing
    }

    @Override
    public void onPublish(Message<MqttPublishVariableHeader, MqttPublishPayload> msg) {
        MqttAdditionalHeader additionalHeader = msg.additionalHeader();
        MqttPublishVariableHeader variableHeader = msg.variableHeader();
        MqttPublishPayload payload = msg.payload();
        MqttMessage mqtt = new MqttPublishMessage(msg.fixedHeader(), variableHeader,
                (payload != null && payload.bytes() != null && payload.bytes().length > 0) ?
                        Unpooled.wrappedBuffer(payload.bytes()) : Unpooled.EMPTY_BUFFER);

        logger.trace("Send PUBLISH message to client {}", additionalHeader.clientId());

        this.registry.sendMessage(mqtt, additionalHeader.clientId(), variableHeader.packetId(), true);
    }

    @Override
    public void onDisconnect(Message<Void, Void> msg) {
        MqttAdditionalHeader additionalHeader = msg.additionalHeader();
        ChannelHandlerContext ctx = this.registry.removeSession(additionalHeader.clientId());
        if (ctx != null) {
            logger.trace("Try to disconnect connected client {}", additionalHeader.clientId());

            ctx.close();
        } else {
            logger.trace("Client {} no longer connected to this node", additionalHeader.clientId());
        }
    }
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/MqttConnectPayload.java<|end_filename|>
/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.handler.codec.mqtt;

import io.netty.util.internal.StringUtil;

/**
 * Payload of {@link MqttConnectMessage}
 */
public class MqttConnectPayload {

    protected String clientId;
    protected String willTopic;
    protected String willMessage;
    protected String userName;
    protected String password;

    private MqttConnectPayload() {
    }

    public MqttConnectPayload(
            String clientId,
            String willTopic,
            String willMessage,
            String userName,
            String password) {
        this.clientId = clientId;
        this.willTopic = willTopic;
        this.willMessage = willMessage;
        this.userName = userName;
        this.password = password;
    }

    public String clientId() {
        return clientId;
    }

    public String willTopic() {
        return willTopic;
    }

    public String willMessage() {
        return willMessage;
    }

    public String userName() {
        return userName;
    }

    public String password() {
        return password;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "clientId=" + clientId
                + ", willTopic=" + willTopic
                + ", willMessage=" + willMessage
                + ", userName=" + userName
                + ", password=" + password
                + ']';
    }
}


<|start_filename|>mithqtt-api/src/test/java/com/github/longkerdandy/mithqtt/api/message/MessageTest.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import com.github.longkerdandy.mithqtt.util.JSONs;
import io.netty.handler.codec.mqtt.*;
import org.junit.Test;

import java.io.IOException;
import java.util.Collections;

/**
 * Message Entity Test
 */
public class MessageTest {

    @Test
    public void serializationTest() throws IOException {
        Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> msg = new Message<>(
                new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_MOST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "Test_Client", "Test_User", "Test_Broker"),
                MqttPacketIdVariableHeader.from(12345),
                new MqttSubscribePayloadGranted(Collections.singletonList(new MqttTopicSubscriptionGranted("abc/+/g/h", MqttGrantedQoS.AT_MOST_ONCE)))
        );
        byte[] bytes = JSONs.Mapper.writeValueAsBytes(msg);
        Message m = JSONs.decodeMessage(bytes);
        assert m != null;
        assert m.fixedHeader() != null;
        assert m.fixedHeader.messageType() == MqttMessageType.SUBSCRIBE;
        assert !m.fixedHeader.dup();
        assert m.fixedHeader.qos() == MqttQoS.AT_MOST_ONCE;
        assert !m.fixedHeader.retain();
        assert m.additionalHeader() != null;
        assert m.additionalHeader().version() == MqttVersion.MQTT_3_1_1;
        assert m.additionalHeader().clientId().equals("Test_Client");
        assert m.additionalHeader().userName().equals("Test_User");
        assert m.additionalHeader().brokerId().equals("Test_Broker");
        assert m.variableHeader() != null;
        assert ((MqttPacketIdVariableHeader) m.variableHeader()).packetId() == 12345;
        assert m.payload() != null;
        assert ((MqttSubscribePayloadGranted) m.payload()).subscriptions() != null;
        assert ((MqttSubscribePayloadGranted) m.payload()).subscriptions().size() == 1;
        assert ((MqttSubscribePayloadGranted) m.payload()).subscriptions().get(0).topic().equals("abc/+/g/h");
        assert ((MqttSubscribePayloadGranted) m.payload()).subscriptions().get(0).grantedQos() == MqttGrantedQoS.AT_MOST_ONCE;
    }
}


<|start_filename|>mithqtt-broker/src/main/java/com/github/longkerdandy/mithqtt/broker/MqttBroker.java<|end_filename|>
package com.github.longkerdandy.mithqtt.broker;

import com.github.longkerdandy.mithqtt.api.auth.Authenticator;
import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.broker.cluster.BrokerClusterListenerFactoryImpl;
import com.github.longkerdandy.mithqtt.broker.handler.SyncStorageHandler;
import com.github.longkerdandy.mithqtt.broker.session.SessionRegistry;
import com.github.longkerdandy.mithqtt.broker.util.Validator;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;

/**
 * MQTT Bridge
 */
public class MqttBroker {

    private static final Logger logger = LoggerFactory.getLogger(MqttBroker.class);

    public static void main(String[] args) throws Exception {

        logger.debug("Starting MQTT broker ...");

        // load config
        logger.debug("Loading MQTT broker config files ...");
        PropertiesConfiguration brokerConfig;
        PropertiesConfiguration storageConfig;
        PropertiesConfiguration clusterConfig;
        PropertiesConfiguration authenticatorConfig;
        if (args.length >= 4) {
            brokerConfig = new PropertiesConfiguration(args[0]);
            storageConfig = new PropertiesConfiguration(args[1]);
            clusterConfig = new PropertiesConfiguration(args[2]);
            authenticatorConfig = new PropertiesConfiguration(args[3]);
        } else {
            brokerConfig = new PropertiesConfiguration("config/broker.properties");
            storageConfig = new PropertiesConfiguration("config/storage.properties");
            clusterConfig = new PropertiesConfiguration("config/cluster.properties");
            authenticatorConfig = new PropertiesConfiguration("config/authenticator.properties");
        }

        final String brokerId = brokerConfig.getString("broker.id");

        // validator
        logger.debug("Initializing validator ...");
        Validator validator = new Validator(brokerConfig);

        // session registry
        logger.debug("Initializing session registry ...");
        SessionRegistry registry = new SessionRegistry();

        // storage
        logger.debug("Initializing storage storage ...");
        SyncStorage storage = (SyncStorage) Class.forName(storageConfig.getString("storage.sync.class")).newInstance();
        storage.init(storageConfig);

        // cluster
        logger.debug("Initializing cluster ...");
        Cluster cluster = (Cluster) Class.forName(clusterConfig.getString("cluster.class")).newInstance();
        cluster.init(clusterConfig, new BrokerClusterListenerFactoryImpl(registry));

        // authenticator
        logger.debug("Initializing authenticator...");
        Authenticator authenticator = (Authenticator) Class.forName(authenticatorConfig.getString("authenticator.class")).newInstance();
        authenticator.init(authenticatorConfig);

        // broker
        final int keepAlive = brokerConfig.getInt("mqtt.keepalive.default");
        final int keepAliveMax = brokerConfig.getInt("mqtt.keepalive.max");
        final boolean ssl = brokerConfig.getBoolean("mqtt.ssl.enabled");
        final SslContext sslContext = ssl ? SslContextBuilder.forServer(new File(brokerConfig.getString("mqtt.ssl.certPath")), new File(brokerConfig.getString("mqtt.ssl.keyPath")), brokerConfig.getString("mqtt.ssl.keyPassword")).build() : null;
        final String host = brokerConfig.getString("mqtt.host");
        final int port = ssl ? brokerConfig.getInt("mqtt.ssl.port") : brokerConfig.getInt("mqtt.port");

        // tcp server
        logger.debug("Initializing tcp server ...");
        InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
        EventLoopGroup bossGroup = brokerConfig.getBoolean("netty.useEpoll") ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        EventLoopGroup workerGroup = brokerConfig.getBoolean("netty.useEpoll") ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        // EventLoopGroup handlerGroup = brokerConfig.getBoolean("netty.useEpoll") ? new EpollEventLoopGroup() : new NioEventLoopGroup();

        // shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                logger.debug("MQTT broker is shutting down ...");

                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
                cluster.destroy();
                authenticator.destroy();
                storage.destroy();

                logger.info("MQTT broker has been shut down.");
            }
        });

        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(brokerConfig.getBoolean("netty.useEpoll") ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        // ssl
                        if (ssl) {
                            p.addLast("ssl", sslContext.newHandler(ch.alloc()));
                        }
                        // idle
                        p.addFirst("idleHandler", new IdleStateHandler(0, 0, keepAlive));
                        // mqtt encoder & decoder
                        p.addLast("encoder", MqttEncoder.INSTANCE);
                        p.addLast("decoder", new MqttDecoder());
                        // logic handler
                        // p.addLast(handlerGroup, "logicHandler", new SyncRedisHandler(authenticator, cluster, storage, registry, validator, brokerId, keepAlive, keepAliveMax));
                        p.addLast("logicHandler", new SyncStorageHandler(authenticator, cluster, storage, registry, validator, brokerId, keepAlive, keepAliveMax));
                    }
                })
                .option(ChannelOption.SO_BACKLOG, brokerConfig.getInt("netty.soBacklog"))
                .childOption(ChannelOption.SO_KEEPALIVE, brokerConfig.getBoolean("netty.soKeepAlive"));

        // Bind and start to accept incoming connections.
        ChannelFuture f = b.bind(host, port).sync();

        logger.info("MQTT broker is up and running.");

        // Wait until the server socket is closed.
        // Do this to gracefully shut down the server.
        f.channel().closeFuture().sync();
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/resources/MqttUnsubscribeResource.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.resources;

import com.github.longkerdandy.mithqtt.api.auth.Authenticator;
import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttAdditionalHeader;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.http.entity.ErrorCode;
import com.github.longkerdandy.mithqtt.http.entity.ErrorEntity;
import com.github.longkerdandy.mithqtt.http.entity.ResultEntity;
import com.github.longkerdandy.mithqtt.http.exception.ValidateException;
import com.github.longkerdandy.mithqtt.http.util.Validator;
import com.github.longkerdandy.mithqtt.util.Topics;
import com.sun.security.auth.UserPrincipal;
import io.dropwizard.auth.Auth;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.security.PermitAll;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import java.util.List;

/**
 * MQTT Un-Subscribe related resource
 */
@Path("/clients/{clientId}/unsubscribe")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class MqttUnsubscribeResource extends AbstractResource {

    private static final Logger logger = LoggerFactory.getLogger(MqttSubscribeResource.class);

    public MqttUnsubscribeResource(String serverId, Validator validator, SyncStorage storage, Cluster cluster, Authenticator authenticator) {
        super(serverId, validator, storage, cluster, authenticator);
    }

    /**
     * Handle MQTT Un-Subscribe Request in RESTful style
     */
    @PermitAll
    @POST
    public ResultEntity<Boolean> unsubscribe(@PathParam("clientId") String clientId, @Auth UserPrincipal user, @QueryParam("protocol") @DefaultValue("4") byte protocol,
                                             @QueryParam("packetId") @DefaultValue("0") int packetId,
                                             List<String> topics) {
        String userName = user.getName();
        MqttVersion version = MqttVersion.fromProtocolLevel(protocol);

        // HTTP interface require valid Client Id
        if (!this.validator.isClientIdValid(clientId)) {
            logger.debug("Protocol violation: Client id {} not valid based on configuration", clientId);
            throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
        }

        // Validate Topic Filter based on configuration
        for (String topic : topics) {
            if (!this.validator.isTopicFilterValid(topic)) {
                logger.debug("Protocol violation: Client {} un-subscription {} is not valid based on configuration", clientId, topic);
                throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
            }
        }

        logger.debug("Message received: Received UNSUBSCRIBE message from client {} user {} topics {}", clientId, userName, ArrayUtils.toString(topics));

        // The Topic Filters (whether they contain wildcards or not) supplied in an UNSUBSCRIBE packet MUST be
        // compared character-by-character with the current set of Topic Filters held by the Server for the Client. If
        // any filter matches exactly then its owning Subscription is deleted, otherwise no additional processing
        // occurs
        // If a Server deletes a Subscription:
        // It MUST stop adding any new messages for delivery to the Client.
        //1 It MUST complete the delivery of any QoS 1 or QoS 2 messages which it has started to send to
        // the Client.
        // It MAY continue to deliver any existing messages buffered for delivery to the Client.
        topics.forEach(topic -> {
            logger.trace("Remove subscription: Remove client {} subscription with topic {}", clientId, topic);
            this.storage.removeSubscription(clientId, Topics.sanitize(topic));
        });

        // Pass message to 3rd party application
        Message<MqttPacketIdVariableHeader, MqttUnsubscribePayload> msg = new Message<>(
                new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(version, clientId, userName, null),
                MqttPacketIdVariableHeader.from(packetId),
                new MqttUnsubscribePayload(topics));
        this.cluster.sendToApplication(msg);

        return new ResultEntity<>(true);
    }
}


<|start_filename|>mithqtt-application-sample/src/main/java/com/github/longkerdandy/mithqtt/application/sample/SampleApp.java<|end_filename|>
package com.github.longkerdandy.mithqtt.application.sample;

import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.application.sample.cluster.ApplicationClusterListenerFactoryImpl;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Sample Application
 */
public class SampleApp {

    private static final Logger logger = LoggerFactory.getLogger(SampleApp.class);

    public static void main(String[] args) throws Exception {

        logger.debug("Starting Sample Application ...");

        // load config
        logger.debug("Loading Sample Application config files ...");
        PropertiesConfiguration clusterConfig;
        if (args.length >= 1 ) {
            clusterConfig = new PropertiesConfiguration(args[0]);
        } else {
            clusterConfig = new PropertiesConfiguration("config/cluster.properties");
        }

        // cluster
        logger.debug("Initializing cluster ...");
        Cluster cluster = (Cluster) Class.forName(clusterConfig.getString("cluster.class")).newInstance();
        cluster.init(clusterConfig, new ApplicationClusterListenerFactoryImpl());
    }
}


<|start_filename|>mithqtt-storage-redis/src/test/java/com/github/longkerdandy/mithqtt/storage/redis/sync/RedisSyncSingleStorageImplTest.java<|end_filename|>
package com.github.longkerdandy.mithqtt.storage.redis.sync;

import com.fasterxml.jackson.databind.JsonNode;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttAdditionalHeader;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.storage.sync.ConnectionState;
import com.github.longkerdandy.mithqtt.storage.redis.RedisKey;
import com.github.longkerdandy.mithqtt.util.Topics;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.MapConfiguration;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import static com.github.longkerdandy.mithqtt.storage.redis.util.JSONs.ObjectMapper;

/**
 * RedisSyncSingleStorageImpl Test
 */
public class RedisSyncSingleStorageImplTest {

    private static RedisSyncSingleStorageImpl redis;

    @BeforeClass
    public static void init() throws ConfigurationException {
        Map<String, Object> map = new HashMap<>();
        map.put("redis.type", "single");
        map.put("redis.address", "localhost");
        map.put("mqtt.inflight.queue.size", 3);
        map.put("mqtt.qos2.queue.size", 3);
        map.put("mqtt.retain.queue.size", 3);
        MapConfiguration config = new MapConfiguration(map);

        redis = new RedisSyncSingleStorageImpl();
        redis.init(config);
    }

    @AfterClass
    public static void destroy() {
        redis.destroy();
    }

    @After
    public void clear() {
        redis.server().flushdb();
    }

    @Test
    public void connectionTest() {

        assert redis.lock("client2", ConnectionState.CONNECTING);
        assert redis.lock("client4", ConnectionState.CONNECTING);
        assert !redis.lock("client3", ConnectionState.DISCONNECTING);
        assert !redis.lock("client5", ConnectionState.DISCONNECTING);
        assert !redis.release("client3", ConnectionState.DISCONNECTED);
        assert !redis.release("client5", ConnectionState.DISCONNECTED);

        assert redis.updateConnectedNode("client1", "node1", 30) == null;
        assert redis.updateConnectedNode("client2", "node1", 30) == null;
        assert redis.updateConnectedNode("client3", "node1", 30) == null;
        assert redis.updateConnectedNode("client4", "node1", 30) == null;
        assert redis.updateConnectedNode("client4", "node2", 30).equals("node1");   // overwrite
        assert redis.updateConnectedNode("client5", "node2", 30) == null;
        assert redis.updateConnectedNode("client5", "node2", 30).equals("node2");   // overwrite

        assert !redis.lock("client2", ConnectionState.CONNECTING);
        assert !redis.lock("client4", ConnectionState.CONNECTING);
        assert !redis.release("client2", ConnectionState.DISCONNECTED);
        assert !redis.release("client4", ConnectionState.DISCONNECTED);
        assert redis.release("client2", ConnectionState.CONNECTED);
        assert redis.release("client4", ConnectionState.CONNECTED);

        assert redis.getConnectedNode("client1").equals("node1");
        assert redis.getConnectedNode("client2").equals("node1");
        assert redis.getConnectedNode("client3").equals("node1");
        assert redis.getConnectedNode("client4").equals("node2");
        assert redis.getConnectedNode("client5").equals("node2");

        assert redis.lock("client2", ConnectionState.DISCONNECTING);
        assert redis.lock("client4", ConnectionState.DISCONNECTING);

        assert redis.removeConnectedNode("client2", "node1");
        assert !redis.removeConnectedNode("client4", "node1");   // not exist

        assert !redis.release("client2", ConnectionState.DISCONNECTED);   // removed
        assert redis.release("client4", ConnectionState.DISCONNECTED);

        assert redis.getConnectedNode("client2") == null;
        assert redis.getConnectedNode("client4").equals("node2");
    }

    @Test
    public void sessionExistTest() {
        assert redis.getSessionExist("client1") == -1;
        redis.updateSessionExist("client1", false);
        assert redis.getSessionExist("client1") == 0;
        redis.updateSessionExist("client1", true);
        assert redis.getSessionExist("client1") == 1;
        redis.removeSessionExist("client1");
        assert redis.getSessionExist("client1") == -1;
    }

    @Test
    public void packetIdTest() {
        assert redis.getNextPacketId("client1") == 1;
        assert redis.getNextPacketId("client1") == 2;
        assert redis.getNextPacketId("client1") == 3;

        redis.string().set(RedisKey.nextPacketId("client1"), "65533");

        assert redis.getNextPacketId("client1") == 65534;
        assert redis.getNextPacketId("client1") == 65535;
        assert redis.getNextPacketId("client1") == 1;
    }

    @Test
    @SuppressWarnings("unchecked")
    public void inFlightTest() throws IOException {
        String json = "{\"menu\": {\n" +
                "  \"id\": \"file\",\n" +
                "  \"value\": \"File\",\n" +
                "  \"popup\": {\n" +
                "    \"menuItem\": [\n" +
                "      {\"value\": \"New\", \"onclick\": \"CreateNewDoc()\"},\n" +
                "      {\"value\": \"Open\", \"onclick\": \"OpenDoc()\"},\n" +
                "      {\"value\": \"Close\", \"onclick\": \"CloseDoc()\"}\n" +
                "    ]\n" +
                "  }\n" +
                "}}";
        JsonNode jn = ObjectMapper.readTree(json);
        Message<MqttPublishVariableHeader, MqttPublishPayload> publish = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPublishVariableHeader.from("menuTopic", 12345),
                new MqttPublishPayload(ObjectMapper.writeValueAsBytes(jn)));

        redis.addInFlightMessage("client1", 12345, publish, false);
        publish = redis.getInFlightMessage("client1", 12345);

        assert publish.fixedHeader().messageType() == MqttMessageType.PUBLISH;
        assert !publish.fixedHeader().dup();
        assert publish.fixedHeader().qos() == MqttQoS.AT_LEAST_ONCE;
        assert !publish.fixedHeader().retain();
        assert publish.additionalHeader().version() == MqttVersion.MQTT_3_1_1;
        assert publish.additionalHeader().clientId().equals("client1");
        assert publish.additionalHeader().userName().equals("user1");
        assert publish.variableHeader().topicName().equals("menuTopic");
        assert publish.variableHeader().packetId() == 12345;
        jn = ObjectMapper.readTree(publish.payload().bytes());
        assert jn.get("menu").get("id").textValue().endsWith("file");
        assert jn.get("menu").get("value").textValue().endsWith("File");
        assert jn.get("menu").get("popup").get("menuItem").get(0).get("value").textValue().equals("New");
        assert jn.get("menu").get("popup").get("menuItem").get(0).get("onclick").textValue().equals("CreateNewDoc()");
        assert jn.get("menu").get("popup").get("menuItem").get(1).get("value").textValue().equals("Open");
        assert jn.get("menu").get("popup").get("menuItem").get(1).get("onclick").textValue().equals("OpenDoc()");
        assert jn.get("menu").get("popup").get("menuItem").get(2).get("value").textValue().equals("Close");
        assert jn.get("menu").get("popup").get("menuItem").get(2).get("onclick").textValue().equals("CloseDoc()");

        publish = redis.getAllInFlightMessages("client1").get(0);
        assert publish.variableHeader().packetId() == 12345;

        redis.removeInFlightMessage("client1", 12345);

        assert redis.getInFlightMessage("client1", 12345) == null;
        assert redis.getAllInFlightMessages("client1").size() == 0;

        Message<MqttPacketIdVariableHeader, Void> pubrel = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPacketIdVariableHeader.from(10000),
                null
        );
        redis.addInFlightMessage("client1", 10000, pubrel, false);

        pubrel = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPacketIdVariableHeader.from(10001),
                null
        );
        redis.addInFlightMessage("client1", 10001, pubrel, false);

        pubrel = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPacketIdVariableHeader.from(10002),
                null
        );
        redis.addInFlightMessage("client1", 10002, pubrel, false);

        assert redis.getAllInFlightMessages("client1").size() == 3;
        pubrel = redis.getAllInFlightMessages("client1").get(1);

        assert pubrel.fixedHeader().messageType() == MqttMessageType.PUBREL;
        assert !pubrel.fixedHeader().dup();
        assert pubrel.fixedHeader().qos() == MqttQoS.AT_LEAST_ONCE;
        assert !pubrel.fixedHeader().retain();
        assert pubrel.additionalHeader().version() == MqttVersion.MQTT_3_1_1;
        assert pubrel.additionalHeader().clientId().equals("client1");
        assert pubrel.additionalHeader().userName().equals("user1");
        assert pubrel.variableHeader().packetId() == 10001;

        pubrel = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPacketIdVariableHeader.from(10003),
                null
        );
        redis.addInFlightMessage("client1", 10003, pubrel, false);

        assert redis.getAllInFlightMessages("client1").size() == 3;
        assert redis.getInFlightMessage("client1", 10000) == null;

        redis.removeAllInFlightMessage("client1");

        assert redis.getInFlightMessage("client1", 10001) == null;
        assert redis.getInFlightMessage("client1", 10002) == null;
        assert redis.getInFlightMessage("client1", 10003) == null;
        assert redis.getAllInFlightMessages("client1").size() == 0;
    }

    @Test
    public void qos2Test() {
        assert redis.addQoS2MessageId("client1", 10000);
        assert redis.addQoS2MessageId("client1", 10001);
        assert redis.addQoS2MessageId("client1", 10002);
        assert !redis.addQoS2MessageId("client1", 10000);

        assert redis.removeQoS2MessageId("client1", 10000);
        assert redis.removeQoS2MessageId("client1", 10001);
        assert redis.removeQoS2MessageId("client1", 10002);
        assert !redis.removeQoS2MessageId("client1", 10001);

        assert redis.addQoS2MessageId("client1", 10003);
        assert redis.addQoS2MessageId("client1", 10004);
        assert redis.addQoS2MessageId("client1", 10005);
        assert redis.addQoS2MessageId("client1", 10006);

        assert !redis.removeQoS2MessageId("client1", 10003);

        redis.removeAllQoS2MessageId("client1");

        assert !redis.removeQoS2MessageId("client1", 10004);
        assert !redis.removeQoS2MessageId("client1", 10005);
        assert !redis.removeQoS2MessageId("client1", 10006);
    }

    @Test
    public void subscriptionTest() {
        redis.updateSubscription("client1", Topics.sanitizeTopicFilter("a/+/e"), MqttQoS.AT_MOST_ONCE);
        redis.updateSubscription("client1", Topics.sanitizeTopicFilter("a/+"), MqttQoS.AT_LEAST_ONCE);
        redis.updateSubscription("client1", Topics.sanitizeTopicName("a/c/e"), MqttQoS.EXACTLY_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicFilter("a/#"), MqttQoS.AT_MOST_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicFilter("a/+"), MqttQoS.AT_LEAST_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicName("a/c/e"), MqttQoS.EXACTLY_ONCE);

        assert redis.getClientSubscriptions("client1").get("a/+/e/" + Topics.END) == MqttQoS.AT_MOST_ONCE;
        assert redis.getClientSubscriptions("client1").get("a/+/" + Topics.END) == MqttQoS.AT_LEAST_ONCE;
        assert redis.getClientSubscriptions("client1").get("a/c/e/" + Topics.END) == MqttQoS.EXACTLY_ONCE;
        assert redis.getClientSubscriptions("client2").get("a/#/" + Topics.END) == MqttQoS.AT_MOST_ONCE;
        assert redis.getClientSubscriptions("client2").get("a/+/" + Topics.END) == MqttQoS.AT_LEAST_ONCE;
        assert redis.getClientSubscriptions("client2").get("a/c/e/" + Topics.END) == MqttQoS.EXACTLY_ONCE;

        assert redis.getTopicSubscriptions(Topics.sanitizeTopicFilter("a/+/e")).get("client1") == MqttQoS.AT_MOST_ONCE;
        assert redis.getTopicSubscriptions(Topics.sanitizeTopicFilter("a/+")).get("client1") == MqttQoS.AT_LEAST_ONCE;
        assert redis.getTopicSubscriptions(Topics.sanitizeTopicFilter("a/+")).get("client2") == MqttQoS.AT_LEAST_ONCE;
        assert redis.getTopicSubscriptions(Topics.sanitizeTopicName("a/c/e")).get("client1") == MqttQoS.EXACTLY_ONCE;
        assert redis.getTopicSubscriptions(Topics.sanitizeTopicName("a/c/e")).get("client2") == MqttQoS.EXACTLY_ONCE;
        assert redis.getTopicSubscriptions(Topics.sanitizeTopicFilter("a/#")).get("client2") == MqttQoS.AT_MOST_ONCE;

        redis.removeSubscription("client1", Topics.sanitizeTopicFilter("a/+"));

        assert !redis.getTopicSubscriptions(Topics.sanitizeTopicFilter("a/+")).containsKey("client1");
        assert !redis.getClientSubscriptions("client1").containsKey("a/+/" + Topics.END);
    }

    @Test
    public void matchTopicFilterTest() {
        redis.updateSubscription("client1", Topics.sanitizeTopicFilter("a/+/e"), MqttQoS.AT_MOST_ONCE);
        redis.updateSubscription("client1", Topics.sanitizeTopicFilter("a/+"), MqttQoS.AT_LEAST_ONCE);
        redis.updateSubscription("client1", Topics.sanitizeTopicFilter("a/c/f/#"), MqttQoS.EXACTLY_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicFilter("a/#"), MqttQoS.AT_MOST_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicFilter("a/c/+/+"), MqttQoS.AT_LEAST_ONCE);
        redis.updateSubscription("client2", Topics.sanitizeTopicFilter("a/d/#"), MqttQoS.EXACTLY_ONCE);
        redis.updateSubscription("client3", Topics.sanitizeTopicName("a/b/c/d"), MqttQoS.AT_LEAST_ONCE);

        Map<String, MqttQoS> result = new HashMap<>();
        redis.getMatchSubscriptions(Topics.sanitizeTopicName("a/c/f"), result);
        assert result.get("client1") == MqttQoS.EXACTLY_ONCE;
        assert result.get("client2") == MqttQoS.AT_MOST_ONCE;
        assert !result.containsKey("client3");

        result.clear();
        redis.getMatchSubscriptions(Topics.sanitizeTopicName("a/d/e"), result);
        assert result.get("client1") == MqttQoS.AT_MOST_ONCE;
        assert result.get("client2") == MqttQoS.EXACTLY_ONCE;
        assert !result.containsKey("client3");

        result.clear();
        redis.getMatchSubscriptions(Topics.sanitizeTopicName("a/b/c/d"), result);
        assert !result.containsKey("client1");
        assert result.get("client2") == MqttQoS.AT_MOST_ONCE;
        assert result.get("client3") == MqttQoS.AT_LEAST_ONCE;
    }

    @Test
    public void retainTest() throws IOException {
        String json = "{\"menu\": {\n" +
                "  \"id\": \"file\",\n" +
                "  \"value\": \"File\",\n" +
                "  \"popup\": {\n" +
                "    \"menuItem\": [\n" +
                "      {\"value\": \"New\", \"onclick\": \"CreateNewDoc()\"},\n" +
                "      {\"value\": \"Open\", \"onclick\": \"OpenDoc()\"},\n" +
                "      {\"value\": \"Close\", \"onclick\": \"CloseDoc()\"}\n" +
                "    ]\n" +
                "  }\n" +
                "}}";
        JsonNode jn = ObjectMapper.readTree(json);
        Message<MqttPublishVariableHeader, MqttPublishPayload> publish = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPublishVariableHeader.from("menuTopic", 12345),
                new MqttPublishPayload(ObjectMapper.writeValueAsBytes(jn)));

        redis.addRetainMessage(Topics.sanitize("a/b/c/d"), publish);
        publish = redis.getMatchRetainMessages(Topics.sanitize("a/b/c/d")).get(0);

        assert publish.fixedHeader().messageType() == MqttMessageType.PUBLISH;
        assert !publish.fixedHeader().dup();
        assert publish.fixedHeader().qos() == MqttQoS.AT_LEAST_ONCE;
        assert !publish.fixedHeader().retain();
        assert publish.additionalHeader().version() == MqttVersion.MQTT_3_1_1;
        assert publish.additionalHeader().userName().equals("user1");
        assert publish.variableHeader().topicName().equals("menuTopic");
        jn = ObjectMapper.readTree(publish.payload().bytes());
        assert jn.get("menu").get("id").textValue().endsWith("file");
        assert jn.get("menu").get("value").textValue().endsWith("File");
        assert jn.get("menu").get("popup").get("menuItem").get(0).get("value").textValue().equals("New");
        assert jn.get("menu").get("popup").get("menuItem").get(0).get("onclick").textValue().equals("CreateNewDoc()");
        assert jn.get("menu").get("popup").get("menuItem").get(1).get("value").textValue().equals("Open");
        assert jn.get("menu").get("popup").get("menuItem").get(1).get("onclick").textValue().equals("OpenDoc()");
        assert jn.get("menu").get("popup").get("menuItem").get(2).get("value").textValue().equals("Close");
        assert jn.get("menu").get("popup").get("menuItem").get(2).get("onclick").textValue().equals("CloseDoc()");

        redis.removeAllRetainMessage(Topics.sanitize("a/b/c/d"));
        assert redis.getMatchRetainMessages(Topics.sanitize("a/b/c/d")).size() == 0;

        redis.addRetainMessage(Topics.sanitize("a/b/c/d"), publish);
        redis.addRetainMessage(Topics.sanitize("a/b/c/d"), publish);

        assert redis.getMatchRetainMessages(Topics.sanitize("a/b/c/d")).size() == 2;

        redis.addRetainMessage(Topics.sanitize("a/b/c/d"), publish);
        redis.addRetainMessage(Topics.sanitize("a/b/c/d"), publish);

        assert redis.getMatchRetainMessages(Topics.sanitize("a/b/c/d")).size() == 3;
    }

    @Test
    public void matchRetainTest() {
        Message<MqttPublishVariableHeader, MqttPublishPayload> p1 = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client1", "user1", "broker1"),
                MqttPublishVariableHeader.from("foo/bar", 100),
                new MqttPublishPayload("Hello Retain 1".getBytes()));
        Message<MqttPublishVariableHeader, MqttPublishPayload> p2 = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client2", "user2", "broker2"),
                MqttPublishVariableHeader.from("foo/bar/zoo", 200),
                new MqttPublishPayload("Hello Retain 2".getBytes()));
        Message<MqttPublishVariableHeader, MqttPublishPayload> p3 = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client3", "user3", "broker3"),
                MqttPublishVariableHeader.from("foo/bar/zoo/rar", 300),
                new MqttPublishPayload("Hello Retain 3".getBytes()));
        Message<MqttPublishVariableHeader, MqttPublishPayload> p4 = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client4", "user4", "broker4"),
                MqttPublishVariableHeader.from("foo/moo", 400),
                new MqttPublishPayload("Hello Retain 4".getBytes()));
        Message<MqttPublishVariableHeader, MqttPublishPayload> p5 = new Message<>(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, "client5", "user5", "broker5"),
                MqttPublishVariableHeader.from("foo/moo/zoo", 500),
                new MqttPublishPayload("Hello Retain 5".getBytes()));

        redis.addRetainMessage(Topics.sanitize("foo/bar"), p1);
        redis.addRetainMessage(Topics.sanitize("foo/bar/zoo"), p2);
        redis.addRetainMessage(Topics.sanitize("foo/bar/zoo/rar"), p3);
        redis.addRetainMessage(Topics.sanitize("foo/moo"), p4);
        redis.addRetainMessage(Topics.sanitize("foo/moo/zoo"), p5);

        assert redis.getMatchRetainMessages(Topics.sanitize("foo/+")).size() == 2;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/bar/+")).size() == 1;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/#")).size() == 5;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/bar/#")).size() == 3;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/bar/zoo/#")).size() == 2;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/bar/zoo/rar/#")).size() == 1;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/bar/zoo/+")).size() == 1;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/zoo/#")).size() == 0;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/+/#")).size() == 5;
        assert redis.getMatchRetainMessages(Topics.sanitize("foo/+/zoo/#")).size() == 3;
        assert redis.getMatchRetainMessages(Topics.sanitize("#")).size() == 5;
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/util/JSONs.java<|end_filename|>
package com.github.longkerdandy.mithqtt.util;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.*;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.message.MqttSubscribePayloadGranted;
import io.netty.handler.codec.mqtt.*;

import java.io.IOException;

/**
 * JSON Utils
 */
public class JSONs {

    // Global JSON ObjectMapper
    public static final ObjectMapper Mapper = new ObjectMapper();

    static {
        Mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        Mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
        Mapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
        Mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        Mapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
    }

    private JSONs() {
    }

    /**
     * Decode bytes (json data) to Message
     */
    public static Message decodeMessage(byte[] data) throws IOException {
        JavaType type = Mapper.getTypeFactory().constructParametricType(Message.class, JsonNode.class, JsonNode.class);
        Message<JsonNode, JsonNode> m = Mapper.readValue(data, type);
        switch (m.fixedHeader().messageType()) {
            case CONNECT:
                MqttConnectVariableHeader cv = Mapper.treeToValue(m.variableHeader(), MqttConnectVariableHeader.class);
                MqttConnectPayload cp = Mapper.treeToValue(m.payload(), MqttConnectPayload.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), cv, cp);
            case CONNACK:
                MqttConnAckVariableHeader cav = Mapper.treeToValue(m.variableHeader(), MqttConnAckVariableHeader.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), cav, null);
            case SUBSCRIBE:
                MqttPacketIdVariableHeader sv = Mapper.treeToValue(m.variableHeader(), MqttPacketIdVariableHeader.class);
                MqttSubscribePayloadGranted sp = Mapper.treeToValue(m.payload(), MqttSubscribePayloadGranted.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), sv, sp);
            case SUBACK:
                MqttPacketIdVariableHeader sav = Mapper.treeToValue(m.variableHeader(), MqttPacketIdVariableHeader.class);
                MqttSubAckPayload sap = Mapper.treeToValue(m.payload(), MqttSubAckPayload.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), sav, sap);
            case UNSUBSCRIBE:
                MqttPacketIdVariableHeader uv = Mapper.treeToValue(m.variableHeader(), MqttPacketIdVariableHeader.class);
                MqttUnsubscribePayload up = Mapper.treeToValue(m.payload(), MqttUnsubscribePayload.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), uv, up);
            case PUBLISH:
                MqttPublishVariableHeader pv = Mapper.treeToValue(m.variableHeader(), MqttPublishVariableHeader.class);
                MqttPublishPayload pp = Mapper.treeToValue(m.payload(), MqttPublishPayload.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), pv, pp);
            case UNSUBACK:
            case PUBACK:
            case PUBREC:
            case PUBREL:
            case PUBCOMP:
                MqttPacketIdVariableHeader iv = Mapper.treeToValue(m.variableHeader(), MqttPacketIdVariableHeader.class);
                return new Message<>(m.fixedHeader(), m.additionalHeader(), iv, null);
            case PINGREQ:
            case PINGRESP:
            case DISCONNECT:
                return m;
            default:
                return null;
        }
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/MqttHttp.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.github.longkerdandy.mithqtt.api.auth.Authenticator;
import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.http.oauth.OAuthAuthenticator;
import com.github.longkerdandy.mithqtt.http.resources.MqttPublishResource;
import com.github.longkerdandy.mithqtt.http.resources.MqttSubscribeResource;
import com.github.longkerdandy.mithqtt.http.resources.MqttUnsubscribeResource;
import com.github.longkerdandy.mithqtt.http.util.Validator;
import com.sun.security.auth.UserPrincipal;
import io.dropwizard.Application;
import io.dropwizard.auth.AuthDynamicFeature;
import io.dropwizard.auth.AuthValueFactoryProvider;
import io.dropwizard.auth.PermitAllAuthorizer;
import io.dropwizard.auth.oauth.OAuthCredentialAuthFilter;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.lang3.ArrayUtils;
import org.glassfish.jersey.server.filter.RolesAllowedDynamicFeature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * MQTT Http Interface
 */
public class MqttHttp extends Application<MqttHttpConfiguration> {

    private static final Logger logger = LoggerFactory.getLogger(MqttHttp.class);

    private static PropertiesConfiguration storageConfig;
    private static PropertiesConfiguration clusterConfig;
    private static PropertiesConfiguration authenticatorConfig;

    public static void main(String[] args) throws Exception {
        logger.debug("Starting MQTT http ...");

        // load config
        logger.debug("Loading MQTT http config files ...");
        if (args.length >= 3) {
            storageConfig = new PropertiesConfiguration(args[0]);
            clusterConfig = new PropertiesConfiguration(args[1]);
            authenticatorConfig = new PropertiesConfiguration(args[2]);

            if (args.length >= 4) {
                args = ArrayUtils.subarray(args, 4, args.length);
            } else {
                args = new String[]{};
            }
        } else {
            storageConfig = new PropertiesConfiguration("config/storage.properties");
            clusterConfig = new PropertiesConfiguration("config/cluster.properties");
            authenticatorConfig = new PropertiesConfiguration("config/authenticator.properties");
        }

        new MqttHttp().run(args);
    }

    @Override
    public void run(MqttHttpConfiguration configuration, Environment environment) throws Exception {
        // validator
        logger.debug("Initializing validator ...");
        Validator validator = new Validator(configuration);

        // storage
        SyncStorage storage = (SyncStorage) Class.forName(storageConfig.getString("storage.sync.class")).newInstance();
        environment.lifecycle().manage(new Managed() {
            @Override
            public void start() throws Exception {
                logger.debug("Initializing storage storage ...");
                storage.init(storageConfig);
            }

            @Override
            public void stop() throws Exception {
                logger.debug("Destroying storage storage ...");
                storage.destroy();
            }
        });

        // authenticator
        Authenticator authenticator = (Authenticator) Class.forName(authenticatorConfig.getString("authenticator.class")).newInstance();
        environment.lifecycle().manage(new Managed() {
            @Override
            public void start() throws Exception {
                logger.debug("Initializing authenticator ...");
                authenticator.init(authenticatorConfig);
            }

            @Override
            public void stop() throws Exception {
                logger.debug("Destroying authenticator ...");
                authenticator.destroy();
            }
        });

        // cluster
        Cluster cluster = (Cluster) Class.forName(clusterConfig.getString("cluster.class")).newInstance();
        environment.lifecycle().manage(new Managed() {
            @Override
            public void start() throws Exception {
                logger.debug("Initializing cluster ...");
                cluster.init(clusterConfig, null);
            }

            @Override
            public void stop() throws Exception {
                logger.debug("Destroying cluster ...");
                cluster.destroy();
            }
        });

        // OAuth
        environment.jersey().register(new AuthDynamicFeature(
                new OAuthCredentialAuthFilter.Builder<UserPrincipal>()
                        .setAuthenticator(new OAuthAuthenticator(authenticator))
                        .setAuthorizer(new PermitAllAuthorizer<>())
                        .setPrefix("Bearer")
                        .buildAuthFilter()));
        environment.jersey().register(RolesAllowedDynamicFeature.class);
        environment.jersey().register(new AuthValueFactoryProvider.Binder<>(UserPrincipal.class));

        // register resources
        environment.jersey().register(new MqttPublishResource(configuration.getServerId(), validator, storage, cluster, authenticator));
        environment.jersey().register(new MqttSubscribeResource(configuration.getServerId(), validator, storage, cluster, authenticator));
        environment.jersey().register(new MqttUnsubscribeResource(configuration.getServerId(), validator, storage, cluster, authenticator));

        // config jackson
        environment.getObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        environment.getObjectMapper().configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
        environment.getObjectMapper().configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
        environment.getObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);
    }
}


<|start_filename|>mithqtt-cluster-nats/src/main/java/com/github/longkerdandy/mithqtt/cluster/nats/NATSClusterImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.cluster.nats;

import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.cluster.ClusterException;
import com.github.longkerdandy.mithqtt.api.cluster.ClusterListener;
import com.github.longkerdandy.mithqtt.api.cluster.ClusterListenerFactory;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.util.JSONs;
import io.nats.client.Connection;
import io.nats.client.ConnectionFactory;
import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import static io.nats.client.ConnectionFactory.*;

/**
 * NATS Server based Cluster
 */
public class NATSClusterImpl implements Cluster {

    private static final Logger logger = LoggerFactory.getLogger(NATSClusterImpl.class);

    // topics
    private String LISTENER_TOPIC;
    private String BROKER_TOPIC_PREFIX;
    private String APPLICATION_TOPIC;

    // active connection to NATS Server
    private Connection conn;

    @SuppressWarnings("unchecked")
    @Override
    public void init(AbstractConfiguration config, ClusterListenerFactory factory) throws ClusterException {
        logger.trace("Loading cluster configurations ...");

        // Topics for broker and application
        LISTENER_TOPIC = config.getString("listener.topic");
        BROKER_TOPIC_PREFIX = config.getString("broker.topic.prefix");
        APPLICATION_TOPIC = config.getString("application.topic");

        // Setup options to include all servers in the cluster
        ConnectionFactory cf = new ConnectionFactory();
        cf.setServers(config.getString("nats.servers", DEFAULT_URL).split(","));

        // Set NATS configurations
        cf.setMaxReconnect(config.getInt("nats.maxReconnect", DEFAULT_MAX_RECONNECT));
        cf.setReconnectWait(config.getInt("nats.reconnectWait", DEFAULT_RECONNECT_WAIT));
        cf.setReconnectBufSize(config.getInt("nats.reconnectBufSize", DEFAULT_RECONNECT_BUF_SIZE));
        cf.setConnectionTimeout(config.getInt("nats.connectionTimeout", DEFAULT_TIMEOUT));
        cf.setPingInterval(config.getInt("nats.pingInterval", DEFAULT_PING_INTERVAL));
        cf.setMaxPingsOut(config.getInt("nats.maxPingsOut", DEFAULT_MAX_PINGS_OUT));
        cf.setMaxPendingMsgs(config.getInt("nats.maxPendingMsgs", DEFAULT_MAX_PENDING_MSGS));
        cf.setMaxPendingBytes(config.getInt("nats.maxPendingBytes", DEFAULT_MAX_PENDING_BYTES));

        // Optionally disable randomization of the server pool
        cf.setNoRandomize(config.getBoolean("nats.noRandomize", false));

        logger.trace("Creating connection with NATS servers ...");

        // Create connection to the NATS servers
        try {
            this.conn = cf.createConnection();
        } catch (IOException | TimeoutException e) {
            throw new ClusterException(e);
        }

        if (StringUtils.isNotBlank(LISTENER_TOPIC) && factory != null) {
            logger.trace("Subscribe to topic {} ...", LISTENER_TOPIC);

            this.conn.subscribeAsync(LISTENER_TOPIC, msg -> {
                try {
                    logger.trace("Received message from NATS topic {}", msg.getSubject());

                    // event listener
                    ClusterListener listener = factory.newListener();

                    // decode message
                    Message m = JSONs.decodeMessage(msg.getData());

                    // handle message
                    if (m != null) {
                        logger.debug("Cluster received: Received {} message for client {}", m.fixedHeader().messageType(), m.additionalHeader().clientId());
                        switch (m.fixedHeader().messageType()) {
                            case CONNECT:
                                listener.onConnect(m);
                                break;
                            case SUBSCRIBE:
                                listener.onSubscribe(m);
                                break;
                            case UNSUBSCRIBE:
                                listener.onUnsubscribe(m);
                                break;
                            case PUBLISH:
                                listener.onPublish(m);
                                break;
                            case DISCONNECT:
                                listener.onDisconnect(m);
                                break;
                            default:
                                logger.warn("Cluster Error: Received message with unknown type {}", m.fixedHeader().messageType());
                        }
                    }
                } catch (IOException e) {
                    logger.warn("Cluster Error: Error when decoding or handling the message", e);
                }
            });
        }
    }

    @Override
    public void destroy() {
        logger.trace("Closing connection with NATS servers ...");

        if (this.conn != null) this.conn.close();
    }

    @Override
    public void sendToBroker(String brokerId, Message message) {
        String brokerTopic = BROKER_TOPIC_PREFIX + "." + brokerId;
        try {
            this.conn.publish(brokerTopic, JSONs.Mapper.writeValueAsBytes(message));
        } catch (IOException e) {
            logger.warn("Cluster Error: Failed to send message {} to topic {}: ", message.fixedHeader().messageType(), brokerTopic, e);
        }
    }

    @Override
    public void sendToApplication(Message message) {
        try {
            this.conn.publish(APPLICATION_TOPIC, JSONs.Mapper.writeValueAsBytes(message));
        } catch (IOException e) {
            logger.warn("Cluster Error: Failed to send message {} to topic {}: ", message.fixedHeader().messageType(), APPLICATION_TOPIC, e);
        }
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/cluster/ClusterListener.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.cluster;

import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.message.MqttSubscribePayloadGranted;
import io.netty.handler.codec.mqtt.*;

/**
 * Cluster Event Listener
 */
public interface ClusterListener {

    /**
     * Received Connect Message Event
     *
     * @param msg Connect Message
     */
    void onConnect(Message<MqttConnectVariableHeader, MqttConnectPayload> msg);

    /**
     * Received Subscribe Message Event
     *
     * @param msg Subscribe Message
     */
    void onSubscribe(Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> msg);

    /**
     * Received Unsubscribe Message Event
     *
     * @param msg Unsubscribe Message
     */
    void onUnsubscribe(Message<MqttPacketIdVariableHeader, MqttUnsubscribePayload> msg);

    /**
     * Received Publish Message Event
     *
     * @param msg Publish Message
     */
    void onPublish(Message<MqttPublishVariableHeader, MqttPublishPayload> msg);

    /**
     * Received Disconnect Message Event
     *
     * @param msg Disconnect Message
     */
    void onDisconnect(Message<Void, Void> msg);
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/message/Message.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.internal.StringUtil;

import java.util.ArrayList;
import java.util.List;

/**
 * Message
 */
public class Message<V, P> {

    protected MqttFixedHeader fixedHeader;
    protected MqttAdditionalHeader additionalHeader;
    protected V variableHeader;
    protected P payload;

    private Message() {
    }

    public Message(MqttFixedHeader fixedHeader, MqttAdditionalHeader additionalHeader, V variableHeader, P payload) {
        this.fixedHeader = fixedHeader;
        this.additionalHeader = additionalHeader;
        this.variableHeader = variableHeader;
        this.payload = payload;
    }

    public static Message fromMqttMessage(MqttMessage msg, MqttVersion version, String clientId, String userName, String brokerId) {
        switch (msg.fixedHeader().messageType()) {
            case CONNECT:
            case CONNACK:
            case SUBACK:
            case UNSUBSCRIBE:
            case UNSUBACK:
            case PUBACK:
            case PUBREC:
            case PUBREL:
            case PUBCOMP:
            case PINGREQ:
            case PINGRESP:
            case DISCONNECT:
                return new Message<>(msg.fixedHeader(), new MqttAdditionalHeader(version, clientId, userName, brokerId), msg.variableHeader(), msg.payload());
            default:
                throw new IllegalArgumentException("unknown message type " + msg.fixedHeader().messageType());
        }
    }

    public static Message<MqttPublishVariableHeader, MqttPublishPayload> fromMqttMessage(MqttPublishMessage msg, MqttVersion version, String clientId, String userName, String brokerId) {
        // forge bytes payload
        byte[] bytes = new byte[0];
        if (msg.payload() != null && msg.payload().readableBytes() > 0) {
            ByteBuf buf = msg.payload().duplicate();
            bytes = new byte[buf.readableBytes()];
            buf.readBytes(bytes);
        }
        return new Message<>(msg.fixedHeader(), new MqttAdditionalHeader(version, clientId, userName, brokerId), msg.variableHeader(), new MqttPublishPayload(bytes));
    }

    public static Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> fromMqttMessage(MqttSubscribeMessage msg, List<MqttGrantedQoS> grantedQoSes, MqttVersion version, String clientId, String userName, String brokerId) {
        // forge topic subscriptions
        List<MqttTopicSubscriptionGranted> subscriptions = new ArrayList<>();
        for (int i = 0; i < msg.payload().subscriptions().size(); i++) {
            MqttTopicSubscriptionGranted subscription = new MqttTopicSubscriptionGranted(msg.payload().subscriptions().get(i).topic(), grantedQoSes.get(i));
            subscriptions.add(subscription);
        }
        return new Message<>(msg.fixedHeader(), new MqttAdditionalHeader(version, clientId, userName, brokerId), msg.variableHeader(), new MqttSubscribePayloadGranted(subscriptions));
    }

    public MqttFixedHeader fixedHeader() {
        return fixedHeader;
    }

    public MqttAdditionalHeader additionalHeader() {
        return additionalHeader;
    }

    public V variableHeader() {
        return variableHeader;
    }

    public P payload() {
        return payload;
    }

    public MqttMessage toMqttMessage() {
        switch (fixedHeader.messageType()) {
            case CONNECT:
            case CONNACK:
            case SUBACK:
            case UNSUBSCRIBE:
                return MqttMessageFactory.newMessage(fixedHeader, variableHeader, payload);
            case UNSUBACK:
            case PUBACK:
            case PUBREC:
            case PUBREL:
            case PUBCOMP:
                return MqttMessageFactory.newMessage(fixedHeader, variableHeader, null);
            case PINGREQ:
            case PINGRESP:
            case DISCONNECT:
                return MqttMessageFactory.newMessage(fixedHeader, null, null);
            case SUBSCRIBE:
                List<MqttTopicSubscription> subscriptions = new ArrayList<>();
                ((MqttSubscribePayloadGranted) payload).subscriptions().forEach(s -> {
                    if (s.grantedQos != MqttGrantedQoS.NOT_GRANTED) {
                        subscriptions.add(new MqttTopicSubscription(s.topic, MqttQoS.valueOf(s.grantedQos.value())));
                    }
                });
                return MqttMessageFactory.newMessage(fixedHeader, variableHeader, new MqttSubscribePayload(subscriptions));
            case PUBLISH:
                MqttPublishPayload p = (MqttPublishPayload) payload;
                return MqttMessageFactory.newMessage(fixedHeader, variableHeader,
                        (p != null && p.bytes != null && p.bytes.length > 0) ? Unpooled.wrappedBuffer(p.bytes) : Unpooled.EMPTY_BUFFER);
            default:
                throw new IllegalStateException("unknown message type " + fixedHeader.messageType());
        }
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "fixedHeader=" + (fixedHeader != null ? fixedHeader.toString() : "")
                + ", additionalHeader=" + (additionalHeader != null ? additionalHeader.toString() : "")
                + ", variableHeader=" + (variableHeader != null ? variableHeader.toString() : "")
                + ", payload=" + (payload != null ? payload.toString() : "")
                + ']';
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/cluster/ClusterListenerFactory.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.cluster;

/**
 * Cluster Listener Factory
 */
public interface ClusterListenerFactory {

    /**
     * Create a new ClusterListener
     *
     * @return ClusterListener
     */
    ClusterListener newListener();
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/message/MqttAdditionalHeader.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.internal.StringUtil;

/**
 * Mqtt Additional Header
 */
public class MqttAdditionalHeader {

    protected MqttVersion version;
    protected String clientId;
    protected String userName;
    protected String brokerId;

    private MqttAdditionalHeader() {
    }

    public MqttAdditionalHeader(
            MqttVersion version,
            String clientId,
            String userName,
            String brokerId) {
        this.version = version;
        this.clientId = clientId;
        this.userName = userName;
        this.brokerId = brokerId;
    }

    public MqttVersion version() {
        return version;
    }

    public String clientId() {
        return clientId;
    }

    public String userName() {
        return userName;
    }

    public String brokerId() {
        return brokerId;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "version=" + version
                + ", clientId=" + clientId
                + ", userName=" + userName
                + ", brokerId=" + brokerId
                + ']';
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/resources/MqttPublishResource.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.resources;

import com.github.longkerdandy.mithqtt.api.auth.Authenticator;
import com.github.longkerdandy.mithqtt.api.auth.AuthorizeResult;
import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttAdditionalHeader;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.http.entity.ErrorCode;
import com.github.longkerdandy.mithqtt.http.entity.ErrorEntity;
import com.github.longkerdandy.mithqtt.http.entity.ResultEntity;
import com.github.longkerdandy.mithqtt.http.exception.AuthorizeException;
import com.github.longkerdandy.mithqtt.http.exception.ValidateException;
import com.github.longkerdandy.mithqtt.http.util.Validator;
import com.github.longkerdandy.mithqtt.util.Topics;
import com.sun.security.auth.UserPrincipal;
import io.dropwizard.auth.Auth;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.security.PermitAll;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * MQTT Publish related resource
 */
@Path("/clients/{clientId}/publish")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.TEXT_PLAIN)
public class MqttPublishResource extends AbstractResource {

    private static final Logger logger = LoggerFactory.getLogger(MqttPublishResource.class);

    public MqttPublishResource(String serverId, Validator validator, SyncStorage storage, Cluster cluster, Authenticator authenticator) {
        super(serverId, validator, storage, cluster, authenticator);
    }

    @PermitAll
    @POST
    public ResultEntity<Boolean> publish(@PathParam("clientId") String clientId, @Auth UserPrincipal user, @QueryParam("protocol") @DefaultValue("4") byte protocol,
                                         @QueryParam("dup") @DefaultValue("false") boolean dup, @QueryParam("qos") @DefaultValue("0") int qos,
                                         @QueryParam("topicName") String topicName, @QueryParam("packetId") @DefaultValue("0") int packetId,
                                         String body) throws UnsupportedEncodingException {
        String userName = user.getName();
        MqttVersion version = MqttVersion.fromProtocolLevel(protocol);
        byte[] bytes = body == null ? null : body.getBytes("ISO-8859-1");

        // HTTP interface require valid Client Id
        if (!this.validator.isClientIdValid(clientId)) {
            logger.debug("Protocol violation: Client id {} not valid based on configuration", clientId);
            throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
        }

        // The Topic Name in the PUBLISH Packet MUST NOT contain wildcard characters
        // Validate Topic Name based on configuration
        if (!this.validator.isTopicNameValid(topicName)) {
            logger.debug("Protocol violation: Client {} sent PUBLISH message contains invalid topic name {}", clientId, topicName);
            throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
        }

        // The Packet Identifier field is only present in PUBLISH Packets where the QoS level is 1 or 2.
        if (packetId <= 0 && (qos == MqttQoS.AT_LEAST_ONCE.value() || qos == MqttQoS.EXACTLY_ONCE.value())) {
            logger.debug("Protocol violation: Client {} sent PUBLISH message does not contain packet id, disconnect the client", clientId);
            throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
        }

        List<String> topicLevels = Topics.sanitizeTopicName(topicName);

        logger.debug("Message received: Received PUBLISH message from client {} user {} topic {}", clientId, user.getName(), topicName);

        AuthorizeResult result = this.authenticator.authPublish(clientId, userName, topicName, qos, false);
        // Authorize successful
        if (result == AuthorizeResult.OK) {
            logger.trace("Authorization PUBLISH succeeded on topic {} for client {}", topicName, clientId);

            // Prepare Message in advance, since the byte[] payload will be used in multiple location
            Message<MqttPublishVariableHeader, MqttPublishPayload> msg = new Message<>(
                    new MqttFixedHeader(MqttMessageType.PUBLISH, dup, MqttQoS.valueOf(qos), false, 0),
                    new MqttAdditionalHeader(version, clientId, userName, null),
                    packetId > 0 ? MqttPublishVariableHeader.from(topicName, packetId)
                            : MqttPublishVariableHeader.from(topicName),
                    new MqttPublishPayload(bytes));

            // When sending a PUBLISH Packet to a Client the Server MUST set the RETAIN flag to 1 if a message is
            // sent as a result of a new subscription being made by a Client. It MUST set the RETAIN
            // flag to 0 when a PUBLISH Packet is sent to a Client because it matches an established subscription
            // regardless of how the flag was set in the message it received.

            // The Server uses a PUBLISH Packet to send an Application Message to each Client which has a
            // matching subscription.
            // When Clients make subscriptions with Topic Filters that include wildcards, it is possible for a Client’s
            // subscriptions to overlap so that a published message might match multiple filters. In this case the Server
            // MUST deliver the message to the Client respecting the maximum QoS of all the matching subscriptions.
            // In addition, the Server MAY deliver further copies of the message, one for each
            // additional matching subscription and respecting the subscription’s QoS in each case.
            Map<String, MqttQoS> subscriptions = new HashMap<>();
            this.storage.getMatchSubscriptions(topicLevels, subscriptions);
            subscriptions.forEach((cid, q) -> {

                // Compare publish QoS and subscription QoS
                MqttQoS fQos = qos > q.value() ? q : MqttQoS.valueOf(qos);

                // Each time a Client sends a new packet of one of these
                // types it MUST assign it a currently unused Packet Identifier. If a Client re-sends a
                // particular Control Packet, then it MUST use the same Packet Identifier in subsequent re-sends of that
                // packet. The Packet Identifier becomes available for reuse after the Client has processed the
                // corresponding acknowledgement packet. In the case of a QoS 1 PUBLISH this is the corresponding
                // PUBACK; in the case of QoS 2 it is PUBCOMP. For SUBSCRIBE or UNSUBSCRIBE it is the
                // corresponding SUBACK or UNSUBACK. The same conditions apply to a Server when it
                // sends a PUBLISH with QoS > 0
                // A PUBLISH Packet MUST NOT contain a Packet Identifier if its QoS value is set to
                int pid = 0;
                if (fQos == MqttQoS.AT_LEAST_ONCE || fQos == MqttQoS.EXACTLY_ONCE) {
                    pid = this.storage.getNextPacketId(cid);
                }

                Message<MqttPublishVariableHeader, MqttPublishPayload> m = new Message<>(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, fQos, false, 0),
                        new MqttAdditionalHeader(MqttVersion.MQTT_3_1_1, cid, null, null),
                        pid > 0 ? MqttPublishVariableHeader.from(topicName, pid)
                                : MqttPublishVariableHeader.from(topicName),
                        msg.payload()
                );

                // Forward to recipient
                boolean d = false;
                String bid = this.storage.getConnectedNode(cid);
                if (StringUtils.isNotBlank(bid)) {
                    logger.trace("Send PUBLISH message to broker {} for client {} subscription", bid, cid);
                    d = true;
                    this.cluster.sendToBroker(bid, m);
                }

                // In the QoS 1 delivery protocol, the Sender
                // MUST treat the PUBLISH Packet as “unacknowledged” until it has received the corresponding
                // PUBACK packet from the receiver.
                // In the QoS 2 delivery protocol, the Sender
                // MUST treat the PUBLISH packet as “unacknowledged” until it has received the corresponding
                // PUBREC packet from the receiver.
                if (fQos == MqttQoS.AT_LEAST_ONCE || fQos == MqttQoS.EXACTLY_ONCE) {
                    logger.trace("Add in-flight PUBLISH message {} with QoS {} for client {}", pid, fQos, cid);
                    this.storage.addInFlightMessage(cid, pid, m, d);
                }
            });

            // Pass message to 3rd party application
            this.cluster.sendToApplication(msg);

            return new ResultEntity<>(true);
        } else {
            logger.trace("Authorization PUBLISH failed on topic {} for client {}", topicName, clientId);

            throw new AuthorizeException(new ErrorEntity(ErrorCode.UNAUTHORIZED));
        }
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/storage/sync/SyncStorage.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.storage.sync;

import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.commons.configuration.AbstractConfiguration;

import java.util.List;
import java.util.Map;

/**
 * Synchronized Storage
 */
@SuppressWarnings("unused")
public interface SyncStorage {

    /**
     * Init the storage
     * Should be invoked before using redis storage
     *
     * @param config Redis Configuration
     */
    void init(AbstractConfiguration config);

    /**
     * Destroy the storage
     * Should be invoked when gracefully shutdown
     */
    void destroy();

    /**
     * Try to lock specific client to specific state
     * State can only be CONNECTING & DISCONNECTING
     *
     * @param clientId Client Id
     * @param state    Client State
     * @return True if current state is CONNECTED or DISCONNECTED
     */
    boolean lock(String clientId, ConnectionState state);

    /**
     * Try to release specific client to specific state
     * State can only be CONNECTED & DISCONNECTED
     *
     * @param clientId Client Id
     * @param state    Client State
     * @return True if current state is CONNECTING or DISCONNECTING
     */
    boolean release(String clientId, ConnectionState state);

    /**
     * Get connected mqtt broker node (id) for the client
     *
     * @param clientId Client Id
     * @return MQTT Broker Node (Id)
     */
    String getConnectedNode(String clientId);

    /**
     * Update connected mqtt broker node (id) for the client
     *
     * @param clientId Client Id
     * @param node     MQTT Broker Node (Id)
     * @param seconds  TTL
     * @return Previous connected MQTT Broker Node (Id), Null if not exist
     */
    String updateConnectedNode(String clientId, String node, int seconds);

    /**
     * Refresh TTL connected mqtt broker node (id) for the client
     *
     * @param clientId Client Id
     * @param node     MQTT Broker Node (Id)
     * @param seconds  TTL
     * @return Refreshed? (Exist)
     */
    boolean refreshConnectedNode(String clientId, String node, int seconds);

    /**
     * Remove connected mqtt broker node (id) for the client
     *
     * @param clientId Client Id
     * @param node     MQTT Broker Node (Id)
     * @return Connected node removed? (Exist)
     */
    boolean removeConnectedNode(String clientId, String node);

    /**
     * Get session existence for the client
     *
     * @param clientId Client Id
     * @return Session Existence (1 clean session, 0 normal session, < 0 not exist)
     */
    int getSessionExist(String clientId);

    /**
     * Update session existence for the client
     *
     * @param clientId     Client Id
     * @param cleanSession Clean Session
     */
    void updateSessionExist(String clientId, boolean cleanSession);

    /**
     * Remove session existence for the client
     *
     * @param clientId Client Id
     * @return Session removed? (Exist)
     */
    boolean removeSessionExist(String clientId);

    /**
     * Remove all session state
     *
     * @param clientId Client Id
     */
    void removeAllSessionState(String clientId);

    /**
     * Get next packet id for the client
     *
     * @param clientId Client Id
     * @return Next Packet Id
     */
    int getNextPacketId(String clientId);

    /**
     * Get specific in-flight message for the client
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     * @return In-Flight Message
     */
    Message getInFlightMessage(String clientId, int packetId);

    /**
     * Add in-flight message for the client
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     * @param msg      In-Flight Message
     * @param dup      Duplicated
     */
    void addInFlightMessage(String clientId, int packetId, Message msg, boolean dup);

    /**
     * Add in-flight message for the client but expires in certain duration
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     * @param msg      In-Flight Message
     * @param dup      Duplicated
     * @param ttl      Time To Live in seconds
     */
    void addInFlightMessage(String clientId, int packetId, Message msg, boolean dup, long ttl);

    /**
     * Remove specific in-flight message for the client
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     */
    void removeInFlightMessage(String clientId, int packetId);

    /**
     * Get all in-flight message's for the client
     * Including:
     * QoS 1 and QoS 2 PUBLISH messages which have been sent to the Client, but have not been acknowledged.
     * QoS 0, QoS 1 and QoS 2 PUBLISH messages pending transmission to the Client.
     * QoS 2 PUBREL messages which have been sent from the Client, but have not been acknowledged.
     *
     * @param clientId Client Id
     * @return List of In-Flight Message
     */
    List<Message> getAllInFlightMessages(String clientId);

    /**
     * Remove all in-flight message for the client
     *
     * @param clientId Client Id
     */
    void removeAllInFlightMessage(String clientId);

    /**
     * Add unacknowledged qos 2 PUBLISH message's packet id from the client
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     * @return Packet Id added? (Not exist)
     */
    boolean addQoS2MessageId(String clientId, int packetId);

    /**
     * Remove unacknowledged qos 2 PUBLISH message's packet id from the client
     *
     * @param clientId Client Id
     * @param packetId Packet Id
     * @return Packet Id removed? (Exist)
     */
    boolean removeQoS2MessageId(String clientId, int packetId);

    /**
     * Remove all unacknowledged qos 2 PUBLISH message's packet id from the client
     *
     * @param clientId Client Id
     */
    void removeAllQoS2MessageId(String clientId);

    /**
     * Get the topic's subscriptions
     * Topic Levels must been sanitized
     *
     * @param topicLevels List of topic levels
     * @return Subscriptions: Key - Client Id, Value - QoS
     */
    Map<String, MqttQoS> getTopicSubscriptions(List<String> topicLevels);

    /**
     * Get the client's subscriptions
     *
     * @param clientId Client Id
     * @return Subscriptions: Key - Topic, Value - QoS
     */
    Map<String, MqttQoS> getClientSubscriptions(String clientId);

    /**
     * Update topic subscription for the client
     * Topic Levels must been sanitized
     *
     * @param clientId    Client Id
     * @param topicLevels List of topic levels
     * @param qos         Subscription QoS
     */
    void updateSubscription(String clientId, List<String> topicLevels, MqttQoS qos);

    /***
     * Remove topic name subscription for the client
     * Topic Levels must been sanitized
     *
     * @param clientId    Client Id
     * @param topicLevels List of topic levels
     */
    void removeSubscription(String clientId, List<String> topicLevels);

    /**
     * Remove all subscriptions for the client
     *
     * @param clientId Client Id
     */
    void removeAllSubscriptions(String clientId);

    /**
     * Get all subscriptions matching the topic
     * This is a recursion method
     *
     * @param topicLevels List of topic levels
     * @param map         RETURN VALUE! Subscriptions: Key - Client Id, Value - QoS
     */
    void getMatchSubscriptions(List<String> topicLevels, Map<String, MqttQoS> map);

    /**
     * Add retain message for the topic name
     * Retain id will be generated
     *
     * @param topicLevels Topic Levels
     * @param msg         Retain Message
     * @return Retain Id
     */
    int addRetainMessage(List<String> topicLevels, Message<MqttPublishVariableHeader, MqttPublishPayload> msg);

    /**
     * Remove all retain messages for the topic name
     *
     * @param topicLevels Topic Levels
     */
    void removeAllRetainMessage(List<String> topicLevels);

    /**
     * Get all retain messages the topic name
     *
     * @param topicLevels Topic Levels
     * @return List of Retain Message
     */
    List<Message<MqttPublishVariableHeader, MqttPublishPayload>> getMatchRetainMessages(List<String> topicLevels);
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/MqttPacketIdVariableHeader.java<|end_filename|>
/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.handler.codec.mqtt;

import io.netty.util.internal.StringUtil;

/**
 * Variable Header containing only Message Id
 * See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#msg-id">MQTTV3.1/msg-id</a>
 */
public final class MqttPacketIdVariableHeader {

    protected int packetId;

    private MqttPacketIdVariableHeader() {
    }

    private MqttPacketIdVariableHeader(int packetId) {
        this.packetId = packetId;
    }

    public static MqttPacketIdVariableHeader from(int packetId) {
        if (packetId < 1 || packetId > 0xffff) {
            throw new IllegalArgumentException("packetId: " + packetId + " (expected: 1 ~ 65535)");
        }
        return new MqttPacketIdVariableHeader(packetId);
    }

    public int packetId() {
        return packetId;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "packetId=" + packetId
                + ']';
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/cluster/Cluster.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.cluster;

import com.github.longkerdandy.mithqtt.api.message.Message;
import org.apache.commons.configuration.AbstractConfiguration;

/**
 * Cluster
 */
public interface Cluster {

    /**
     * Init the cluster
     *
     * @param config  Configuration
     * @param factory Cluster Listener Factory
     * @throws ClusterException if there is a exception when trying to init cluster.
     */
    void init(AbstractConfiguration config, ClusterListenerFactory factory) throws ClusterException;

    /**
     * Destroy the cluster
     */
    void destroy();

    /**
     * Send message to broker
     *
     * @param brokerId Broker Id
     * @param message  Message
     */
    void sendToBroker(String brokerId, Message message);

    /**
     * Send message to outside
     * Other application can pick up the message from there
     *
     * @param message Message
     */
    void sendToApplication(Message message);
}


<|start_filename|>mithqtt-broker/src/main/java/com/github/longkerdandy/mithqtt/broker/util/Validator.java<|end_filename|>
package com.github.longkerdandy.mithqtt.broker.util;

import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.lang3.StringUtils;

import java.util.regex.Pattern;

/**
 * Validator
 */
@SuppressWarnings("unused")
public class Validator {

    // MQTT client id validate regex pattern
    protected Pattern clientIdPattern;
    // MQTT user name validate regex pattern
    protected Pattern userNamePattern;
    // MQTT password validate regex pattern
    protected Pattern passwordPattern;
    // MQTT topic name validate regex pattern
    protected Pattern topicNamePattern;
    // MQTT topic filter validate regex pattern
    protected Pattern topicFilterPattern;

    public Validator(AbstractConfiguration config) {
        if (StringUtils.isNotBlank(config.getString("mqtt.clientId.validator")))
            this.clientIdPattern = Pattern.compile(config.getString("mqtt.clientId.validator"));
        if (StringUtils.isNotBlank(config.getString("mqtt.userName.validator")))
            this.userNamePattern = Pattern.compile(config.getString("mqtt.userName.validator"));
        if (StringUtils.isNotBlank(config.getString("mqtt.password.validator")))
            this.passwordPattern = Pattern.compile(config.getString("mqtt.password.validator"));
        if (StringUtils.isNotBlank(config.getString("mqtt.topicName.validator")))
            this.topicNamePattern = Pattern.compile(config.getString("mqtt.topicName.validator"));
        if (StringUtils.isNotBlank(config.getString("mqtt.topicFilter.validator")))
            this.topicFilterPattern = Pattern.compile(config.getString("mqtt.topicFilter.validator"));
    }

    /**
     * Is MQTT topic name (no wildcards) valid
     *
     * @param topicName Topic Name
     * @return True if valid
     */
    public boolean isTopicNameValid(String topicName) {
        return !StringUtils.isEmpty(topicName) &&
                !topicName.contains("+") &&
                !topicName.contains("#") &&
                (this.topicNamePattern == null || this.topicNamePattern.matcher(topicName).matches());
    }

    /**
     * Is MQTT topic filter (may contain wildcards) valid
     *
     * @param topicFilter Topic Filter
     * @return True if valid
     */
    public boolean isTopicFilterValid(String topicFilter) {
        return !StringUtils.isEmpty(topicFilter) &&
                (this.topicFilterPattern == null || this.topicFilterPattern.matcher(topicFilter).matches());
    }

    /**
     * Is MQTT client id valid
     *
     * @param clientId Client Id
     * @return True if valid
     */
    public boolean isClientIdValid(String clientId) {
        return this.clientIdPattern == null || this.clientIdPattern.matcher(clientId).matches();
    }

    /**
     * Is MQTT user name valid
     *
     * @param userName User Name
     * @return True if valid
     */
    public boolean isUserNameValid(String userName) {
        return this.userNamePattern == null || this.userNamePattern.matcher(userName).matches();
    }

    /**
     * Is MQTT password valid
     *
     * @param password Password
     * @return True if valid
     */
    public boolean isPasswordValid(String password) {
        return this.passwordPattern == null || this.passwordPattern.matcher(password).matches();
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/cluster/ClusterException.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.cluster;

/**
 * Cluster Exception
 */
public class ClusterException extends Exception {

    public ClusterException() {
    }

    public ClusterException(String message) {
        super(message);
    }

    public ClusterException(String message, Throwable cause) {
        super(message, cause);
    }

    public ClusterException(Throwable cause) {
        super(cause);
    }

    public ClusterException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
        super(message, cause, enableSuppression, writableStackTrace);
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/message/MqttTopicSubscriptionGranted.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import io.netty.handler.codec.mqtt.MqttGrantedQoS;
import io.netty.util.internal.StringUtil;

/**
 * Contains a topic name and granted Qos Level.
 * This is part of the {@link MqttSubscribePayloadGranted}
 */
public class MqttTopicSubscriptionGranted {

    protected String topic;
    protected MqttGrantedQoS grantedQos;

    private MqttTopicSubscriptionGranted() {
    }

    public MqttTopicSubscriptionGranted(String topic, MqttGrantedQoS grantedQos) {
        this.topic = topic;
        this.grantedQos = grantedQos;
    }

    public String topic() {
        return topic;
    }

    public MqttGrantedQoS grantedQos() {
        return grantedQos;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "topic=" + topic
                + ", grantedQos=" + grantedQos
                + ']';
    }
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/Mqtts.java<|end_filename|>
package io.netty.handler.codec.mqtt;

import org.apache.commons.lang3.StringUtils;

import static com.github.longkerdandy.mithqtt.util.UUIDs.shortUuid;

/**
 * MQTT Utils
 */
public class Mqtts {

    /**
     * Sanitize MQTT message
     *
     * @param msg MQTT message
     */
    public static void sanitize(MqttMessage msg) {
        switch (msg.fixedHeader().messageType()) {
            case CONNECT:
                if (StringUtils.isBlank(((MqttConnectMessage) msg).payload().clientId())
                        && ((MqttConnectMessage) msg).variableHeader().cleanSession())
                    ((MqttConnectMessage) msg).payload().clientId = shortUuid();
            case CONNACK:
            case PUBACK:
            case PUBREC:
            case PUBCOMP:
            case SUBACK:
            case UNSUBACK:
            case PINGREQ:
            case PINGRESP:
            case DISCONNECT:
                msg.fixedHeader().dup = false;
                msg.fixedHeader().qos = MqttQoS.AT_MOST_ONCE;
                msg.fixedHeader().retain = false;
                break;
            case PUBREL:
            case SUBSCRIBE:
            case UNSUBSCRIBE:
                msg.fixedHeader().dup = false;
                msg.fixedHeader().qos = MqttQoS.AT_LEAST_ONCE;
                msg.fixedHeader().retain = false;
                break;
            case PUBLISH:
                if (msg.fixedHeader().qos == MqttQoS.AT_MOST_ONCE)
                    ((MqttPublishVariableHeader) msg.variableHeader()).packetId = 0;
                break;
        }
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/message/MqttSubscribePayloadGranted.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.util.internal.StringUtil;
import org.apache.commons.lang3.ArrayUtils;

import java.util.List;

/**
 * Payload of the {@link MqttSubscribeMessage}
 * After subscription was granted
 */
public class MqttSubscribePayloadGranted {

    protected List<MqttTopicSubscriptionGranted> subscriptions;

    private MqttSubscribePayloadGranted() {
    }

    public MqttSubscribePayloadGranted(List<MqttTopicSubscriptionGranted> subscriptions) {
        this.subscriptions = subscriptions;
    }

    public List<MqttTopicSubscriptionGranted> subscriptions() {
        return subscriptions;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "subscriptions=" + ArrayUtils.toString(subscriptions)
                + ']';
    }
}


<|start_filename|>mithqtt-application-sample/src/main/java/com/github/longkerdandy/mithqtt/application/sample/cluster/ApplicationClusterListenerImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.application.sample.cluster;

import com.github.longkerdandy.mithqtt.api.cluster.ClusterListener;
import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.message.MqttSubscribePayloadGranted;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Broker Cluster Listener Implementation
 */
public class ApplicationClusterListenerImpl implements ClusterListener {

    private static final Logger logger = LoggerFactory.getLogger(ApplicationClusterListenerImpl.class);

    @Override
    public void onConnect(Message<MqttConnectVariableHeader, MqttConnectPayload> msg) {
        logger.debug("Received CONNECT message {}", msg);
    }

    @Override
    public void onSubscribe(Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> msg) {
        logger.debug("Received SUBSCRIBE message {}", msg);
    }

    @Override
    public void onUnsubscribe(Message<MqttPacketIdVariableHeader, MqttUnsubscribePayload> msg) {
        logger.debug("Received UNSUBSCRIBE message {}", msg);
    }

    @Override
    public void onPublish(Message<MqttPublishVariableHeader, MqttPublishPayload> msg) {
        logger.debug("Received PUBLISH message {}", msg);
    }

    @Override
    public void onDisconnect(Message<Void, Void> msg) {
        logger.debug("Received DISCONNECT message {}", msg);
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/util/UUIDs.java<|end_filename|>
package com.github.longkerdandy.mithqtt.util;

import com.fasterxml.uuid.Generators;
import com.fasterxml.uuid.impl.NameBasedGenerator;
import com.fasterxml.uuid.impl.RandomBasedGenerator;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.UUID;

/**
 * UUID Utils
 */
@SuppressWarnings("unused")
public class UUIDs {

    private UUIDs() {
    }

    /**
     * Generate short version UUID from given uri
     *
     * @param uri Uri/Url
     * @return UUID
     */
    public static String shortUuid(URI uri) {
        NameBasedGenerator generator = Generators.nameBasedGenerator(NameBasedGenerator.NAMESPACE_URL);
        UUID uuid = generator.generate(uri.toString());
        // https://gist.github.com/LeeSanghoon/5811136
        long l = ByteBuffer.wrap(uuid.toString().getBytes()).getLong();
        return Long.toString(l, Character.MAX_RADIX);
    }

    /**
     * Generate short version random UUID
     *
     * @return UUID
     */
    public static String shortUuid() {
        RandomBasedGenerator generator = Generators.randomBasedGenerator();
        UUID uuid = generator.generate();
        // https://gist.github.com/LeeSanghoon/5811136
        long l = ByteBuffer.wrap(uuid.toString().getBytes()).getLong();
        return Long.toString(l, Character.MAX_RADIX);
    }
}

<|start_filename|>mithqtt-storage-redis/src/main/java/com/github/longkerdandy/mithqtt/storage/redis/sync/RedisSyncMasterSlaveStorageImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.storage.redis.sync;

import com.lambdaworks.redis.ReadFrom;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisURI;
import com.lambdaworks.redis.api.sync.*;
import com.lambdaworks.redis.codec.Utf8StringCodec;
import com.lambdaworks.redis.masterslave.MasterSlave;
import com.lambdaworks.redis.masterslave.StatefulRedisMasterSlaveConnection;
import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 * Synchronized Storage for Master Slave Redis setup
 */
@SuppressWarnings("unused")
public class RedisSyncMasterSlaveStorageImpl extends RedisSyncSingleStorageImpl {

    // A scalable thread-safe Redis client. Multiple threads may share one connection if they avoid
    // blocking and transactional operations such as BLPOP and MULTI/EXEC.
    private RedisClient lettuceMasterSlave;
    // A thread-safe connection to a redis server. Multiple threads may share one StatefulRedisConnection
    private StatefulRedisMasterSlaveConnection<String, String> lettuceMasterSlaveConn;

    protected RedisHashCommands<String, String> hash() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisKeyCommands<String, String> key() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisStringCommands<String, String> string() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisListCommands<String, String> list() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisSetCommands<String, String> set() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisSortedSetCommands<String, String> sortedSet() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisScriptingCommands<String, String> script() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisServerCommands<String, String> server() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisHLLCommands<String, String> hll() {
        return this.lettuceMasterSlaveConn.sync();
    }

    protected RedisGeoCommands<String, String> geo() {
        return this.lettuceMasterSlaveConn.sync();
    }

    @Override
    public void init(AbstractConfiguration config) {
        if (!config.getString("redis.type").equals("master_slave")) {
            throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with master slave redis setup, but redis.type value is " + config.getString("redis.type"));
        }

        List<String> address = parseRedisAddress(config.getString("redis.address"), 6379);
        int databaseNumber = config.getInt("redis.database", 0);
        String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

        // lettuce
        RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
        this.lettuceMasterSlave = RedisClient.create(lettuceURI);
        this.lettuceMasterSlaveConn = MasterSlave.connect(this.lettuceMasterSlave, new Utf8StringCodec(), lettuceURI);
        this.lettuceMasterSlaveConn.setReadFrom(ReadFrom.valueOf(config.getString("redis.read")));

        // params
        initParams(config);
    }

    @Override
    public void destroy() {
        // shutdown this client and close all open connections
        if (this.lettuceMasterSlaveConn != null) this.lettuceMasterSlaveConn.close();
        if (this.lettuceMasterSlave != null) this.lettuceMasterSlave.shutdown();
    }
}


<|start_filename|>mithqtt-application-sample/src/main/java/com/github/longkerdandy/mithqtt/application/sample/cluster/ApplicationClusterListenerFactoryImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.application.sample.cluster;

import com.github.longkerdandy.mithqtt.api.cluster.ClusterListener;
import com.github.longkerdandy.mithqtt.api.cluster.ClusterListenerFactory;

/**
 * Broker Cluster Listener Factory Implementation
 */
public class ApplicationClusterListenerFactoryImpl implements ClusterListenerFactory {

    @Override
    public ClusterListener newListener() {
        return new ApplicationClusterListenerImpl();
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/storage/sync/ConnectionState.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.storage.sync;

/**
 * MQTT Client Connection State
 */
public enum ConnectionState {
    DISCONNECTED(-1),
    DISCONNECTING(0),
    CONNECTING(1),
    CONNECTED(2);

    private final int value;

    ConnectionState(int value) {
        this.value = value;
    }

    public static ConnectionState valueOf(int value) {
        for (ConnectionState s : values()) {
            if (s.value == value) {
                return s;
            }
        }
        throw new IllegalArgumentException("invalid connection state: " + value);
    }

    public int value() {
        return value;
    }
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/MqttPublishVariableHeader.java<|end_filename|>
/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.handler.codec.mqtt;

import io.netty.util.internal.StringUtil;

/**
 * Variable Header of the {@link MqttPublishMessage}
 */
public class MqttPublishVariableHeader {

    protected String topicName;
    protected int packetId;

    private MqttPublishVariableHeader() {
    }

    private MqttPublishVariableHeader(String topicName, int packetId) {
        this.topicName = topicName;
        this.packetId = packetId;
    }

    public static MqttPublishVariableHeader from(String topicName) {
        return new MqttPublishVariableHeader(topicName, 0);
    }

    public static MqttPublishVariableHeader from(String topicName, int packetId) {
        if (packetId < 1 || packetId > 0xffff) {
            throw new IllegalArgumentException("packetId: " + packetId + " (expected: 1 ~ 65535)");
        }
        return new MqttPublishVariableHeader(topicName, packetId);
    }

    public String topicName() {
        return topicName;
    }

    public int packetId() {
        return packetId;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "topic=" + topicName
                + ", packetId=" + packetId
                + ']';
    }
}


<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/api/message/MqttPublishPayload.java<|end_filename|>
package com.github.longkerdandy.mithqtt.api.message;

import io.netty.buffer.ByteBuf;
import io.netty.util.internal.StringUtil;

/**
 * Mqtt Publish Payload with byte[]
 */
public class MqttPublishPayload {

    protected byte[] bytes;

    private MqttPublishPayload() {
    }

    public MqttPublishPayload(byte[] bytes) {
        this.bytes = bytes;
    }

    public MqttPublishPayload(ByteBuf buf) {
        ByteBuf b = buf.duplicate();
        this.bytes = new byte[b.readableBytes()];
        b.readBytes(this.bytes);
    }

    public byte[] bytes() {
        return bytes;
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(this)
                + '['
                + "payload " + bytes.length + " bytes"
                + ']';
    }
}


<|start_filename|>mithqtt-storage-redis/src/main/java/com/github/longkerdandy/mithqtt/storage/redis/sync/RedisSyncSingleStorageImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.storage.redis.sync;

import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import com.github.longkerdandy.mithqtt.api.storage.sync.ConnectionState;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.storage.redis.RedisKey;
import com.github.longkerdandy.mithqtt.storage.redis.RedisLua;
import com.github.longkerdandy.mithqtt.util.Topics;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisURI;
import com.lambdaworks.redis.ScriptOutputType;
import com.lambdaworks.redis.api.StatefulRedisConnection;
import com.lambdaworks.redis.api.sync.*;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.github.longkerdandy.mithqtt.storage.redis.util.Converter.mapToMessage;
import static com.github.longkerdandy.mithqtt.storage.redis.util.Converter.messageToMap;
import static com.github.longkerdandy.mithqtt.util.Topics.END;

/**
 * Synchronized Storage for Single Redis setup
 */
public class RedisSyncSingleStorageImpl implements SyncStorage {

    // Max in-flight queue size per client
    private int inFlightQueueSize;
    // Max QoS 2 ids queue size per client
    private int qos2QueueSize;
    // Max retain queue size per topic
    private int retainQueueSize;

    // A scalable thread-safe Redis client. Multiple threads may share one connection if they avoid
    // blocking and transactional operations such as BLPOP and MULTI/EXEC.
    private RedisClient lettuce;
    // A thread-safe connection to a redis server. Multiple threads may share one StatefulRedisConnection
    private StatefulRedisConnection<String, String> lettuceConn;

    protected RedisHashCommands<String, String> hash() {
        return this.lettuceConn.sync();
    }

    protected RedisKeyCommands<String, String> key() {
        return this.lettuceConn.sync();
    }

    protected RedisStringCommands<String, String> string() {
        return this.lettuceConn.sync();
    }

    protected RedisListCommands<String, String> list() {
        return this.lettuceConn.sync();
    }

    protected RedisSetCommands<String, String> set() {
        return this.lettuceConn.sync();
    }

    protected RedisSortedSetCommands<String, String> sortedSet() {
        return this.lettuceConn.sync();
    }

    protected RedisScriptingCommands<String, String> script() {
        return this.lettuceConn.sync();
    }

    protected RedisServerCommands<String, String> server() {
        return this.lettuceConn.sync();
    }

    protected RedisHLLCommands<String, String> hll() {
        return this.lettuceConn.sync();
    }

    protected RedisGeoCommands<String, String> geo() {
        return this.lettuceConn.sync();
    }

    @Override
    public void init(AbstractConfiguration config) {
        if (!config.getString("redis.type").equals("single")) {
            throw new IllegalStateException("RedisSyncSingleStorageImpl class can only be used with single redis setup, but redis.type value is " + config.getString("redis.type"));
        }

        List<String> address = parseRedisAddress(config.getString("redis.address"), 6379);
        int databaseNumber = config.getInt("redis.database", 0);
        String password = StringUtils.isNotEmpty(config.getString("redis.password")) ? config.getString("redis.password") + "@" : "";

        // lettuce
        RedisURI lettuceURI = RedisURI.create("redis://" + password + address.get(0) + "/" + databaseNumber);
        this.lettuce = RedisClient.create(lettuceURI);
        this.lettuceConn = this.lettuce.connect();

        // params
        initParams(config);
    }

    @Override
    public void destroy() {
        // shutdown this client and close all open connections
        if (this.lettuceConn != null) this.lettuceConn.close();
        if (this.lettuce != null) this.lettuce.shutdown();
    }

    /**
     * Parse address string to a List of host:port String
     *
     * @param address Address String
     * @return List of host:port String
     */
    List<String> parseRedisAddress(String address, int defaultPort) {
        List<String> list = new ArrayList<>();
        String[] array = address.split(",");
        for (String s : array) {
            if (!s.contains(":"))
                s = s + ":" + defaultPort;
            list.add(s);
        }
        return list;
    }

    /**
     * Initialize MQTT parameters
     *
     * @param config Redis Configuration
     */
    void initParams(AbstractConfiguration config) {
        this.inFlightQueueSize = config.getInt("mqtt.inflight.queue.size", 0);
        this.qos2QueueSize = config.getInt("mqtt.qos2.queue.size", 0);
        this.retainQueueSize = config.getInt("mqtt.retain.queue.size", 0);
    }

    @Override
    public boolean lock(String clientId, ConnectionState state) {
        // nil(-1):DISCONNECTED 0:DISCONNECTING 1:CONNECTING 2:CONNECTED
        long r = this.script().eval("local current = redis.call('HGET', KEYS[1], 'state')\n" +
                "if (not current or '2' == current) and '1' == ARGV[1]\n" +
                "then\n" +
                "   redis.call('HSET', KEYS[1], 'state', ARGV[1])\n" +
                "   return 1\n" +
                "end\n" +
                "if '2' == current and '0' == ARGV[1]\n" +
                "then\n" +
                "   redis.call('HSET', KEYS[1], 'state', ARGV[1])\n" +
                "   return 1\n" +
                "end\n" +
                "return 0", ScriptOutputType.INTEGER, new String[]{RedisKey.connection(clientId)}, String.valueOf(state.value()));
        return r == 1;
    }

    @Override
    public boolean release(String clientId, ConnectionState state) {
        // nil(-1):DISCONNECTED 0:DISCONNECTING 1:CONNECTING 2:CONNECTED
        long r = this.script().eval("local current = redis.call('HGET', KEYS[1], 'state')\n" +
                "if '1' == current and '2' == ARGV[1]\n" +
                "then\n" +
                "   redis.call('HSET', KEYS[1], 'state', ARGV[1])\n" +
                "   return 1\n" +
                "end\n" +
                "if '0' == current and '-1' == ARGV[1]\n" +
                "then\n" +
                "   redis.call('HDEL', KEYS[1], 'state')\n" +
                "   return 1\n" +
                "end\n" +
                "return 0", ScriptOutputType.INTEGER, new String[]{RedisKey.connection(clientId)}, String.valueOf(state.value()));
        return r == 1;
    }

    @Override
    public String getConnectedNode(String clientId) {
        return this.hash().hget(RedisKey.connection(clientId), "node");
    }

    @Override
    public String updateConnectedNode(String clientId, String node, int seconds) {
        return this.script().eval("local old = redis.call('HGET', KEYS[1], 'node')\n" +
                "redis.call('HSET', KEYS[1], 'node', ARGV[1])\n" +
                "redis.call('EXPIRE', KEYS[1], ARGV[2])\n" +
                "return old", ScriptOutputType.VALUE, new String[]{RedisKey.connection(clientId)}, node, String.valueOf(seconds));
    }

    @Override
    public boolean refreshConnectedNode(String clientId, String node, int seconds) {
        long r = this.script().eval("if ARGV[1] == redis.call('HGET', KEYS[1], 'node')\n" +
                "then\n" +
                "   redis.call('EXPIRE', KEYS[1], ARGV[2])\n" +
                "   return 1\n" +
                "end\n" +
                "return 0", ScriptOutputType.INTEGER, new String[]{RedisKey.connection(clientId)}, node, String.valueOf(seconds));
        return r == 1;
    }

    @Override
    public boolean removeConnectedNode(String clientId, String node) {
        long r = this.script().eval("if ARGV[1] == redis.call('HGET', KEYS[1], 'node')\n" +
                "then\n" +
                "   redis.call('DEL', KEYS[1])\n" +
                "   return 1\n" +
                "end\n" +
                "return 0", ScriptOutputType.INTEGER, new String[]{RedisKey.connection(clientId)}, node);
        return r == 1;
    }


    @Override
    public int getSessionExist(String clientId) {
        String r = this.string().get(RedisKey.session(clientId));
        if (r != null) return Integer.parseInt(r);
        else return -1;
    }

    @Override
    public void updateSessionExist(String clientId, boolean cleanSession) {
        this.string().set(RedisKey.session(clientId), BooleanUtils.toString(cleanSession, "1", "0"));
    }

    @Override
    public boolean removeSessionExist(String clientId) {
        return this.key().del(RedisKey.session(clientId)) == 1;
    }

    @Override
    public void removeAllSessionState(String clientId) {
        removeSessionExist(clientId);
        removeAllSubscriptions(clientId);
        removeAllQoS2MessageId(clientId);
        removeAllInFlightMessage(clientId);
    }

    @Override
    public int getNextPacketId(String clientId) {
        return Math.toIntExact(this.script().eval(RedisLua.INCRLIMIT, ScriptOutputType.INTEGER, new String[]{RedisKey.nextPacketId(clientId)}, "65535"));
    }

    @Override
    public Message getInFlightMessage(String clientId, int packetId) {
        Message m = mapToMessage(this.hash().hgetall(RedisKey.inFlightMessage(clientId, packetId)));
        if (m == null) removeInFlightMessage(clientId, packetId);
        return m;
    }

    @Override
    public void addInFlightMessage(String clientId, int packetId, Message msg, boolean dup) {
        Map<String, String> map = messageToMap(msg);
        map.put("dup", BooleanUtils.toString(dup, "1", "0"));
        String r = this.script().eval(RedisLua.RPUSHLIMIT, ScriptOutputType.VALUE, new String[]{RedisKey.inFlightList(clientId)}, String.valueOf(packetId), String.valueOf(this.inFlightQueueSize));
        if (r != null) this.key().del(RedisKey.inFlightMessage(clientId, Integer.parseInt(r)));
        this.hash().hmset(RedisKey.inFlightMessage(clientId, packetId), map);
    }

    @Override
    public void addInFlightMessage(String clientId, int packetId, Message msg, boolean dup, long ttl) {
        addInFlightMessage(clientId, packetId, msg, dup);
        this.key().expire(RedisKey.inFlightMessage(clientId, packetId), ttl);
    }

    @Override
    public void removeInFlightMessage(String clientId, int packetId) {
        this.list().lrem(RedisKey.inFlightList(clientId), 0, String.valueOf(packetId));
        this.key().del(RedisKey.inFlightMessage(clientId, packetId));
    }

    @Override
    public List<Message> getAllInFlightMessages(String clientId) {
        List<Message> r = new ArrayList<>();
        List<String> ids = this.list().lrange(RedisKey.inFlightList(clientId), 0, -1);
        if (ids != null) {
            ids.forEach(packetId -> {
                Message m = getInFlightMessage(clientId, Integer.parseInt(packetId));
                if (m != null) r.add(m);
                else removeInFlightMessage(clientId, Integer.parseInt(packetId));
            });
        }
        return r;
    }

    @Override
    public void removeAllInFlightMessage(String clientId) {
        List<String> ids = this.list().lrange(RedisKey.inFlightList(clientId), 0, -1);
        if (ids != null) {
            ids.forEach(packetId ->
                    removeInFlightMessage(clientId, Integer.parseInt(packetId)));
        }
    }

    @Override
    public boolean addQoS2MessageId(String clientId, int packetId) {
        long r = this.script().eval(RedisLua.ZADDLIMIT, ScriptOutputType.INTEGER,
                new String[]{RedisKey.qos2Set(clientId)},
                String.valueOf(System.currentTimeMillis()),
                String.valueOf(packetId),
                String.valueOf(this.qos2QueueSize));
        return r == 1;
    }

    @Override
    public boolean removeQoS2MessageId(String clientId, int packetId) {
        return this.sortedSet().zrem(RedisKey.qos2Set(clientId), String.valueOf(packetId)) == 1;
    }

    @Override
    public void removeAllQoS2MessageId(String clientId) {
        this.key().del(RedisKey.qos2Set(clientId));
    }

    @Override
    public Map<String, MqttQoS> getTopicSubscriptions(List<String> topicLevels) {
        Map<String, String> subscriptions;
        if (Topics.isTopicFilter(topicLevels)) {
            subscriptions = this.hash().hgetall(RedisKey.topicFilter(topicLevels));
        } else {
            subscriptions = this.hash().hgetall(RedisKey.topicName(topicLevels));
        }
        return subscriptions.entrySet()
                .stream()
                .collect(Collectors.toMap(Map.Entry::getKey, entry -> MqttQoS.valueOf(Integer.parseInt(entry.getValue()))));
    }

    @Override
    public Map<String, MqttQoS> getClientSubscriptions(String clientId) {
        return this.hash().hgetall(RedisKey.subscription(clientId))
                .entrySet()
                .stream()
                .collect(Collectors.toMap(Map.Entry::getKey, entry -> MqttQoS.valueOf(Integer.parseInt(entry.getValue()))));
    }

    @Override
    public void updateSubscription(String clientId, List<String> topicLevels, MqttQoS qos) {
        if (Topics.isTopicFilter(topicLevels)) {
            boolean b1 = this.hash().hset(RedisKey.subscription(clientId), String.join("/", topicLevels), String.valueOf(qos.value()));
            boolean b2 = this.hash().hset(RedisKey.topicFilter(topicLevels), clientId, String.valueOf(qos.value()));
            if (b1 && b2) {
                List<String> keys = new ArrayList<>();
                List<String> argv = new ArrayList<>();
                // topic filter tree
                for (int i = 0; i < topicLevels.size(); i++) {
                    keys.add(RedisKey.topicFilterChild(topicLevels.subList(0, i)));
                    argv.add(topicLevels.get(i));
                }
                this.script().eval("local length = table.getn(KEYS)\n" +
                                "for i = 1, length do\n" +
                                "   redis.call('HINCRBY', KEYS[i], ARGV[i], 1)\n" +
                                "end\n" +
                                "return redis.status_reply('OK')",
                        ScriptOutputType.STATUS, keys.toArray(new String[keys.size()]), argv.toArray(new String[argv.size()]));
            }
        } else {
            this.hash().hset(RedisKey.subscription(clientId), String.join("/", topicLevels), String.valueOf(qos.value()));
            this.hash().hset(RedisKey.topicName(topicLevels), clientId, String.valueOf(qos.value()));
        }
    }

    @Override
    public void removeSubscription(String clientId, List<String> topicLevels) {
        if (Topics.isTopicFilter(topicLevels)) {
            long b1 = this.hash().hdel(RedisKey.subscription(clientId), String.join("/", topicLevels));
            long b2 = this.hash().hdel(RedisKey.topicFilter(topicLevels), clientId);
            if (b1 == 1 && b2 == 1) {
                List<String> keys = new ArrayList<>();
                List<String> argv = new ArrayList<>();
                // topic filter tree
                for (int i = 0; i < topicLevels.size(); i++) {
                    keys.add(RedisKey.topicFilterChild(topicLevels.subList(0, i)));
                    argv.add(topicLevels.get(i));
                }
                this.script().eval("local length = table.getn(KEYS)\n" +
                                "for i = 1, length do\n" +
                                "   local count = redis.call('HINCRBY', KEYS[i], ARGV[i], -1)\n" +
                                "   if count == 0\n" +
                                "   then\n" +
                                "       redis.call('HDEL', KEYS[i], ARGV[i])\n" +
                                "   end\n" +
                                "end\n" +
                                "return redis.status_reply('OK')",
                        ScriptOutputType.STATUS, keys.toArray(new String[keys.size()]), argv.toArray(new String[argv.size()]));
            }
        } else {
            this.hash().hdel(RedisKey.subscription(clientId), String.join("/", topicLevels));
            this.hash().hdel(RedisKey.topicName(topicLevels), clientId);
        }
    }

    @Override
    public void removeAllSubscriptions(String clientId) {
        this.hash().hgetall(RedisKey.subscription(clientId)).
                forEach((topic, qos) ->
                        removeSubscription(clientId, Topics.sanitize(topic)));
    }

    /**
     * Get possible topic filter tree sub nodes matching the topic
     * Topic Levels must been sanitized
     *
     * @param topicLevels List of topic levels
     * @param index       Current match level
     * @return Possible matching children
     */
    private List<String> getMatchTopicFilter(List<String> topicLevels, int index) {
        if (index == topicLevels.size() - 1) {
            return this.hash().hmget(RedisKey.topicFilterChild(topicLevels.subList(0, index)), END, "#");
        } else {
            return this.hash().hmget(RedisKey.topicFilterChild(topicLevels.subList(0, index)), topicLevels.get(index), "#", "+");
        }
    }

    /**
     * Get and handle all topic filter subscriptions matching the topic
     * This is a recursion method
     * Topic Levels must been sanitized
     *
     * @param topicLevels List of topic levels
     * @param index       Current match level (use 0 if you have doubt)
     * @param map         RETURN VALUE! Subscriptions: Key - Client Id, Value - QoS
     */
    private void getMatchSubscriptions(List<String> topicLevels, int index, Map<String, MqttQoS> map) {
        List<String> children = getMatchTopicFilter(topicLevels, index);

        // last one
        if (children.size() == 2) {
            int c = children.get(0) == null ? 0 : Integer.parseInt(children.get(0)); // char
            int s = children.get(1) == null ? 0 : Integer.parseInt(children.get(1)); // #
            if (c > 0) {
                Map<String, MqttQoS> subscriptions = getTopicSubscriptions(topicLevels);
                if (subscriptions != null) {
                    subscriptions.forEach((clientId, qos) -> {
                        if (qos.value() >= map.getOrDefault(clientId, MqttQoS.AT_MOST_ONCE).value()) {
                            map.put(clientId, qos);
                        }
                    });
                }
            }
            if (s > 0) {
                List<String> newTopicLevels = new ArrayList<>(topicLevels.subList(0, index));
                newTopicLevels.add("#");
                newTopicLevels.add(END);
                Map<String, MqttQoS> subscriptions = getTopicSubscriptions(newTopicLevels);
                if (subscriptions != null) {
                    subscriptions.forEach((clientId, qos) -> {
                        if (qos.value() >= map.getOrDefault(clientId, MqttQoS.AT_MOST_ONCE).value()) {
                            map.put(clientId, qos);
                        }
                    });
                }
            }
        }
        // not last one
        else if (children.size() == 3) {
            int c = children.get(0) == null ? 0 : Integer.parseInt(children.get(0)); // char
            int s = children.get(1) == null ? 0 : Integer.parseInt(children.get(1)); // #
            int p = children.get(2) == null ? 0 : Integer.parseInt(children.get(2)); // +
            if (c > 0) {
                getMatchSubscriptions(topicLevels, index + 1, map);
            }
            if (s > 0) {
                List<String> newTopicLevels = new ArrayList<>(topicLevels.subList(0, index));
                newTopicLevels.add("#");
                newTopicLevels.add(END);
                Map<String, MqttQoS> subscriptions = getTopicSubscriptions(newTopicLevels);
                if (subscriptions != null) {
                    subscriptions.forEach((clientId, qos) -> {
                        if (qos.value() >= map.getOrDefault(clientId, MqttQoS.AT_MOST_ONCE).value()) {
                            map.put(clientId, qos);
                        }
                    });
                }
            }
            if (p > 0) {
                List<String> newTopicLevels = new ArrayList<>(topicLevels);
                newTopicLevels.set(index, "+");
                getMatchSubscriptions(newTopicLevels, index + 1, map);
            }
        }
    }

    @Override
    public void getMatchSubscriptions(List<String> topicLevels, Map<String, MqttQoS> map) {
        if (Topics.isTopicFilter(topicLevels)) {
            throw new IllegalArgumentException("it must be topic name not topic filter");
        }

        // topic name
        map.putAll(getTopicSubscriptions(topicLevels));

        // topic filter
        getMatchSubscriptions(topicLevels, 0, map);
    }

    @Override
    public int addRetainMessage(List<String> topicLevels, Message<MqttPublishVariableHeader, MqttPublishPayload> msg) {
        // retainId
        int retainId = Math.toIntExact(this.script().eval(RedisLua.INCRLIMIT, ScriptOutputType.INTEGER, new String[]{RedisKey.nextRetainId(topicLevels)}, new String[]{"65535"}));

        // retain's message list
        String r = this.script().eval(RedisLua.RPUSHLIMIT, ScriptOutputType.VALUE, new String[]{RedisKey.topicRetainList(topicLevels)}, String.valueOf(retainId), String.valueOf(this.retainQueueSize));
        if (r != null) {
            List<String> keys = new ArrayList<>();
            List<String> argv = new ArrayList<>();
            for (int i = 0; i < topicLevels.size(); i++) {
                keys.add(RedisKey.topicRetainChild(topicLevels.subList(0, i)));
                argv.add(topicLevels.get(i));
            }
            this.script().eval("local length = table.getn(KEYS)\n" +
                            "for i = 1, length do\n" +
                            "   local count = redis.call('HINCRBY', KEYS[i], ARGV[i], -1)\n" +
                            "   if count == 0\n" +
                            "   then\n" +
                            "       redis.call('HDEL', KEYS[i], ARGV[i])\n" +
                            "   end\n" +
                            "end\n" +
                            "return redis.status_reply('OK')",
                    ScriptOutputType.STATUS, keys.toArray(new String[keys.size()]), argv.toArray(new String[argv.size()]));

            this.key().del(RedisKey.topicRemainMessage(topicLevels, retainId));
        }

        // retain tree
        List<String> keys = new ArrayList<>();
        List<String> argv = new ArrayList<>();
        for (int i = 0; i < topicLevels.size(); i++) {
            keys.add(RedisKey.topicRetainChild(topicLevels.subList(0, i)));
            argv.add(topicLevels.get(i));
        }
        this.script().eval("local length = table.getn(KEYS)\n" +
                        "for i = 1, length do\n" +
                        "    redis.call('HINCRBY', KEYS[i], ARGV[i], 1)\n" +
                        "end\n" +
                        "return redis.status_reply('OK')",
                ScriptOutputType.STATUS, keys.toArray(new String[keys.size()]), argv.toArray(new String[argv.size()]));

        // retain message
        this.hash().hmset(RedisKey.topicRemainMessage(topicLevels, retainId), messageToMap(msg));

        return retainId;
    }

    /**
     * Remove the specific retain message
     *
     * @param topicLevels Topic Levels
     * @param retainId    Retain Id
     */
    private void removeRetainMessage(List<String> topicLevels, int retainId) {
        // retain's message list
        long b = this.list().lrem(RedisKey.topicRetainList(topicLevels), 1, String.valueOf(retainId));

        // retain tree
        if (b == 1) {
            List<String> keys = new ArrayList<>();
            List<String> argv = new ArrayList<>();
            for (int i = 0; i < topicLevels.size(); i++) {
                keys.add(RedisKey.topicRetainChild(topicLevels.subList(0, i)));
                argv.add(topicLevels.get(i));
            }
            this.script().eval("local length = table.getn(KEYS)\n" +
                            "for i = 1, length do\n" +
                            "   local count = redis.call('HINCRBY', KEYS[i], ARGV[i], -1)\n" +
                            "   if count == 0\n" +
                            "   then\n" +
                            "       redis.call('HDEL', KEYS[i], ARGV[i])\n" +
                            "   end\n" +
                            "end\n" +
                            "return redis.status_reply('OK')",
                    ScriptOutputType.STATUS, keys.toArray(new String[keys.size()]), argv.toArray(new String[argv.size()]));
        }

        // retain message
        this.key().del(RedisKey.topicRemainMessage(topicLevels, retainId));
    }

    @Override
    public void removeAllRetainMessage(List<String> topicLevels) {
        List<String> ids = this.list().lrange(RedisKey.topicRetainList(topicLevels), 0, -1);
        if (ids != null) {
            ids.forEach(retainId ->
                    removeRetainMessage(topicLevels, Integer.parseInt(retainId)));
        }
    }

    /**
     * Get all retain message topics matching the specific prefix
     * This used to match topic wildcard '#'
     * This is a recursion method
     * Topic Levels must been sanitized
     *
     * @param topicLevels Prefix of retain message
     * @param list        RETURN VALUE! List of retain message topics
     */
    private void getMatchRetainPrefix(List<String> topicLevels, List<List<String>> list) {
        Map<String, String> nodes = this.hash().hgetall(RedisKey.topicRetainChild(topicLevels));
        if (nodes != null) {
            nodes.forEach((node, count) -> {
                int c = Integer.parseInt(count);
                if (c > 0) {
                    List<String> l = new ArrayList<>(topicLevels);
                    l.add(node);
                    if (node.equals(Topics.END)) {
                        list.add(l);
                    } else {
                        getMatchRetainPrefix(l, list);
                    }
                }
            });
        }
    }

    /**
     * Get all retain message topics matching the topic filter
     * This is a recursion method
     * Topic Levels must been sanitized
     *
     * @param topicLevels Topic Filter
     * @param index       Current match level (use 0 if you have doubt)
     * @param list        RETURN VALUE! List of retain message topics
     */
    private void getMatchRetainMessages(List<String> topicLevels, int index, List<List<String>> list) {
        String level = topicLevels.get(index);

        switch (level) {
            case "#":
                List<String> t1 = new ArrayList<>(topicLevels.subList(0, index));
                getMatchRetainPrefix(t1, list);
                break;
            case "+":
                Map<String, String> nodes = this.hash().hgetall(RedisKey.topicRetainChild(topicLevels.subList(0, index)));
                if (nodes != null) {
                    nodes.forEach((node, count) -> {
                        if (!node.equals(Topics.END) && Integer.parseInt(count) > 0) {
                            if (node.equals(Topics.END)) {
                                List<String> t2 = new ArrayList<>(topicLevels.subList(0, index));
                                t2.add(node);
                                list.add(t2);
                            } else {
                                List<String> t2 = new ArrayList<>(topicLevels);
                                t2.set(index, node);
                                getMatchRetainMessages(t2, index + 1, list);
                            }
                        }
                    });
                }
                break;
            default:
                String count = this.hash().hget(RedisKey.topicRetainChild(topicLevels.subList(0, index)), level);
                if (count != null && Integer.parseInt(count) > 0) {
                    if (level.equals(Topics.END) && index == topicLevels.size() - 1) {
                        list.add(topicLevels);
                    } else {
                        getMatchRetainMessages(topicLevels, index + 1, list);
                    }
                }
                break;
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public List<Message<MqttPublishVariableHeader, MqttPublishPayload>> getMatchRetainMessages(List<String> topicLevels) {
        List<Message<MqttPublishVariableHeader, MqttPublishPayload>> r = new ArrayList<>();
        if (Topics.isTopicFilter(topicLevels)) {
            List<List<String>> l = new ArrayList<>();
            getMatchRetainMessages(topicLevels, 0, l);
            l.forEach(t -> {
                List<String> ids = this.list().lrange(RedisKey.topicRetainList(t), 0, -1);
                if (ids != null) {
                    ids.forEach(retainId -> {
                        Message<MqttPublishVariableHeader, MqttPublishPayload> m = mapToMessage(this.hash().hgetall(RedisKey.topicRemainMessage(t, Integer.parseInt(retainId))));
                        if (m != null) r.add(m);
                    });
                }
            });
        } else {
            List<String> ids = this.list().lrange(RedisKey.topicRetainList(topicLevels), 0, -1);
            if (ids != null) {
                ids.forEach(retainId -> {
                    Message<MqttPublishVariableHeader, MqttPublishPayload> m = mapToMessage(this.hash().hgetall(RedisKey.topicRemainMessage(topicLevels, Integer.parseInt(retainId))));
                    if (m != null) r.add(m);
                });
            }
        }

        return r;
    }
}


<|start_filename|>mithqtt-api/src/main/java/io/netty/handler/codec/mqtt/MqttConnectReturnCode.java<|end_filename|>
/*
 * Copyright 2014 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package io.netty.handler.codec.mqtt;

/**
 * Return Code of {@link MqttConnAckMessage}
 */
public enum MqttConnectReturnCode {
    CONNECTION_ACCEPTED((byte) 0x00),
    CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION((byte) 0X01),
    CONNECTION_REFUSED_IDENTIFIER_REJECTED((byte) 0x02),
    CONNECTION_REFUSED_SERVER_UNAVAILABLE((byte) 0x03),
    CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD((byte) 0x04),
    CONNECTION_REFUSED_NOT_AUTHORIZED((byte) 0x05);

    private final byte byteValue;

    MqttConnectReturnCode(byte byteValue) {
        this.byteValue = byteValue;
    }

    public static MqttConnectReturnCode valueOf(byte b) {
        for (MqttConnectReturnCode r : values()) {
            if (r.byteValue == b) {
                return r;
            }
        }
        throw new IllegalArgumentException("unknown connect return code: " + (b & 0xFF));
    }

    public byte byteValue() {
        return byteValue;
    }
}


<|start_filename|>mithqtt-broker/src/main/java/com/github/longkerdandy/mithqtt/broker/cluster/BrokerClusterListenerFactoryImpl.java<|end_filename|>
package com.github.longkerdandy.mithqtt.broker.cluster;

import com.github.longkerdandy.mithqtt.api.cluster.ClusterListener;
import com.github.longkerdandy.mithqtt.api.cluster.ClusterListenerFactory;
import com.github.longkerdandy.mithqtt.broker.session.SessionRegistry;

/**
 * Broker Cluster Listener Factory Implementation
 */
public class BrokerClusterListenerFactoryImpl implements ClusterListenerFactory {

    private final SessionRegistry registry;

    public BrokerClusterListenerFactoryImpl(SessionRegistry registry) {
        this.registry = registry;
    }

    @Override
    public ClusterListener newListener() {
        return new BrokerClusterListenerImpl(registry);
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/exception/AuthorizeException.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.exception;

import com.github.longkerdandy.mithqtt.http.entity.ErrorEntity;

import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Response;

/**
 * Authorize Exception
 */
@SuppressWarnings("unused")
public class AuthorizeException extends WebApplicationException {

    /**
     * Create a HTTP 401 (Unauthorized) exception.
     */
    public AuthorizeException() {
        super(401);
    }

    /**
     * Create a HTTP 401 (Unauthorized) exception.
     *
     * @param entity the error response entity
     */
    public AuthorizeException(ErrorEntity entity) {
        super(Response.status(401).entity(entity).type("application/json").build());
    }
}

<|start_filename|>mithqtt-api/src/main/java/com/github/longkerdandy/mithqtt/util/Topics.java<|end_filename|>
package com.github.longkerdandy.mithqtt.util;

import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * MQTT Topic Utils
 */
@SuppressWarnings("unused")
public class Topics {

    // Present an empty level in the wildcard topic filter
    public static final String EMPTY = "Ø";
    // Present the end of the wildcard topic filter
    public static final String END = "†";

    private Topics() {
    }

    /**
     * Restore sanitized topic to original topic
     *
     * @param topic Topic
     * @return Original Topic Name or Topic Filter
     */
    public static String antidote(String topic) {
        return topic.substring(0, topic.length() - 2).replaceAll(EMPTY, "");
    }

    /**
     * Restore sanitized topic levels to original topic
     *
     * @param topicLevels Topic Levels
     * @return Original Topic Name or Topic Filter
     */
    public static String antidote(List<String> topicLevels) {
        return antidote(String.join("/", topicLevels));
    }

    /**
     * Validate the topic, add EMPTY and END, return as a List of levels
     *
     * @param topic Topic Name or Topic Filter
     * @return List of levels
     */
    public static List<String> sanitize(String topic) {
        if (topic.contains("+") || topic.endsWith("#") || topic.endsWith("#/" + END)) {
            return sanitizeTopicFilter(topic);
        } else {
            return sanitizeTopicName(topic);
        }
    }

    /**
     * Validate the topic name, add EMPTY and END, return as a List of levels
     * No wildcard allowed!
     *
     * @param topicName Topic Name
     * @return List of levels
     */
    public static List<String> sanitizeTopicName(String topicName) {
        if (StringUtils.isEmpty(topicName)) throw new IllegalArgumentException("Empty topic name");
        if (topicName.contains("+")) throw new IllegalArgumentException("Topic name contains wildcard");
        if (topicName.contains("#")) throw new IllegalArgumentException("Topic name contains wildcard");

        List<String> levels = new ArrayList<>();

        if (topicName.startsWith("/")) topicName = EMPTY + topicName;
        if (topicName.endsWith("/")) topicName = topicName + EMPTY;

        String[] tokens = StringUtils.splitByWholeSeparatorPreserveAllTokens(topicName, "/");
        for (String token : tokens) {
            levels.add(StringUtils.isNotEmpty(token) ? token : EMPTY);
        }

        if (!topicName.endsWith(END)) levels.add(END);

        return levels;
    }

    /**
     * Validate the topic filter, add EMPTY and END, return as a List of levels
     *
     * @param topicFilter Topic Filter
     * @return List of levels
     */
    public static List<String> sanitizeTopicFilter(String topicFilter) {
        if (StringUtils.isEmpty(topicFilter)) throw new IllegalArgumentException("Empty topic filer");
        if (!topicFilter.contains("+") && !topicFilter.contains("#"))
            throw new IllegalArgumentException("Topic filter does not contain wildcard");

        List<String> levels = new ArrayList<>();

        if (topicFilter.startsWith("/")) topicFilter = EMPTY + topicFilter;
        if (topicFilter.endsWith("/")) topicFilter = topicFilter + EMPTY;

        String[] tokens = StringUtils.splitByWholeSeparatorPreserveAllTokens(topicFilter, "/");
        for (int i = 0; i < tokens.length; i++) {
            if (tokens[i].contains("+") && !tokens[i].equals("+"))
                throw new IllegalArgumentException("Illegal topic filter: " + topicFilter);
            if (tokens[i].contains("#") && !tokens[i].equals("#"))
                throw new IllegalArgumentException("Illegal topic filter: " + topicFilter);
            if (tokens[i].equals("#") && !(i == tokens.length - 1 || (i == tokens.length - 2 && END.equals(tokens[i + 1]))))
                throw new IllegalArgumentException("Illegal topic filter: " + topicFilter);
            levels.add(StringUtils.isNotEmpty(tokens[i]) ? tokens[i] : EMPTY);
        }

        if (!topicFilter.endsWith(END)) levels.add(END);

        return levels;
    }

    /**
     * Is sanitized topic levels a topic filter (contains wildcard)
     *
     * @param topicLevels Sanitized Topic Levels
     * @return True if is topic filter
     */
    public static boolean isTopicFilter(List<String> topicLevels) {
        return topicLevels.contains("+") || topicLevels.get(topicLevels.size() - 2).equals("#");
    }
}


<|start_filename|>mithqtt-api/src/test/java/com/github/longkerdandy/mithqtt/util/TopicsTest.java<|end_filename|>
package com.github.longkerdandy.mithqtt.util;

import org.junit.Test;

import java.util.Arrays;

import static com.github.longkerdandy.mithqtt.util.Topics.EMPTY;
import static com.github.longkerdandy.mithqtt.util.Topics.END;

/**
 * MQTT Topic Utils Test
 */
public class TopicsTest {

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail00() {
        Topics.sanitizeTopicFilter(null);
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail01() {
        Topics.sanitizeTopicFilter("");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail02() {
        Topics.sanitizeTopicFilter("abc/def/g+/h");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail03() {
        Topics.sanitizeTopicFilter("abc/def#/g/h");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail05() {
        Topics.sanitizeTopicFilter("abc/def/g/#/h");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicFilterFail06() {
        Topics.sanitizeTopicFilter("abc/def/g/#/");
    }

    @Test
    public void sanitizeTopicFilterTest() {
        assert Arrays.equals(Topics.sanitizeTopicFilter("abc/+/g/h").toArray(), new String[]{"abc", "+", "g", "h", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("abc/def/g/#").toArray(), new String[]{"abc", "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("abc/def/#").toArray(), new String[]{"abc", "def", "#", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("+/+/g/#").toArray(), new String[]{"+", "+", "g", "#", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("/abc/def/g/#").toArray(), new String[]{EMPTY, "abc", "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("/abc//def/g/#").toArray(), new String[]{EMPTY, "abc", EMPTY, "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitizeTopicFilter("/abc/+/g/h/").toArray(), new String[]{EMPTY, "abc", "+", "g", "h", EMPTY, END});
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicNameFail00() {
        Topics.sanitizeTopicName(null);
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicNameFail01() {
        Topics.sanitizeTopicName("");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicNameFail02() {
        Topics.sanitizeTopicName("abc/def/g/+/h");
    }

    @Test(expected = IllegalArgumentException.class)
    public void sanitizeTopicNameFail03() {
        Topics.sanitizeTopicName("abc/def/g/h/#");
    }

    @Test
    public void sanitizeTopicNameTest() {
        assert Arrays.equals(Topics.sanitizeTopicName("abc/def/g/h").toArray(), new String[]{"abc", "def", "g", "h", END});
        assert Arrays.equals(Topics.sanitizeTopicName("/abc/def/g/h").toArray(), new String[]{EMPTY, "abc", "def", "g", "h", END});
        assert Arrays.equals(Topics.sanitizeTopicName("/abc/def/g/h/").toArray(), new String[]{EMPTY, "abc", "def", "g", "h", EMPTY, END});
        assert Arrays.equals(Topics.sanitizeTopicName("/abc/def//g/h").toArray(), new String[]{EMPTY, "abc", "def", EMPTY, "g", "h", END});
    }

    @Test
    public void sanitizeTest() {
        assert Arrays.equals(Topics.sanitize("abc/+/g/h").toArray(), new String[]{"abc", "+", "g", "h", END});
        assert Arrays.equals(Topics.sanitize("abc/def/g/#").toArray(), new String[]{"abc", "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitize("abc/def/#").toArray(), new String[]{"abc", "def", "#", END});
        assert Arrays.equals(Topics.sanitize("+/+/g/#").toArray(), new String[]{"+", "+", "g", "#", END});
        assert Arrays.equals(Topics.sanitize("/abc/def/g/#").toArray(), new String[]{EMPTY, "abc", "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitize("/abc//def/g/#").toArray(), new String[]{EMPTY, "abc", EMPTY, "def", "g", "#", END});
        assert Arrays.equals(Topics.sanitize("/abc/+/g/h/").toArray(), new String[]{EMPTY, "abc", "+", "g", "h", EMPTY, END});

        assert Arrays.equals(Topics.sanitize("abc/def/g/h").toArray(), new String[]{"abc", "def", "g", "h", END});
        assert Arrays.equals(Topics.sanitize("/abc/def/g/h").toArray(), new String[]{EMPTY, "abc", "def", "g", "h", END});
        assert Arrays.equals(Topics.sanitize("/abc/def/g/h/").toArray(), new String[]{EMPTY, "abc", "def", "g", "h", EMPTY, END});
        assert Arrays.equals(Topics.sanitize("/abc/def//g/h").toArray(), new String[]{EMPTY, "abc", "def", EMPTY, "g", "h", END});
    }

    @Test
    public void isTopicFilterTest() {
        assert Topics.isTopicFilter(Arrays.asList("abc", "+", "g", "h", END));
        assert Topics.isTopicFilter(Arrays.asList("abc", "def", "g", "#", END));
        assert Topics.isTopicFilter(Arrays.asList("abc", "def", "#", END));
        assert Topics.isTopicFilter(Arrays.asList("+", "+", "g", "#", END));
        assert !Topics.isTopicFilter(Arrays.asList("abc", "def", "g", "h", END));
        assert !Topics.isTopicFilter(Arrays.asList(EMPTY, "abc", "def", "g", "h", END));
    }

    @Test
    public void antidoteTest() {
        assert Topics.antidote(Arrays.asList(new String[]{"abc", "+", "g", "h", END})).equals("abc/+/g/h");
        assert Topics.antidote(Arrays.asList(new String[]{"abc", "def", "g", "#", END})).equals("abc/def/g/#");
        assert Topics.antidote(Arrays.asList(new String[]{"abc", "def", "#", END})).equals("abc/def/#");
        assert Topics.antidote(Arrays.asList(new String[]{"+", "+", "g", "#", END})).equals("+/+/g/#");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", "def", "g", "#", END})).equals("/abc/def/g/#");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", EMPTY, "def", "g", "#", END})).equals("/abc//def/g/#");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", "+", "g", "h", EMPTY, END})).equals("/abc/+/g/h/");

        assert Topics.antidote(Arrays.asList(new String[]{"abc", "def", "g", "h", END})).equals("abc/def/g/h");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", "def", "g", "h", END})).equals("/abc/def/g/h");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", "def", "g", "h", EMPTY, END})).equals("/abc/def/g/h/");
        assert Topics.antidote(Arrays.asList(new String[]{EMPTY, "abc", "def", EMPTY, "g", "h", END})).equals("/abc/def//g/h");
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/resources/AbstractResource.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.resources;

import com.github.longkerdandy.mithqtt.api.auth.Authenticator;
import com.github.longkerdandy.mithqtt.api.cluster.Cluster;
import com.github.longkerdandy.mithqtt.api.storage.sync.SyncStorage;
import com.github.longkerdandy.mithqtt.http.util.Validator;

/**
 * Abstract Base Resource
 */
public abstract class AbstractResource {

    protected final String serverId;
    protected final Validator validator;
    protected final SyncStorage storage;
    protected final Cluster cluster;
    protected final Authenticator authenticator;

    public AbstractResource(String serverId, Validator validator, SyncStorage storage, Cluster cluster, Authenticator authenticator) {
        this.serverId = serverId;
        this.validator = validator;
        this.storage = storage;
        this.cluster = cluster;
        this.authenticator = authenticator;
    }
}


<|start_filename|>mithqtt-storage-redis/src/main/java/com/github/longkerdandy/mithqtt/storage/redis/util/Converter.java<|end_filename|>
package com.github.longkerdandy.mithqtt.storage.redis.util;

import com.github.longkerdandy.mithqtt.api.message.Message;
import com.github.longkerdandy.mithqtt.api.message.MqttAdditionalHeader;
import com.github.longkerdandy.mithqtt.api.message.MqttPublishPayload;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.lang3.BooleanUtils;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

/**
 * Converter Utils
 */
public class Converter {

    /**
     * Convert Map to (MQTT) Message
     *
     * @param map Map
     * @return (MQTT) Message
     */
    public static Message mapToMessage(Map<String, String> map) {
        if (map == null || map.isEmpty()) return null;

        int type = Integer.parseInt(map.get("type"));
        if (type == MqttMessageType.PUBLISH.value()) {
            byte[] bytes = new byte[0];
            if (map.get("payload") != null) try {
                bytes = map.get("payload").getBytes("ISO-8859-1");
            } catch (UnsupportedEncodingException ignore) {
            }
            int pid = Integer.parseInt(map.getOrDefault("packetId", "0"));
            return new Message<>(
                    new MqttFixedHeader(
                            MqttMessageType.PUBLISH,
                            BooleanUtils.toBoolean(map.getOrDefault("dup", "0"), "1", "0"),
                            MqttQoS.valueOf(Integer.parseInt(map.getOrDefault("qos", "0"))),
                            BooleanUtils.toBoolean(map.getOrDefault("retain", "0"), "1", "0"),
                            0
                    ),
                    new MqttAdditionalHeader(
                            MqttVersion.valueOf(map.getOrDefault("version", MqttVersion.MQTT_3_1_1.toString())),
                            map.get("clientId"),
                            map.get("userName"),
                            null
                    ),
                    pid > 0 ? MqttPublishVariableHeader.from(map.get("topicName"), pid)
                            : MqttPublishVariableHeader.from(map.get("topicName")),
                    new MqttPublishPayload(
                            bytes
                    ));
        } else if (type == MqttMessageType.PUBREL.value()) {
            return new Message<>(
                    new MqttFixedHeader(
                            MqttMessageType.PUBREL,
                            false,
                            MqttQoS.AT_LEAST_ONCE,
                            false,
                            0
                    ),
                    new MqttAdditionalHeader(
                            MqttVersion.valueOf(map.getOrDefault("version", MqttVersion.MQTT_3_1_1.toString())),
                            map.get("clientId"),
                            map.get("userName"),
                            null
                    ),
                    MqttPacketIdVariableHeader.from(
                            Integer.parseInt(map.getOrDefault("packetId", "0"))
                    ),
                    null
            );
        } else {
            throw new IllegalArgumentException("Invalid in-flight MQTT message type: " + MqttMessageType.valueOf(type));
        }
    }

    /**
     * Convert (MQTT) Message to Map
     *
     * @param msg (MQTT) Message
     * @return Map
     */
    public static Map<String, String> messageToMap(Message msg) {
        Map<String, String> map = new HashMap<>();
        if (msg == null) return map;

        if (msg.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
            MqttPublishVariableHeader variableHeader = (MqttPublishVariableHeader) msg.variableHeader();
            MqttPublishPayload payload = (MqttPublishPayload) msg.payload();
            map.put("type", String.valueOf(MqttMessageType.PUBLISH.value()));
            map.put("retain", BooleanUtils.toString(msg.fixedHeader().retain(), "1", "0"));
            map.put("qos", String.valueOf(msg.fixedHeader().qos().value()));
            map.put("dup", BooleanUtils.toString(msg.fixedHeader().dup(), "1", "0"));
            map.put("version", msg.additionalHeader().version().toString());
            if (!msg.fixedHeader().retain()) map.put("clientId", msg.additionalHeader().clientId());
            map.put("userName", msg.additionalHeader().userName());
            map.put("topicName", variableHeader.topicName());
            if (!msg.fixedHeader().retain()) map.put("packetId", String.valueOf(variableHeader.packetId()));
            if (payload != null && payload.bytes() != null && payload.bytes().length > 0) try {
                map.put("payload", new String(payload.bytes(), "ISO-8859-1"));
            } catch (UnsupportedEncodingException ignore) {
            }
            return map;
        } else if (msg.fixedHeader().messageType() == MqttMessageType.PUBREL) {
            MqttPacketIdVariableHeader variableHeader = (MqttPacketIdVariableHeader) msg.variableHeader();
            map.put("type", String.valueOf(MqttMessageType.PUBREL.value()));
            map.put("version", msg.additionalHeader().version().toString());
            map.put("clientId", msg.additionalHeader().clientId());
            map.put("userName", msg.additionalHeader().userName());
            map.put("packetId", String.valueOf(variableHeader.packetId()));
            return map;
        } else {
            throw new IllegalArgumentException("Invalid in-flight MQTT message type: " + msg.fixedHeader().messageType());
        }
    }
}


<|start_filename|>mithqtt-http/src/main/java/com/github/longkerdandy/mithqtt/http/entity/ErrorCode.java<|end_filename|>
package com.github.longkerdandy.mithqtt.http.entity;


/**
 * Error Code
 */
@SuppressWarnings("unused")
public class ErrorCode {

    public static final int UNAUTHORIZED = 1;

    public static final int INVALID = 11;

    public static final int UNSUPPORTED = 21;

    private ErrorCode() {
    }
}