thingsboard-gateway从thingsboard获取拓展配置源码分析
isRemoteConfiguration = configuration.getRemoteConfiguration();
if (isRemoteConfiguration) {
String label = configuration.getLabel();
log.info(“[{}] Initializing gateway”, configuration.getLabel());
TenantServiceRegistry tenantServiceRegistry = new TenantServiceRegistry();
GatewayService service = null;
try {
// isRemoteConfiguration为true时，会响应服务端返回的配置信息
service = getGatewayService(configuration, c -> tenantServiceRegistry.updateExtensionConfiguration©);
tenantServiceRegistry.setService(service);
gateways.put(label, tenantServiceRegistry);
} catch (Exception e) {
log.info("[{}] Failed to initialize the service ", label, e);
try {
if (service != null) {
service.destroy();
}
} catch (Exception exc) {
log.info("[{}] Failed to stop the service ", label, exc);
}
}
} else {
String label = configuration.getLabel();
log.info(“[{}] Initializing gateway”, configuration.getLabel());
GatewayService service = null;
try {
TenantServiceRegistry tenantServiceRegistry = new TenantServiceRe 《一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》无偿开源 威信搜索公众号【编程进阶路】 gistry();
// isRemoteConfiguration为false时，会忽略服务端返回的配置信息
service = getGatewayService(configuration, c -> {});
tenantServiceRegistry.setService(service);
for (TbExtensionConfiguration extensionConfiguration : configuration.getExtensions()) {
log.info(“[{}] Initializing extension: [{}]”, configuration.getLabel(), extensionConfiguration.getType());
ExtensionService extension = tenantServiceRegistry.createExtensionServiceByType(service, extensionConfiguration.getType());
extension.init(extensionConfiguration, isRemoteConfiguration);
if (extensionConfiguration.getType().equals(“HTTP”)) {
httpServices.add((HttpService) extension);
}
}
gateways.put(label, (TenantServiceRegistry) tenantServiceRegistry);
} catch (Exception e) {
log.info("[{}] Failed to initialize the service ", label, e);
try {
if (service != null) {
service.destroy();
}
} catch (Exception exc) {
log.info("[{}] Failed to stop the service ", label, exc);
}
}
}
}
}
gateway通过mqtt连接上tb后，会向tb发送获取configuration配置的rpc请求，具体在MqttGatewayService的initMqttClient方法中：
private MqttClient initMqttClient() {
try {
MqttClientConfig mqttClientConfig = getMqttClientConfig();
mqttClientConfig.setUsername(connection.getSecurity().getAccessToken());
tbClient = MqttClient.create(mqttClientConfig, this);
tbClient.setCallback(this);
tbClient.setEventLoop(nioEventLoopGroup);
Promise connectResult = (Promise) tbClient.connect(connection.getHost(), connection.getPort());
connectResult.addListener(future -> {
if (future.isSuccess()) {
MqttConnectResult result = (MqttConnectResult) future.getNow();
log.debug(“Gateway connect result code: [{}]”, result.getReturnCode());
} else {
log.error(“Unable to connect to mqtt server!”);
if (future.cause() != null) {
log.error(future.cause().getMessage(), future.cause());