import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
private List<Client> getConnectedClients(boolean includeSubscriptions) {
    return clientService.getLocalConnectedClients().stream()
            .map(clientId -> {
                ClientData clientData = clientService.getClientData(clientId);

                String ip = null;
                if (clientData.getInetAddress().isPresent()) {
                    ip = clientData.getInetAddress().get().getHostAddress();
                }

                List<String> subscriptions = includeSubscriptions ? getSubscriptions(clientId) : null;

                return new Client(clientId, true, ip, subscriptions);
            })
            .collect(Collectors.toList());
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onConnect(final CONNECT connect, final ClientData clientData) throws RefusedConnectionException {

    //Throw out clients which didn't provide a client certificate
    if (!clientData.getCertificate().isPresent()) {
        log.error("Client {} didn't provide a X509 client certificate. Disconnecting client", clientData.getClientId());
        throw new RefusedConnectionException(ReturnCode.REFUSED_NOT_AUTHORIZED);
    }
    log.info("Client {} connected with X509 client certificate", clientData.getClientId());

    final Certificate certificate = clientData.getCertificate().get().certificate();

    //We need to downcast to the X509 certificate
    if (certificate instanceof X509Certificate) {

        final X509Certificate x509Certificate = (X509Certificate) certificate;
        log.info("X509 client certificate provided by client {} has the serial number {}",
                clientData.getClientId(), x509Certificate.getSerialNumber());
    }

}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This method is called from the HiveMQ, when a new MQTT {@link com.hivemq.spi.message.PUBLISH} message arrives
 * at the broker. In this acme the method is just logging each message to the console.
 *
 * @param publish    The publish message send by the client.
 * @param clientData Useful information about the clients authentication state and credentials.
 * @throws com.hivemq.spi.callback.exception.OnPublishReceivedException When the exception is thrown, the publish is not
 *                                                                      accepted and will NOT be delivered to the subscribing clients.
 */
@Override
public void onPublishReceived(final PUBLISH publish, final ClientData clientData) throws OnPublishReceivedException {
    if (publish.getTopic().equals("fetch/all/clients")) {

        String clientID = clientData.getClientId();
        String topic = publish.getTopic();
        String message = new String(publish.getPayload(), Charsets.UTF_8);

        logger.info("Client " + clientID + " sent a message to topic " + topic + ": " + message);

        // Get all clients
        String allClients = getAllClients();
        publish.setPayload(allClients.getBytes(Charsets.UTF_8));

        // This redirects the message with the help of the PublishService
        redirectPublish(allClientsTopic, publish);

        logger.info("Ignoring message and sending list of all clients to topic {}", allClientsTopic);
        throw new OnPublishReceivedException("This message should not be published", false);
    }

}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onPublishReceived(PUBLISH message, ClientData cd) throws OnPublishReceivedException {
	if (!noPuts) {
		Integer mi = message.getMessageId();
		boolean delivered = false;
		//for both qos 1 and 2
		if (message.isDuplicateDelivery()) {
			//check against registry
			delivered = msgRegistry.get(mi) != null;
		}
		if (!delivered) {
			Future<RecordMetadata> f = producer.send(new ProducerRecord<String, byte[]>(sinkTopic, message.getTopic().substring(kafkaRecordKeyRange[0], kafkaRecordKeyRange[1]), message.getPayload()));
			if (!asyncPuts) {
				try {
					RecordMetadata rm = f.get();
					msgRegistry.put(mi, dummy);
				}
				catch (ExecutionException | InterruptedException ee) {
					//check offset before putting again?
					log.error("Failed when putting to kafka", ee);
				}
			}
			else {
				msgRegistry.put(mi, dummy);
			}
		}
	}
	if (logFrequency != -1) {
		int np = puts.incrementAndGet();
		if (np % logFrequency == 0) {
			log.info("Put {} in {} ms. Size of registry {}", puts, System.currentTimeMillis() - start, msgRegistry.size());
			start = System.currentTimeMillis();
			puts.set(0);
		}
	}
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This is an example authorization method, which shows how to allow a client only
 * to access topics, which have its client id up front.
 * This can be easily changed to implement your authorization logic of choice: read from a database
 * or call your authorization webservice.
 *
 * @param clientData All information from the MQTT client requesting authorization.
 * @return a list of valid topic permissions
 */
@Override
@Cached(timeToLive = 5, timeUnit = TimeUnit.MINUTES)
public List<MqttTopicPermission> getPermissionsForClient(ClientData clientData) {
    ArrayList<MqttTopicPermission> mqttTopicPermissions = new ArrayList<>();
    mqttTopicPermissions.add(
            new MqttTopicPermission(
                    clientData.getClientId() + "/#",            // Topic
                    MqttTopicPermission.TYPE.ALLOW,        // Type
                    MqttTopicPermission.QOS.ALL,        // QoS
                    MqttTopicPermission.ACTIVITY.ALL)); // Publish, Subscribe, All

    return mqttTopicPermissions;
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
@Cached(timeToLive = 10, timeUnit = TimeUnit.SECONDS)
public List<MqttTopicPermission> getPermissionsForClient(ClientData clientData) {

    final List<MqttTopicPermission> permissions = new ArrayList<>();
    permissions.add(new MqttTopicPermission("debug/#", ALLOW));
    permissions.add(new MqttTopicPermission("system/#", DENY));

    return permissions;
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
@Cached(timeToLive = 10, timeUnit = TimeUnit.SECONDS)
public List<MqttTopicPermission> getPermissionsForClient(ClientData clientData) {

    final List<MqttTopicPermission> permissions = new ArrayList<>();
    permissions.add(new MqttTopicPermission("client/" + clientData.getClientId(), TYPE.ALLOW, QOS.ONE, ACTIVITY.SUBSCRIBE));
    permissions.add(new MqttTopicPermission("client/#", TYPE.ALLOW, QOS.ALL, ACTIVITY.PUBLISH, RETAIN.NOT_RETAINED));

    return permissions;
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
@Cached(timeToLive = 10, timeUnit = TimeUnit.SECONDS)
public List<MqttTopicPermission> getPermissionsForClient(ClientData clientData) {

    final List<MqttTopicPermission> permissions = new ArrayList<>();
    permissions.add(new MqttTopicPermission("client/" + clientData.getClientId(), TYPE.ALLOW, QOS.ONE));
    permissions.add(new MqttTopicPermission("client/#", TYPE.DENY, ACTIVITY.SUBSCRIBE));

    return permissions;
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public List<MqttTopicPermission> getPermissionsForClient(final ClientData clientData) {

    final List<MqttTopicPermission> permissions = new ArrayList<>();

    final X509Certificate certificate = (X509Certificate) clientData.getCertificate().get();

    if ("dc-square".equals(certificate.getIssuerDN().getName())) {
        log.info("Issuer was dc-square, client is allowed to use all topics with all permissions");
        permissions.add(new MqttTopicPermission("#", MqttTopicPermission.TYPE.ALLOW));
    }

    return permissions;
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This method is called from the HiveMQ, when a new MQTT {@link PUBLISH} message arrives
 * at the broker. In this acme the method is just logging each message to the console.
 *
 * @param publish    The publish message send by the client.
 * @param clientData Useful information about the clients authentication state and credentials.
 * @throws OnPublishReceivedException When the exception is thrown, the publish is not
 *                                    accepted and will NOT be delivered to the subscribing clients.
 */
@Override
public void onPublishReceived(PUBLISH publish, ClientData clientData) throws OnPublishReceivedException {
    String clientID = clientData.getClientId();
    String topic = publish.getTopic();
    String message = new String(publish.getPayload(), Charsets.UTF_8);

    logger.info("Client " + clientID + " sent a message to topic " + topic + ": " + message);

}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onPublishReceived(final PUBLISH publish, final ClientData clientData) throws OnPublishReceivedException {
    final String clientID = clientData.getClientId();
    final String topic = publish.getTopic();
    final String payload = new String(publish.getPayload(), Charsets.UTF_8);

    log.info("Client {} sent a message to topic \"{}\": \"{}\" (QoS: {}, retained: {})",
            clientID, topic, payload, publish.getQoS().getQosNumber(), publish.isRetain());
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onConnect(final CONNECT connect, final ClientData clientData) throws RefusedConnectionException {
    log.info("Client {} connected", clientData.getClientId());

    if (connect.isWill()) {
        log.info("Client {} set Last Will and Testament. Topic: \"{}\", Payload: \"{}\", QoS: {}, retain: {}",
                clientData.getClientId(), connect.getWillTopic(), connect.getWillMessage(), connect.getWillQos().getQosNumber(), connect.isWillRetain());
    }
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onDisconnect(final ClientData clientData, final boolean abruptDisconnect) {
    if (!abruptDisconnect) {
        log.info("Client {} sent DISCONNECT message", clientData.getClientId());
    } else {
        log.info("Client {} disconnected", clientData.getClientId());
    }
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onPubrelReceived(PUBREL pubrel, ClientData clientData) {
	//QoS 2 workflow complete. cleanup so that same message id can be reused as per MQTT spec
	msgRegistry.remove(pubrel.getMessageId());
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onPubackSend(PUBACK puback, ClientData clientData) {
	//QoS 1 workflow complete. cleanup so that same message id can be reused as per MQTT spec
	msgRegistry.remove(puback.getMessageId());
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onSubscribe(final SUBSCRIBE message, final ClientData clientData) throws InvalidSubscriptionException {
    log.info("Subscribe from client {} received: {}", clientData.getClientId(), buildSubscriptionString(message.getTopics()));
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
@Override
public void onUnsubscribe(UNSUBSCRIBE message, ClientData clientData) {
    log.info("Unsubscribe from client {} received: {}", clientData.getClientId(), buildSubscriptionString(message.getTopics()));

}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This is the callback method, which is called by the HiveMQ core, if a client has sent,
 * a {@link com.hivemq.spi.message.CONNECT} Message and was successfully authenticated. In this acme there is only
 * a logging statement, normally the behavior would be implemented in here.
 *
 * @param connect    The {@link com.hivemq.spi.message.CONNECT} message from the client.
 * @param clientData Useful information about the clients authentication state and credentials.
 * @throws com.hivemq.spi.callback.exception.RefusedConnectionException This exception should be thrown, if the client is
 *                                                                      not allowed to connect.
 */
@Override
public void onConnect(CONNECT connect, ClientData clientData) throws RefusedConnectionException {
    final String clientId = clientData.getClientId();

    log.info("Client {} is connecting", clientId);

    // Adding a subscription without automatically for the client
    addClientToTopic(clientId, "devices/" + clientId + "/sensor");
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This is the callback method, which is called by the HiveMQ core, if a client has sent,
 * a {@link CONNECT} Message and was successfully authenticated. In this acme there is only
 * a logging statement, normally the behavior would be implemented in here.
 *
 * @param connect    The {@link CONNECT} message from the client.
 * @param clientData Useful information about the clients authentication state and credentials.
 * @throws RefusedConnectionException This exception should be thrown, if the client is
 *                                    not allowed to connect.
 */
@Override
public void onConnect(CONNECT connect, ClientData clientData) throws RefusedConnectionException {
    log.info("Client {} is connecting", clientData.getClientId());
}
 

import com.hivemq.spi.security.ClientData; //导入依赖的package包/类
/**
 * This method is called from the HiveMQ on a client disconnect.
 *
 * @param clientData       Useful information about the clients authentication state and credentials.
 * @param abruptDisconnect When true the connection of the client broke down without a
 *                         {@link com.hivemq.spi.message.DISCONNECT} message and if false then the client
 *                         disconnected properly with a {@link com.hivemq.spi.message.DISCONNECT} message.
 */
@Override
public void onDisconnect(ClientData clientData, boolean abruptDisconnect) {
    log.info("Client {} is disconnected", clientData.getClientId());
}