thingsboard-gateway从thingsboard获取拓展配置源码分析

悠悠 2023-02-15 12:46 2阅读 0赞

本文基于thingsboard-gateway的早期java版本，现在thingsboard-gateway已改为python语言，下图为thingsboard-gateway配置示意图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3_size_16_color_FFFFFF_t_70][]

一切缘起remoteConfiguration这个配置，当remoteConfiguration设置为true时网关就会向thingsboard（下面简称tb）发送获取设备共享属性configuration的请求，如果tb端针对该网关有configuration这个共享属性，那么就会将这个配置下发到gateway,gateway就会使用下发的配置创建对应的服务（比如mqtt、opc、modbus等），如果remoteConfiguration配置为false，那么gateway就会使用extensions部分配置的扩展进行初始化，下面是DefaultTenantManagerService初始化相关代码：

@PostConstruct
        public void init() {
            gateways = new HashMap<>();
            httpServices = new ArrayList<>();
            for (TbTenantConfiguration configuration : configuration.getTenants()) {
                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(c));
                        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 TenantServiceRegistry();
                        // 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<MqttConnectResult> connectResult = (Promise<MqttConnectResult>) 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());
                        }
                    }
                });
                connectResult.get(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
    
    
                tbClient.on(DEVICE_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
                tbClient.on(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
    
                tbClient.on(GATEWAY_RESPONSES_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
                tbClient.on(GATEWAY_ATTRIBUTES_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
                tbClient.on(GATEWAY_RPC_TOPIC, this).await(connection.getConnectionTimeout(), TimeUnit.MILLISECONDS);
    
                // 向tb索取该网关对应的configuration配置，注意一定是共享属性
                byte[] msgData = toBytes(newNode().put("sharedKeys", "configuration"));
                persistMessage(DEVICE_GET_ATTRIBUTES_REQUEST_TOPIC, msgIdSeq.incrementAndGet(), msgData, null,
                        null,
                        error -> log.warn("Error getiing attributes", error));
                return tbClient;
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
                Thread.currentThread().interrupt();
                return null;
            } catch (ExecutionException e) {
                log.error(e.getMessage(), e);
                throw new RuntimeException(e);
            } catch (TimeoutException e) {
                String message = "Unable to connect to ThingsBoard. Connection timed out after [" + connection.getConnectionTimeout() + "] milliseconds";
                log.error(message, e);
                throw new RuntimeException(message);
            }
        }

共享属性查看页面

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3_size_16_color_FFFFFF_t_70 1][]

PS：

java版本thingsboard-gateway可以通过[https://gitee.com/johnHust/thingsboard-gateway][https_gitee.com_johnHust_thingsboard-gateway]下载

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200605152331514.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200605154245205.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pvaG4xMzM3,size_16,color_FFFFFF,t_70
[https_gitee.com_johnHust_thingsboard-gateway]: https://gitee.com/johnHust/thingsboard-gateway