private Future<Void> send(
        final String tenantId,
        final String deviceId,
        final Buffer payload,
        final boolean useShortTopicName,
        final BiConsumer<AsyncResult<Integer>, Promise<Void>> sendAttemptHandler) {

    final String topic = String.format(
            TOPIC_TEMPLATE,
            useShortTopicName ? EventConstants.EVENT_ENDPOINT_SHORT : EventConstants.EVENT_ENDPOINT,
            tenantId,
            deviceId);
    final Promise<Void> result = Promise.promise();
    mqttClient.publish(
            topic,
            payload,
            MqttQoS.AT_LEAST_ONCE,
            false, // is duplicate
            false, // is retained
            sendAttempt -> sendAttemptHandler.accept(sendAttempt, result));
    return result.future();
}
 

@Override
public Optional<MqttMessage> process(MqttMessage req, String clientId, String remoteIp) throws BrokerException {
    MqttPublishMessage msg = (MqttPublishMessage) req;
    log.info("PUBLISH, {} Qos: {}", msg.variableHeader().topicName(), msg.fixedHeader().qosLevel());

    switch (msg.fixedHeader().qosLevel()) {
        case AT_MOST_ONCE: {
            this.sessionStore.publishMessage(msg, false);
            return Optional.empty();
        }

        case AT_LEAST_ONCE: {
            boolean result = this.sessionStore.publishMessage(msg, false);
            MqttQoS qos = result ? MqttQoS.AT_LEAST_ONCE : MqttQoS.FAILURE;
            MqttMessage rsp = MqttMessageFactory.newMessage(new MqttFixedHeader(MqttMessageType.PUBACK, false, qos, false, ProtocolProcess.fixLengthOfMessageId),
                    MqttMessageIdVariableHeader.from(msg.variableHeader().packetId()), null);
            return Optional.of(rsp);
        }

        case EXACTLY_ONCE:
        default: {
            log.error("DOT NOT support Qos=2, close");
            throw new BrokerException(ErrorCode.MQTT_NOT_SUPPORT_QOS2);
        }
    }
}
 

@Test
public void testMatches() throws Exception {
	String testTopic = "testTopic/test";
	Session session = new Session("1", "1", 1, 50, true, null);

	TopicSubscription ts0 = new TopicSubscription(session.clientId(), "testTopic/#", MqttQoS.AT_MOST_ONCE);
	TopicSubscription ts1 = new TopicSubscription(session.clientId(), "testTopic/+", MqttQoS.AT_LEAST_ONCE);
	TopicSubscription ts2 = new TopicSubscription(session.clientId(), testTopic, MqttQoS.EXACTLY_ONCE);

	TopicSubscription.NEXUS.put(ts0);
	TopicSubscription.NEXUS.put(ts1);
	TopicSubscription.NEXUS.put(ts2);

	Assert.assertEquals(3, TopicSubscription.NEXUS.topicFiltersOf(session.clientId()).size());

	TopicSubscription target = session.matches(testTopic);

	Assert.assertTrue(target.topicFilter().equals(testTopic));
}
 

@Test
public void testSubscribeWildcard() {
    try {
        MessageListener listener = new MessageListener();
        IMqttToken token = this.mqttClient.subscribeWithResponse("#", listener);
        token.waitForCompletion();

        Assert.assertEquals(token.getGrantedQos()[0], MqttQoS.AT_LEAST_ONCE.value());

        MqttMessage message = new MqttMessage(this.content.getBytes(StandardCharsets.UTF_8));
        this.mqttClient.publish(this.topicName, message);

        Thread.sleep(this.actionTimeout);
        Assert.assertTrue(listener.received > 0);
    } catch (Exception e) {
        log.error("exception", e);
        Assert.fail();
    }
}
 

/**
 * Example for handling client published message
 * @param endpoint
 */
public void example6(MqttEndpoint endpoint) {

  // handling incoming published messages
  endpoint.publishHandler(message -> {

    System.out.println("Just received message [" + message.payload().toString(Charset.defaultCharset()) + "] with QoS [" + message.qosLevel() + "]");

    if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
      endpoint.publishAcknowledge(message.messageId());
    } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
      endpoint.publishReceived(message.messageId());
    }

  }).publishReleaseHandler(messageId -> {

    endpoint.publishComplete(messageId);
  });
}
 

public void processPublish(Channel channel, BorkerMessage bMsgInfo) {
	MqttQoS qosLevel = MqttQoS.valueOf(bMsgInfo.getIQosLevel());
	String clientId = NettyUtil.getClientId(channel);
	int packetId = bMsgInfo.getSourceMsgId();
	String topicName = bMsgInfo.getTopicName();
	byte[] msgBytes = bMsgInfo.getMsgBytes();

	boolean bSaveRelMsg = getMustStoreRelMessage(qosLevel);
	if (bSaveRelMsg) {
		ProcedureMessage relMsgObj = ProcedureMessage.builder().sourceClientId(clientId)
				.sourceMsgId(packetId).topicName(topicName).iQosLevel(qosLevel.value()).msgBytes(msgBytes)
				//.borkerMsgId(bMsgInfo.getBorkerMsgId())
				.build();
		procedureData.putPubRelMessage(clientId, relMsgObj);
	}
}
 

private void sendSubscribMessage(String clientId, Channel channel, MqttQoS respQoS, boolean bSavePubMsg,
		boolean bSend, BorkerMessage bMsgInfo) {

	if ((!bSavePubMsg) && (!bSend)) {
		return ;
	}
	
	String topicName = bMsgInfo.getTopicName();
	boolean isDup = bMsgInfo.isDup();
	boolean isRetain = bMsgInfo.isRetain();
	byte[] msgBytes = bMsgInfo.getMsgBytes();
	int bMsgId =  getMustSendMessageId(respQoS, clientId); 

	if (bSavePubMsg) {
		ConsumerMessage pubMsgObj = ConsumerMessage.builder().sourceClientId(clientId).topic(topicName)
				.mqttQoS(respQoS.value()).messageBytes(msgBytes).messageId(bMsgId).build();
		consumerData.putPublishMessage(clientId, pubMsgObj);
	}

	if (bSend && channel != null) {
		MqttPublishMessage publishMessage = ProtocolUtil.publishMessage(topicName, isDup, respQoS.value(), isRetain,
				bMsgId, msgBytes);
		
		this.sendProcess.sendPublishMessage(channel, publishMessage);
	}
}
 

@Override
protected Future<Void> send(
        final String tenantId,
        final String deviceId,
        final Buffer payload,
        final boolean useShortTopicName) {

    final String topic = String.format(
            TOPIC_TEMPLATE,
            useShortTopicName ? TelemetryConstants.TELEMETRY_ENDPOINT_SHORT : TelemetryConstants.TELEMETRY_ENDPOINT,
            tenantId,
            deviceId);
    final Promise<Void> result = Promise.promise();
    mqttClient.publish(
            topic,
            payload,
            MqttQoS.AT_LEAST_ONCE,
            false, // is duplicate
            false, // is retained
            sendAttempt -> handlePublishAttempt(sendAttempt, result));
    return result.future();
}
 

public static MqttConnectMessage connect(ConnectOptions options) {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false,
			10);
	MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(options.version().protocolName(),
			options.version().protocolLevel(), options.userName() != null, options.password() != null,
			options.will() == null ? false : options.will().isRetain(),
			options.will() == null ? 0 : options.will().qos().value(), options.will() != null,
			options.cleanSession(), options.keepAliveTimeSeconds());

	MqttConnectPayload payload = new MqttConnectPayload(Strings.nullToEmpty(options.clientId()),
			options.will() == null ? "" : options.will().topicName(),
			options.will() == null ? "" : new String(options.will().message(), CharsetUtil.UTF_8),
			Strings.nullToEmpty(options.userName()), Strings.nullToEmpty(options.password()));

	return new MqttConnectMessage(fixedHeader, variableHeader, payload);
}
 

@Override
public MqttEndpointImpl publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain, int messageId, Handler<AsyncResult<Integer>> publishSentHandler) {
  Future<Integer> fut = publish(topic, payload, qosLevel, isDup, isRetain, messageId);
  if (publishSentHandler != null) {
    fut.onComplete(publishSentHandler);
  }
  return this;
}
 

public   MqttPendingPublish(int messageId, Promise<Void> future, ByteBuf payload, MqttPublishMessage message, MqttQoS qos) {
    this.messageId = messageId;
    this.future = future;
    this.payload = payload;
    this.message = message;
    this.qos = qos;

    this.publishRetransmissionHandler.setOriginalMessage(message);
}
 

public static void main(String[] args) {
    Verticle verticle = new AbstractVerticle() {
        @Override
        public void start() {
            MqttClient client = MqttClient.create(vertx, new MqttClientOptions()
                    //开启遗言
                    .setWillFlag(true)
                    .setWillTopic("willTopic")
                    .setWillMessage("hello")

                    .setUsername("admin")
                    .setPassword("123456")
                    .setMaxMessageSize(8192));


            client.connect(PORT,HOST, asyncResult -> {
                Runnable publishTask = () -> {
                    Buffer buffer = Buffer.buffer("发布数据" + PUBLISH_COUNT.incrementAndGet());
                    client.publish("/hello",buffer,
                            MqttQoS.EXACTLY_ONCE, true, true,
                            asyncResult1 -> {
                                if (asyncResult1.succeeded()) {
                                    logger.info("publish {}", asyncResult1);
                                }
                            }
                    );
                };
                Executors.newScheduledThreadPool(1)
                        .scheduleAtFixedRate(publishTask, 0, 15, TimeUnit.MILLISECONDS);
            });
        }
    };
    Vertx.vertx().deployVerticle(verticle);
}
 

@Override
public void run() {
	Statistics.INSTANCE.data().entrySet().stream().forEach(e -> {
		if (TopicSubscriber.NEXUS.clientIdsOf(e.getKey()).size() <= 0) { return; }

		byte[] msg = e.getValue().value().getBytes(CharsetUtil.UTF_8);

		Message message = new Message(-1, e.getKey(), ClusterDataFactory.INSTANCE.currentId(), msg,
				MqttQoS.AT_MOST_ONCE, false);

		Topic.NEXUS.prepare(message).publish(message);
	});
}
 

@Test
public void testSubscribeNotExist() {
    try {
        MessageListener listener = new MessageListener();
        IMqttToken token = this.mqttClient.subscribeWithResponse("not_exist", listener);
        token.waitForCompletion();

        Assert.assertNotEquals(token.getGrantedQos()[0], MqttQoS.AT_LEAST_ONCE.value());
    } catch (MqttException e) {
        log.error("exception", e);
        Assert.fail();
    }
}
 

public OutboundMessageStatus(String messageKey, String clientId, int messageId, String topicName, Status status,
		MqttQoS qos) {
	super(messageKey, clientId, messageId, topicName);

	this.status = status;
	this.qos = qos;
}
 

/**
 * @param clientID
 *            the clientID
 * @param username
 *            the username
 * @param msg
 *            the subscribe message to verify
 * @return the list of verified topics for the given subscribe message.
 */
List<MqttTopicSubscription> verifyTopicsReadAccess(String clientID, String username, MqttSubscribeMessage msg) {
    List<MqttTopicSubscription> ackTopics = new ArrayList<>();

    final int messageId = messageId(msg);
    for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
        Topic topic = new Topic(req.topicName());
        if (!policy.canRead(topic, username, clientID)) {
            // send SUBACK with 0x80, the user hasn't credentials to read the topic
            LOG.warn("Client does not have read permissions on the topic CId={}, username: {}, messageId: {}, " +
                     "topic: {}", clientID, username, messageId, topic);
            ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
        } else {
            MqttQoS qos;
            if (topic.isValid()) {
                LOG.debug("Client will be subscribed to the topic CId={}, username: {}, messageId: {}, topic: {}",
                          clientID, username, messageId, topic);
                qos = req.qualityOfService();
            } else {
                LOG.warn("Topic filter is not valid CId={}, username: {}, messageId: {}, topic: {}", clientID,
                         username, messageId, topic);
                qos = FAILURE;
            }
            ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
        }
    }
    return ackTopics;
}
 

/**
 * Verifies that the adapter does not forward a message published by a device if the topic is empty and closes the
 * connection to the device.
 */
@Test
public void testUploadTelemetryMessageFailsForEmptyTopic() {

    // GIVEN an adapter
    config.setAuthenticationRequired(false);
    final MqttServer server = getMqttServer(false);
    final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server);
    forceClientMocksToConnected();
    final DownstreamSender sender = givenAQoS0TelemetrySender();

    final MqttEndpoint endpoint = mockEndpoint();
    when(endpoint.isConnected()).thenReturn(Boolean.TRUE);
    adapter.handleEndpointConnection(endpoint);
    @SuppressWarnings("unchecked")
    final ArgumentCaptor<Handler<MqttPublishMessage>> messageHandler = ArgumentCaptor.forClass(Handler.class);
    verify(endpoint).publishHandler(messageHandler.capture());

    // WHEN a device publishes a message that has no topic
    final MqttPublishMessage msg = mock(MqttPublishMessage.class);
    when(msg.topicName()).thenReturn(null);
    when(msg.qosLevel()).thenReturn(MqttQoS.AT_MOST_ONCE);

    messageHandler.getValue().handle(msg);

    // THEN the device gets disconnected
    verify(endpoint).close();
    // and the message is not forwarded downstream
    verify(sender, never()).send(any(Message.class), any());
    // and the message has not been reported as processed
    verify(metrics, never()).reportTelemetry(
            any(MetricsTags.EndpointType.class),
            anyString(),
            any(),
            eq(MetricsTags.ProcessingOutcome.FORWARDED),
            any(MetricsTags.QoS.class),
            anyInt(),
            any());
}
 

public static MqttUnsubAckMessage unsuback(int messageId) {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false,
			2);
	MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);

	return new MqttUnsubAckMessage(fixedHeader, variableHeader);
}
 

public void sendSubscribMessage(BorkerMessage bMsgInfo, List<SubscribeTopicInfo> topicClientList) {
	topicClientList.forEach(topicClient -> {
		String clientId = topicClient.getClientId();
	
		// 订阅者收到MQTT消息的QoS级别, 最终取决于发布消息的QoS和主题订阅的QoS
		MqttQoS respQoS = bMsgInfo.getIQosLevel() > topicClient.getMqttQoS()
				? MqttQoS.valueOf(topicClient.getMqttQoS())
				: MqttQoS.valueOf(bMsgInfo.getIQosLevel());

	    sendSubscribMessageForPublic(clientId, respQoS, bMsgInfo);	
	});
}
 

public static void writeExtension(MqttQoS qos, BrokerMessage brokerMessage) {
    byte[] extension = new byte[] {
            (byte) ((qos.value() >> 24) & 0xFF),
            (byte) ((qos.value() >> 16) & 0xFF),
            (byte) ((qos.value() >> 8) & 0xFF),
            (byte) (qos.value() & 0xFF)
    };
    brokerMessage.setExtension(extension);
}
 

/**
 * Verifies that the adapter rejects messages published to topics containing an endpoint
 * other than <em>telemetry</em> or <em>event</em>.
 *
 * @param ctx The helper to use for running tests on vert.x.
 */
@Test
public void testMapTopicFailsForUnknownEndpoint(final VertxTestContext ctx) {

    givenAnAdapter();

    // WHEN a device publishes a message to a topic with an unknown endpoint
    final MqttPublishMessage message = newMessage(MqttQoS.AT_MOST_ONCE, "unknown");
    adapter.mapTopic(newContext(message, null)).onComplete(ctx.failing(t -> {
        // THEN the message cannot be mapped to a topic
        assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_NOT_FOUND);
        ctx.completeNow();
    }));
}
 

private void clientSendThreePublishMessages(MqttQoS mqttQoS, TestContext context) {
  Async async = context.async(3);
  MqttClient client = MqttClient.create(vertx);

  Queue<Integer> expectOrder = new LinkedList<>();
  // order we expect to receive acknowledgment for published message
  expectOrder.add(2);
  expectOrder.add(1);
  expectOrder.add(3);

  client.publishCompletionHandler(h -> {
    context.assertEquals(h.intValue(), expectOrder.poll());
    log.info("[CLIENT] Publish completed for message with id: " + h);
    async.countDown();
  });

  client.connect(MqttClientOptions.DEFAULT_PORT, MqttClientOptions.DEFAULT_HOST, c -> {

    // publish QoS = 1 message three times
    for (int i = 0; i < 3; i++)
      client.publish(MQTT_TOPIC,
        Buffer.buffer(MQTT_MESSAGE.getBytes()),
        mqttQoS,
        false,
        false, h -> log.info("[CLIENT] publishing message id = " + h.result()));
  });

  async.await();
  client.disconnect();
}
 

@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);
}
 

/**
 * Example for sending publish message
 *
 * @param client
 */
public void example3(MqttClient client) {
  client.publish("temperature",
    Buffer.buffer("hello"),
    MqttQoS.AT_LEAST_ONCE,
    false,
    false);
}
 

/**
 * Verifies that the adapter forwards commands with Qos 0 and response hence and forth between an application and a
 * device.
 *
 * @param endpointConfig The endpoints to use for sending/receiving commands.
 * @param ctx The vert.x test context.
 * @throws InterruptedException if not all commands and responses are exchanged in time.
 */
@ParameterizedTest(name = IntegrationTestSupport.PARAMETERIZED_TEST_NAME_PATTERN)
@MethodSource("allCombinations")
public void testSendCommandSucceedsWithQos0(
        final MqttCommandEndpointConfiguration endpointConfig,
        final VertxTestContext ctx) throws InterruptedException {

    testSendCommandSucceeds(ctx, endpointConfig, MqttQoS.AT_MOST_ONCE);
}
 

private CompletionStage<?> send(AtomicReference<MqttClient> reference, Message<?> msg) {
    MqttClient client = reference.get();
    String actualTopicToBeUsed = this.topic;
    MqttQoS actualQoS = MqttQoS.valueOf(this.qos);
    boolean isRetain = false;

    if (msg instanceof SendingMqttMessage) {
        MqttMessage<?> mm = ((SendingMqttMessage<?>) msg);
        actualTopicToBeUsed = mm.getTopic() == null ? topic : mm.getTopic();
        actualQoS = mm.getQosLevel() == null ? actualQoS : mm.getQosLevel();
        isRetain = mm.isRetain();
    }

    if (actualTopicToBeUsed == null) {
        log.ignoringNoTopicSet();
        return CompletableFuture.completedFuture(msg);
    }

    return client.publish(actualTopicToBeUsed, convert(msg.getPayload()), actualQoS, false, isRetain)
            .onItemOrFailure().produceUni((s, f) -> {
                if (f != null) {
                    return Uni.createFrom().completionStage(msg.nack(f).thenApply(x -> msg));
                } else {
                    return Uni.createFrom().completionStage(msg.ack().thenApply(x -> msg));
                }
            })
            .subscribeAsCompletionStage();
}
 

public MqttQoS getQoSForTopic(String topic) {
    List<Integer> qosList = mqttQoSMap.entrySet()
            .stream()
            .filter(entry -> entry.getKey().matches(topic))
            .map(Map.Entry::getValue)
            .collect(Collectors.toList());
    if (!qosList.isEmpty()) {
        return MqttQoS.valueOf(qosList.get(0));
    } else {
        return MqttQoS.AT_LEAST_ONCE;
    }
}
 

private void unsubscribe(TestContext context, MqttQoS qos) {

    this.messageId = 0;

    Async async = context.async();
    MqttClient client = MqttClient.create(Vertx.vertx());

    client.unsubscribeCompletionHandler(unsubackid -> {
      assertTrue(unsubackid == messageId);
      log.info("unsubscribing complete for message id = " + unsubackid);
      client.disconnect();
      async.countDown();
    });

    client.subscribeCompletionHandler(suback -> {
      assertTrue(suback.messageId() == messageId);
      assertTrue(suback.grantedQoSLevels().contains(qos.value()));
      log.info("subscribing complete for message id = " + suback.messageId() + " with QoS " + suback.grantedQoSLevels());

      client.unsubscribe(MQTT_TOPIC, ar2 -> {
        assertTrue(ar2.succeeded());
        messageId = ar2.result();
        log.info("unsubscribing on [" + MQTT_TOPIC + "] message id = " + messageId);
      });
    });

    client.connect(TestUtil.BROKER_PORT, TestUtil.BROKER_ADDRESS, ar -> {
      assertTrue(ar.succeeded());

      client.subscribe(MQTT_TOPIC, qos.value(), ar1 -> {
        assertTrue(ar1.succeeded());
        messageId = ar1.result();
        log.info("subscribing on [" + MQTT_TOPIC + "] with QoS [" + qos.value() + "] message id = " + messageId);
      });
    });

    async.await();
  }
 

public static MqttSubAckMessage suback(int messageId, List<Integer> grantedQoSLevels) {
	MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false,
			2 + grantedQoSLevels.size());
	MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
	MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);

	return new MqttSubAckMessage(fixedHeader, variableHeader, payload);
}
 

private MqttPublishMessage createMqttPublishMsg(DeviceSessionCtx ctx, String topic, JsonElement json) {
  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false,
      0);
  MqttPublishVariableHeader header = new MqttPublishVariableHeader(topic, ctx.nextMsgId());
  ByteBuf payload = ALLOCATOR.buffer();
  payload.writeBytes(GSON.toJson(json).getBytes(UTF8));
  return new MqttPublishMessage(mqttFixedHeader, header, payload);
}