PublishReceiver.java

/*
 * Copyright 2016 The Lannister Project
 * 
 * Licensed 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 net.anyflow.lannister.packetreceiver;

import java.util.Date;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import net.anyflow.lannister.AbnormalDisconnectEventArgs;
import net.anyflow.lannister.Statistics;
import net.anyflow.lannister.message.InboundMessageStatus;
import net.anyflow.lannister.message.Message;
import net.anyflow.lannister.plugin.DisconnectEventListener;
import net.anyflow.lannister.plugin.IMessage;
import net.anyflow.lannister.plugin.Plugins;
import net.anyflow.lannister.plugin.PublishEventArgs;
import net.anyflow.lannister.plugin.PublishEventListener;
import net.anyflow.lannister.session.Session;
import net.anyflow.lannister.topic.Topic;
import net.anyflow.lannister.topic.TopicMatcher;

@Sharable
public class PublishReceiver extends SimpleChannelInboundHandler<MqttPublishMessage> {
	private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PublishReceiver.class);
	public static final PublishReceiver INSTANCE = new PublishReceiver();

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttPublishMessage msg) throws Exception {
		logger.debug("packet incoming [message={}]", msg.toString());
		Statistics.INSTANCE.add(Statistics.Criterion.MESSAGES_PUBLISH_RECEIVED, 1);

		Session session = Session.NEXUS.get(ctx.channel().id());
		if (session == null) {
			logger.error("None exist session message [message={}]", msg.toString());

			ctx.channel().disconnect().addListener(ChannelFutureListener.CLOSE).addListener(fs -> // [MQTT-4.8.0-1]
			Plugins.INSTANCE.get(DisconnectEventListener.class).disconnected(new AbnormalDisconnectEventArgs()));
			return;
		}

		session.setLastIncomingTime(new Date());

		if (!TopicMatcher.isValid(msg.variableHeader().topicName(), false)) {
			session.dispose(true);
			return;
		}

		Message message = Message.newMessage(session.clientId(), msg);

		if (!Plugins.INSTANCE.get(PublishEventListener.class).allowPublish(new PublishEventArgs() {
			@Override
			public IMessage message() {
				return message;
			}
		})) {
			session.dispose(true);
			return;
		}

		// TODO What to do when sender re-publish message corrensponds to
		// unacked status?

		// TODO Until it has received the corresponding PUBREL packet, the
		// Receiver MUST acknowledge any subsequent PUBLISH packet with the same
		// Packet Identifier by sending a PUBREC. It MUST NOT cause duplicate
		// messages to be delivered to any onward recipients in this
		// case.[MQTT-4.3.3-2].

		final Topic topic = Topic.NEXUS.prepare(message);

		if (message.isRetain()) {// else do nothing [MQTT-3.3.1-12]
			topic.setRetainedMessage(message); // [MQTT-3.3.1-5]
		}

		message.setRetain(false);

		topic.publish(message);

		MqttMessage toSend;
		final String log;

		switch (msg.fixedHeader().qosLevel()) {
		case AT_MOST_ONCE:
			return; // QoS 0 do not send any acknowledge packet [MQTT-3.3.4-1]

		case AT_LEAST_ONCE:
			toSend = MqttMessageFactory.puback(msg.variableHeader().messageId());
			log = toSend.toString();

			session.send(toSend, f -> {
				if (!f.isSuccess()) {
					logger.error("packet outgoing failed [{}] {}", log, f.cause());
					return;
				}

				InboundMessageStatus.NEXUS.removeByKey(msg.variableHeader().messageId(), session.clientId());
			}); // [MQTT-3.3.4-1],[MQTT-2.3.1-6]
			logger.debug("Inbound message status REMOVED [clientId={}, messageId={}]", session.clientId(),
					msg.variableHeader().messageId());
			return;

		case EXACTLY_ONCE:
			toSend = MqttMessageFactory.pubrec(msg.variableHeader().messageId());
			log = toSend.toString();

			session.send(MqttMessageFactory.pubrec(msg.variableHeader().messageId()), f -> {
				if (!f.isSuccess()) {
					logger.error("packet outgoing failed [{}] {}", log, f.cause());
					return;
				}

				InboundMessageStatus.NEXUS.update(msg.variableHeader().messageId(), session.clientId(),
						InboundMessageStatus.Status.PUBRECED);
			}); // [MQTT-3.3.4-1],[MQTT-2.3.1-6]
			return;

		default:
			session.dispose(true); // [MQTT-3.3.1-4]
			return;
		}
	}
}