@SuppressWarnings("unused")
public interface Authenticator {

    /**
     * Init the authenticator
     *
     * @param config Authenticator Configuration
     */
    void init(AbstractConfiguration config);

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

    /**
     * Authorize client CONNECT
     *
     * @param clientId Client Id
     * @param userName User Name
     * @param password Password
     * @return Authorize Result
     */
    AuthorizeResult authConnect(String clientId, String userName, String password);

    /**
     * Authorize client PUBLISH
     *
     * @param clientId  Client Id
     * @param userName  User Name
     * @param topicName Topic Name
     * @param qos       QoS
     * @param retain    Retain
     * @return Authorize Result
     */
    AuthorizeResult authPublish(String clientId, String userName, String topicName, int qos, boolean retain);

    /**
     * Authorize client SUBSCRIBE
     *
     * @param clientId             Client Id
     * @param userName             User Name
     * @param requestSubscriptions List of request Topic Subscription
     * @return List of granted QoS
     */
    List<MqttGrantedQoS> authSubscribe(String clientId, String userName, List<MqttTopicSubscription> requestSubscriptions);

    /**
     * OAuth in HTTP request
     *
     * @param credentials OAuth2 bearer-token
     * @return User Name
     */
    String oauth(String credentials);
} 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);
} 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() : "")
                + ']';
    }
} 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
                + ']';
    }
} 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)
                + ']';
    }
} 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
                + ']';
    }
} @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);
} @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() {
    }
} @SuppressWarnings("unused")
public class ErrorEntity<T> {

    private int code;
    private String message;
    private T data;

    public ErrorEntity(int code) {
        this(code, null);
    }

    public ErrorEntity(int code, T data) {
        this.code = code;
        this.data = data;
    }

    public int getCode() {
        return code;
    }

    public void setCode(int code) {
        this.code = code;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }
} @SuppressWarnings("unused")
public class ResultEntity<T> {

    private T result;

    public ResultEntity(T result) {
        this.result = result;
    }

    public T getResult() {
        return result;
    }

    public void setResult(T result) {
        this.result = result;
    }
} public class Subscription {

    private String topic;
    private int qos;

    public Subscription() {
    }

    public Subscription(String topic, int qos) {
        this.topic = topic;
        this.qos = qos;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getQos() {
        return qos;
    }

    public void setQos(int qos) {
        this.qos = qos;
    }
} @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());
    }
} @SuppressWarnings("unused")
public class ValidateException extends WebApplicationException {

    /**
     * Create a HTTP 422 (UnProcessable Entity) exception.
     */
    public ValidateException() {
        super(422);
    }

    /**
     * Create a HTTP 422 (UnProcessable Entity) exception.
     *
     * @param entity the error response entity
     */
    public ValidateException(ErrorEntity entity) {
        super(Response.status(422).entity(entity).type("application/json").build());
    }
} @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();
    }
} @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("#");
    }
}
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.message.MqttSubscribePayloadGranted;
import com.github.longkerdandy.mithqtt.api.message.MqttTopicSubscriptionGranted;
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.entity.Subscription;
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.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.ArrayList;
import java.util.List;
import java.util.Map;
package com.github.longkerdandy.mithqtt.http.resources;



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

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

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

    /**
     * Handle MQTT Subscribe Request in RESTful style
     * Granted QoS Levels will send back to client.
     * Retain Messages matched the subscriptions will NOT send back to client.
     */
    @PermitAll
    @POST
    public ResultEntity<List<MqttGrantedQoS>> subscribe(@PathParam("clientId") String clientId, @Auth UserPrincipal user, @QueryParam("protocol") @DefaultValue("4") byte protocol,
                                                        @QueryParam("packetId") @DefaultValue("0") int packetId,
                                                        List<Subscription> subscriptions) {
        String userName = user.getName();
        MqttVersion version = MqttVersion.fromProtocolLevel(protocol);
        List<MqttTopicSubscription> requestSubscriptions = new ArrayList<>();
        List<MqttTopicSubscriptionGranted> grantedSubscriptions = new ArrayList<>();

        // 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 (Subscription subscription : subscriptions) {
            if (!this.validator.isTopicFilterValid(subscription.getTopic())) {
                logger.debug("Protocol violation: Client {} subscription {} is not valid based on configuration", clientId, subscription.getTopic());
                throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
            }
            MqttQoS requestQos;
            try {
                requestQos = MqttQoS.valueOf(subscription.getQos());
            } catch (IllegalArgumentException e) {
                logger.debug("Protocol violation: Client {} subscription qos {} is not valid", clientId, subscription.getQos());
                throw new ValidateException(new ErrorEntity(ErrorCode.INVALID));
            }
            requestSubscriptions.add(new MqttTopicSubscription(subscription.getTopic(), requestQos));
        }

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

        // Authorize client subscribe using provided Authenticator
        List<MqttGrantedQoS> grantedQosLevels = this.authenticator.authSubscribe(clientId, userName, requestSubscriptions);
        if (subscriptions.size() != grantedQosLevels.size()) {
            logger.warn("Authorization error: SUBSCRIBE message's subscriptions count not equal to granted QoS count");
            throw new AuthorizeException(new ErrorEntity(ErrorCode.UNAUTHORIZED));
        }
        logger.trace("Authorization granted on topic {} as {} for client {}", ArrayUtils.toString(requestSubscriptions), ArrayUtils.toString(grantedQosLevels), clientId);

        for (int i = 0; i < requestSubscriptions.size(); i++) {

            MqttGrantedQoS grantedQoS = grantedQosLevels.get(i);
            String topic = requestSubscriptions.get(i).topic();
            List<String> topicLevels = Topics.sanitize(topic);
            grantedSubscriptions.add(new MqttTopicSubscriptionGranted(topic, grantedQoS));

            // Granted only
            if (grantedQoS != MqttGrantedQoS.NOT_GRANTED) {

                // If a Server receives a SUBSCRIBE Packet containing a Topic Filter that is identical to an existing
                // Subscription’s Topic Filter then it MUST completely replace that existing Subscription with a new
                // Subscription. The Topic Filter in the new Subscription will be identical to that in the previous Subscription,
                // although its maximum QoS value could be different.
                logger.trace("Update subscription: Update client {} subscription with topic {} QoS {}", clientId, topic, grantedQoS);
                this.storage.updateSubscription(clientId, topicLevels, MqttQoS.valueOf(grantedQoS.value()));
            }
        }

        // Pass message to 3rd party application
        Message<MqttPacketIdVariableHeader, MqttSubscribePayloadGranted> msg = new Message<>(