MqttWebTest.java

/*
 * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
 *
 * 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.dreamlu.iot.mqtt.http;

import net.dreamlu.iot.mqtt.core.server.MqttServer;
import net.dreamlu.iot.mqtt.core.server.MqttServerCreator;
import net.dreamlu.iot.mqtt.core.server.dispatcher.IMqttMessageDispatcher;
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
import net.dreamlu.iot.mqtt.core.server.http.api.MqttHttpApi;
import net.dreamlu.iot.mqtt.core.server.http.api.auth.BasicAuthFilter;
import net.dreamlu.iot.mqtt.core.server.http.core.MqttWebServer;
import net.dreamlu.iot.mqtt.core.server.http.handler.MqttHttpRoutes;
import net.dreamlu.iot.mqtt.core.server.model.Message;
import net.dreamlu.iot.mqtt.core.server.support.DefaultMqttMessageDispatcher;
import net.dreamlu.iot.mqtt.core.server.websocket.MqttWsMsgHandler;
import org.tio.core.intf.AioHandler;
import org.tio.server.ServerTioConfig;
import org.tio.websocket.server.handler.IWsMsgHandler;

/**
 * mqtt websocket 子协议测试
 */
public class MqttWebTest {

	public static void main(String[] args) throws Exception {
		// 1. 消息转发处理器，可用来实现集群
		IMqttMessageDispatcher messageDispatcher = new DefaultMqttMessageDispatcher();
		// 2. 收到消息，将消息转发出去
		IMqttMessageListener messageListener = (clientId, topic, mqttQoS, payload) -> {
			Message message = new Message();
			message.setTopic(topic);
			message.setQos(mqttQoS.value());
			message.setPayload(payload.array());
			messageDispatcher.send(message);
		};

		// 3. 启动服务
		MqttServerCreator serverCreator = MqttServer.create()
			.ip("0.0.0.0")
			.port(1883)
			.readBufferSize(512)
			.messageDispatcher(messageDispatcher)
			.messageListener(messageListener)
			.websocketEnable(false)
			.debug();
		MqttServer mqttServer = serverCreator.start();
		ServerTioConfig serverConfig = mqttServer.getServerConfig();
		AioHandler aioHandler = serverConfig.getAioHandler();

		MqttHttpApi httpApi = new MqttHttpApi(serverCreator.getMessageDispatcher(), serverCreator.getSessionManager());
		httpApi.register();
		MqttHttpRoutes.addFilter(new BasicAuthFilter("123", "123"));

		IWsMsgHandler mqttWsMsgHandler = new MqttWsMsgHandler(aioHandler);
		MqttWebServer httpServerStarter = new MqttWebServer(serverCreator, mqttWsMsgHandler);
		ServerTioConfig httpIioConfig = httpServerStarter.getServerTioConfig();
		httpIioConfig.share(serverConfig);
		// 启动http服务器
		httpServerStarter.start();
	}

}