opensource/iot-dc3
3
0
Fork 0
mirror of https://gitee.com/pnoker/iot-dc3.git synced 2026-11-03 11:42:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(edge-gateway): add a new type driver: edge-gateway

Browse Source
This commit is contained in:
pnoker
2021-08-02 14:55:50 +08:00
parent 2ee8a0f998
commit c0db070266
25 changed files with 862 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dc3-api/dc3-api-manager/src/main/java/com/dc3/api/center/manager/feign/DriverClient.java
+4 -3
View File
@@ -81,14 +81,15 @@ public interface DriverClient {
R<Driver> selectByServiceName(@NotNull @PathVariable(value = "serviceName") String serviceName);
/**
* 根据 HOST & PORT 查询 Driver
* 根据 TYPE & HOST & PORT 查询 Driver
*
* @param type Driver type
* @param host Driver Host
* @param port Driver Port
* @return Driver
*/
@GetMapping("/host/{host}/port/{port}")
R<Driver> selectByHostPort(@NotNull @PathVariable(value = "host") String host, @NotNull @PathVariable(value = "port") Integer port, @RequestHeader(value = Common.Service.DC3_AUTH_TENANT_ID, defaultValue = "-1") Long tenantId);
@GetMapping("/type/{type}/host/{host}/port/{port}")
R<Driver> selectByHostPort(@NotNull @PathVariable(value = "type") String type, @NotNull @PathVariable(value = "host") String host, @NotNull @PathVariable(value = "port") Integer port, @RequestHeader(value = Common.Service.DC3_AUTH_TENANT_ID, defaultValue = "-1") Long tenantId);
/**
* 分页查询 Driver
dc3-api/dc3-api-manager/src/main/java/com/dc3/api/center/manager/hystrix/DriverClientHystrix.java
+1 -1
View File
@@ -64,7 +64,7 @@ public class DriverClientHystrix implements FallbackFactory<DriverClient> {
}
@Override
public R<Driver> selectByHostPort(String host, Integer port, Long tenantId) {
public R<Driver> selectByHostPort(String type, String host, Integer port, Long tenantId) {
return R.fail(message);
}
dc3-center/dc3-auth/src/main/resources/application-dev.yml
+1 -1
View File
@@ -1,3 +1,3 @@
eureka:
instance:
ip-address: localhost
ip-address: 10.221.91.80
dc3-center/dc3-manager/src/main/java/com/dc3/center/manager/api/DriverApi.java
+2 -2
View File
@@ -94,9 +94,9 @@ public class DriverApi implements DriverClient {
}
@Override
public R<Driver> selectByHostPort(String host, Integer port, Long tenantId) {
public R<Driver> selectByHostPort(String type, String host, Integer port, Long tenantId) {
try {
Driver select = driverService.selectByHostPort(host, port, tenantId);
Driver select = driverService.selectByHostPort(type, host, port, tenantId);
return R.ok(select);
} catch (Exception e) {
return R.fail(e.getMessage());
dc3-center/dc3-manager/src/main/java/com/dc3/center/manager/service/DriverService.java
+2 -1
View File
@@ -38,11 +38,12 @@ public interface DriverService extends Service<Driver, DriverDto> {
/**
* 根据 驱动 Host & Port 查询 驱动
*
* @param type Driver Type
* @param host Driver Service Host
* @param port Driver Service Port
* @return Driver
*/
Driver selectByHostPort(String host, Integer port, Long tenantId);
Driver selectByHostPort(String type, String host, Integer port, Long tenantId);
/**
* 根据 驱动Id 查询 驱动
dc3-center/dc3-manager/src/main/java/com/dc3/center/manager/service/impl/DriverServiceImpl.java
+5 -4
View File
@@ -62,7 +62,7 @@ public class DriverServiceImpl implements DriverService {
put = {
@CachePut(value = Common.Cache.DRIVER + Common.Cache.ID, key = "#driver.id", condition = "#result!=null"),
@CachePut(value = Common.Cache.DRIVER + Common.Cache.SERVICE_NAME, key = "#driver.serviceName", condition = "#result!=null"),
@CachePut(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#driver.host+'.'+#driver.port+'.'+#driver.tenantId", condition = "#result!=null")
@CachePut(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#driver.type+'.'+#driver.host+'.'+#driver.port+'.'+#driver.tenantId", condition = "#result!=null")
},
evict = {
@CacheEvict(value = Common.Cache.DRIVER + Common.Cache.DEVICE_ID, allEntries = true, condition = "#result!=null"),
@@ -105,7 +105,7 @@ public class DriverServiceImpl implements DriverService {
put = {
@CachePut(value = Common.Cache.DRIVER + Common.Cache.ID, key = "#driver.id", condition = "#result!=null"),
@CachePut(value = Common.Cache.DRIVER + Common.Cache.SERVICE_NAME, key = "#driver.serviceName", condition = "#result!=null"),
@CachePut(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#driver.host+'.'+#driver.port+'.'+#driver.tenantId", condition = "#result!=null")
@CachePut(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#driver.type+'.'+#driver.host+'.'+#driver.port+'.'+#driver.tenantId", condition = "#result!=null")
},
evict = {
@CacheEvict(value = Common.Cache.DRIVER + Common.Cache.DEVICE_ID, allEntries = true, condition = "#result!=null"),
@@ -155,9 +155,10 @@ public class DriverServiceImpl implements DriverService {
}
@Override
@Cacheable(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#host+'.'+#port+'.'+#tenantId", unless = "#result==null")
public Driver selectByHostPort(String host, Integer port, Long tenantId) {
@Cacheable(value = Common.Cache.DRIVER + Common.Cache.HOST_PORT, key = "#type+'.'+#host+'.'+#port+'.'+#tenantId", unless = "#result==null")
public Driver selectByHostPort(String type, String host, Integer port, Long tenantId) {
LambdaQueryWrapper<Driver> queryWrapper = Wrappers.<Driver>query().lambda();
queryWrapper.eq(Driver::getType, type);
queryWrapper.eq(Driver::getHost, host);
queryWrapper.eq(Driver::getPort, port);
queryWrapper.eq(Driver::getTenantId, tenantId);
dc3-common/dc3-base/src/main/java/com/dc3/common/model/Driver.java
+2 -1
View File
@@ -64,11 +64,12 @@ public class Driver extends Description {
@TableField(exist = false)
private String status;
public Driver(String name, String serviceName, String host, Integer port) {
public Driver(String name, String serviceName, String host, Integer port, String type) {
super();
this.name = name;
this.serviceName = serviceName;
this.host = host;
this.port = port;
this.type = type;
}
}
dc3-common/dc3-sdk/src/main/java/com/dc3/common/sdk/service/impl/DriverMetadataServiceImpl.java
+1 -1
View File
@@ -64,7 +64,7 @@ public class DriverMetadataServiceImpl implements DriverMetadataService {
throw new ServiceException("The driver port is invalid, port range is 8600-8799");
}
Driver driver = new Driver(driverProperty.getName(), this.serviceName, localHost, this.port);
Driver driver = new Driver(driverProperty.getName(), this.serviceName, localHost, this.port, driverProperty.getType());
driver.setDescription(driverProperty.getDescription());
log.info("The driver {}/{} is initializing", driver.getServiceName(), driver.getName());
dc3-driver/dc3-driver-edge-gateway/Dockerfile
+42
View File
@@ -0,0 +1,42 @@
#
# Copyright 2016-2021 Pnoker. All Rights Reserved.
#
# 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.
#
FROM registry.cn-beijing.aliyuncs.com/dc3/alpine-java:1.8.251
MAINTAINER pnoker <pnokers.icloud.com>
ENV JAVA_OPS -server -Xms128m -Xmx1024m -XX:CompressedClassSpaceSize=128m -XX:MetaspaceSize=200m -XX:MaxMetaspaceSize=200m
ENV GC_LOG -XX:+PrintGCDetails \
-XX:+PrintGCDateStamps \
-XX:+PrintTenuringDistribution \
-XX:+PrintHeapAtGC \
-XX:+PrintReferenceGC \
-XX:+PrintGCApplicationStoppedTime \
-XX:+UseGCLogFileRotation \
-XX:NumberOfGCLogFiles=10 \
-XX:GCLogFileSize=10M \
-Xloggc:dc3/logs/driver/edge-gateway/gc/dc3-driver-edge-gateway-gc.log
RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
WORKDIR /dc3-driver/dc3-driver-edge-gateway
ADD ./target/dc3-driver-edge-gateway.jar ./
EXPOSE 8701
VOLUME /dc3-driver/dc3-driver-edge-gateway/dc3/logs
CMD mkdir -p /dc3-driver/dc3-driver-edge-gateway/dc3/logs/driver/edge-gateway/gc \
&& java ${JAVA_OPS} ${GC_LOG} -Djava.security.egd=file:/dev/./urandom -jar dc3-driver-edge-gateway.jar
dc3-driver/dc3-driver-edge-gateway/pom.xml
+48
View File
@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016-2021 Pnoker. All Rights Reserved.
~ 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.dc3</groupId>
<artifactId>dc3-driver</artifactId>
<version>1.3.2.SR</version>
</parent>
<artifactId>dc3-driver-edge-gateway</artifactId>
<packaging>jar</packaging>
<description>IOT DC3 平台 边缘网关 驱动，该驱动是一个网关类型的驱动。</description>
<properties>
<bouncycastle.version>1.46</bouncycastle.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-jdk16</artifactId>
<version>${bouncycastle.version}</version>
</dependency>
</dependencies>
</project>
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/EdgeGatewayDriverApplication.java
+31
View File
@@ -0,0 +1,31 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver;
import com.dc3.config.autoconfigure.Dc3Application;
import com.dc3.config.boot.SpringApplication;
import org.springframework.cache.annotation.EnableCaching;
/**
* @author pnoker
*/
@EnableCaching
@Dc3Application
public class EdgeGatewayDriverApplication {
public static void main(String[] args) {
SpringApplication.run(EdgeGatewayDriverApplication.class, args);
}
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/bean/DevicePayLoad.java
+32
View File
@@ -0,0 +1,32 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.bean;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
/**
* @author pnoker
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
public class DevicePayLoad {
private Long deviceId;
private Long pointId;
private String value;
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/bean/MqttProperty.java
+66
View File
@@ -0,0 +1,66 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.bean;
import com.dc3.common.valid.Insert;
import com.dc3.common.valid.Update;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import javax.validation.constraints.Size;
import java.util.List;
/**
* @author pnoker
*/
@Setter
@Getter
@Validated({Insert.class, Update.class})
@ConfigurationProperties(prefix = "driver.mqtt")
public class MqttProperty {
@NotBlank(message = "url can't be empty")
private String url;
@NotNull
private AuthTypeEnum authType = AuthTypeEnum.NONE;
private String username;
private String password;
private String caCrt = "classpath:/certs/ca.crt";
private String clientKeyPass = "dc3-client";
private String clientKey = "classpath:/certs/client.key";
private String clientCrt = "classpath:/certs/client.crt";
@Size(min = 1, message = "at least one qos")
private List<Integer> qos;
@Size(min = 1, message = "at least one topic")
private List<String> topics;
@NotNull(message = "keep alive interval can't be empty")
private Integer keepAlive;
@NotNull(message = "completion timeout can't be empty")
private Integer completionTimeout;
public enum AuthTypeEnum {
NONE, CLIENT_ID, USERNAME, X509
}
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/config/MqttConfig.java
+230
View File
@@ -0,0 +1,230 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.config;
import cn.hutool.core.util.StrUtil;
import com.dc3.common.constant.Common;
import com.dc3.common.exception.ServiceException;
import com.dc3.driver.bean.MqttProperty;
import lombok.extern.slf4j.Slf4j;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.PEMReader;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import javax.annotation.Resource;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.ByteArrayInputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyPair;
import java.security.KeyStore;
import java.security.Security;
import java.security.cert.X509Certificate;
/**
* @author pnoker
*/
@Slf4j
@Configuration
@IntegrationComponentScan
@EnableConfigurationProperties({MqttProperty.class})
public class MqttConfig {
@Value("${driver.mqtt.client.id}")
private String clientId;
@Value("${driver.mqtt.default.topic}")
private String defaultTopic;
@Value("${driver.mqtt.default.qos}")
private Integer defaultQos;
@Resource
private MqttProperty mqttProperty;
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions());
return factory;
}
@Bean
public MqttConnectOptions getMqttConnectOptions() {
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
// username & password
if (mqttProperty.getAuthType().equals(MqttProperty.AuthTypeEnum.USERNAME)) {
mqttConnectOptions.setUserName(mqttProperty.getUsername());
mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
}
// tls x509
if (mqttProperty.getAuthType().equals(MqttProperty.AuthTypeEnum.X509)) {
mqttConnectOptions.setSocketFactory(getSocketFactory(
mqttProperty.getCaCrt(),
mqttProperty.getClientCrt(),
mqttProperty.getClientKey(),
StrUtil.isBlank(mqttProperty.getClientKeyPass()) ? "" : mqttProperty.getClientKeyPass()
));
if (!StrUtil.isBlank(mqttProperty.getUsername()) && !StrUtil.isBlank(mqttProperty.getPassword())) {
mqttConnectOptions.setUserName(mqttProperty.getUsername());
mqttConnectOptions.setPassword(mqttProperty.getPassword().toCharArray());
}
}
mqttConnectOptions.setServerURIs(new String[]{mqttProperty.getUrl()});
mqttConnectOptions.setKeepAliveInterval(mqttProperty.getKeepAlive());
return mqttConnectOptions;
}
@Bean
public MessageChannel defaultMqttInputChannel() {
return new DirectChannel();
}
/**
* 说明：
* ConditionalOnProperty(value = "driver.mqtt.default.receive.enable")
* 根据配置属性driver.mqtt.default.receive.enable选择是否开启 Default Topic 主题的数据接收逻辑
*
* @return
*/
@Bean
@ConditionalOnProperty(value = "driver.mqtt.default.receive.enable")
public MessageProducer defaultInbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
clientId + "_default_inbound_",
mqttClientFactory(),
defaultTopic
);
adapter.setCompletionTimeout(mqttProperty.getCompletionTimeout());
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(defaultQos);
adapter.setOutputChannel(defaultMqttInputChannel());
return adapter;
}
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
@Bean
public MessageChannel mqttOutputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
clientId + "_inbound",
mqttClientFactory(),
mqttProperty.getTopics().toArray(new String[0])
);
adapter.setCompletionTimeout(mqttProperty.getCompletionTimeout());
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(mqttProperty.getQos().stream().mapToInt(Integer::valueOf).toArray());
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "mqttOutputChannel")
public MessageHandler outbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultQos(defaultQos);
messageHandler.setDefaultTopic(defaultTopic);
return messageHandler;
}
private SSLSocketFactory getSocketFactory(final String caCrtFile, final String crtFile, final String keyFile, final String password) {
try {
Security.addProvider(new BouncyCastleProvider());
PEMReader reader = null;
// load CA certificate
if (caCrtFile.startsWith(Common.CLASSPATH)) {
reader = new PEMReader(new InputStreamReader(this.getClass().getResourceAsStream(caCrtFile.replace(Common.CLASSPATH, ""))));
} else {
reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(caCrtFile)))));
}
X509Certificate caCert = (X509Certificate) reader.readObject();
reader.close();
// load client certificate
if (crtFile.startsWith(Common.CLASSPATH)) {
reader = new PEMReader(new InputStreamReader(this.getClass().getResourceAsStream(crtFile.replace(Common.CLASSPATH, ""))));
} else {
reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(crtFile)))));
}
X509Certificate cert = (X509Certificate) reader.readObject();
reader.close();
// load client private key
if (keyFile.startsWith(Common.CLASSPATH)) {
reader = new PEMReader(new InputStreamReader(this.getClass().getResourceAsStream(keyFile.replace(Common.CLASSPATH, ""))), password::toCharArray);
} else {
reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(keyFile)))), password::toCharArray);
}
KeyPair key = (KeyPair) reader.readObject();
reader.close();
// CA certificate is used to authenticate server
KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
caKs.load(null, null);
caKs.setCertificateEntry("cacertfile", caCert);
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(caKs);
// client key and certificates are sent to server so it can authenticate us
KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
ks.load(null, null);
ks.setCertificateEntry("certfile", cert);
ks.setKeyEntry("keyfile", key.getPrivate(), password.toCharArray(), new java.security.cert.Certificate[]{cert});
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, password.toCharArray());
// finally, create SSL socket factory
SSLContext context = SSLContext.getInstance("TLSv1.2");
context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
return context.getSocketFactory();
} catch (Exception e) {
throw new ServiceException(e.getMessage());
}
}
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/service/impl/DriverCustomServiceImpl.java
+84
View File
@@ -0,0 +1,84 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.service.impl;
import com.dc3.common.bean.driver.AttributeInfo;
import com.dc3.common.constant.Common;
import com.dc3.common.model.Device;
import com.dc3.common.model.Point;
import com.dc3.common.sdk.bean.DriverContext;
import com.dc3.common.sdk.service.DriverCustomService;
import com.dc3.common.sdk.service.DriverService;
import com.dc3.driver.service.mqtt.MqttSendHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Map;
import static com.dc3.common.sdk.util.DriverUtils.attribute;
/**
* @author pnoker
*/
@Slf4j
@Service
public class DriverCustomServiceImpl implements DriverCustomService {
@Resource
private DriverContext driverContext;
@Resource
private DriverService driverService;
@Resource
private MqttSendHandler mqttSendHandler;
@Override
public void initial() {
}
@Override
public String read(Map<String, AttributeInfo> driverInfo, Map<String, AttributeInfo> pointInfo, Device device, Point point) throws Exception {
return "nil";
}
@Override
public Boolean write(Map<String, AttributeInfo> driverInfo, Map<String, AttributeInfo> pointInfo, Device device, AttributeInfo values) throws Exception {
String commandTopic = attribute(pointInfo, "commandTopic"), value = values.getValue();
try {
int commandQos = attribute(pointInfo, "commandQos");
mqttSendHandler.sendToMqtt(commandTopic, commandQos, value);
} catch (Exception e) {
mqttSendHandler.sendToMqtt(commandTopic, value);
}
return true;
}
@Override
public void schedule() {
/*
TODO:设备状态
上传设备状态，可自行灵活拓展，不一定非要在schedule()接口中实现，也可以在read中实现设备状态的设置；
你可以通过某种判断机制确定设备的状态，然后通过driverService.deviceEventSender接口将设备状态交给SDK管理。
设备状态（DeviceStatus）如下：
ONLINE:在线
OFFLINE:离线
MAINTAIN:维护
FAULT:故障
*/
driverContext.getDriverMetadata().getDeviceMap().keySet().forEach(id -> driverService.deviceEventSender(id, Common.Device.Event.HEARTBEAT, Common.Device.Status.ONLINE));
}
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/service/mqtt/MqttReceiveHandler.java
+82
View File
@@ -0,0 +1,82 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.service.mqtt;
import com.alibaba.fastjson.JSON;
import com.dc3.common.model.PointValue;
import com.dc3.common.sdk.service.DriverService;
import com.dc3.driver.bean.DevicePayLoad;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.MessageHandler;
import javax.annotation.Resource;
/**
* mqtt上行数据例子：
* {
* deviceId:1,
* pointId:1,
* value:"1212"
* }
* 可根据实际情况定义 DevicePayLoad 类属性
*
* @author pnoker
*/
@Slf4j
@Configuration
public class MqttReceiveHandler {
@Resource
private DriverService driverService;
/**
* 说明：
* ConditionalOnProperty(value = "driver.mqtt.default.receive.enable")
* 根据配置属性driver.mqtt.default.receive.enable选择是否开启 Default Topic 主题的数据接收逻辑
*
* @return
*/
@Bean
@ServiceActivator(inputChannel = "defaultMqttInputChannel")
@ConditionalOnProperty(value = "driver.mqtt.default.receive.enable")
public MessageHandler defaultHandler() {
return message -> {
log.info(
"defaultTopicReceiver\nheader:{},\npayload:{}",
JSON.toJSONString(message.getHeaders(), true),
JSON.toJSONString(message.getPayload(), true)
);
};
}
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return message -> {
log.info(
"{}\nheader:{}, payload:{}",
message.getHeaders().get("mqtt_receivedTopic"),
JSON.toJSONString(message.getHeaders(), true),
JSON.toJSONString(message.getPayload(), true)
);
DevicePayLoad devicePayLoad = JSON.parseObject(message.getPayload().toString(), DevicePayLoad.class);
PointValue pointValue = new PointValue(devicePayLoad.getDeviceId(), devicePayLoad.getPointId(), devicePayLoad.getValue(),
driverService.convertValue(devicePayLoad.getDeviceId(), devicePayLoad.getPointId(), devicePayLoad.getValue()));
driverService.pointValueSender(pointValue);
};
}
}
dc3-driver/dc3-driver-edge-gateway/src/main/java/com/dc3/driver/service/mqtt/MqttSendHandler.java
+56
View File
@@ -0,0 +1,56 @@
/*
* Copyright 2016-2021 Pnoker. All Rights Reserved.
* 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 com.dc3.driver.service.mqtt;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
/**
* @author pnoker
*/
@MessagingGateway(defaultRequestChannel = "mqttOutputChannel")
public interface MqttSendHandler {
/**
* 使用 Default Topic & Default Qos 发送数据
*
* @param data string
*/
void sendToMqtt(String data);
/**
* 使用 Default Topic & 自定义 Qos 发送数据
*
* @param qos 自定义 Qos
* @param data string
*/
void sendToMqtt(@Header(MqttHeaders.QOS) Integer qos, String data);
/**
* 使用 自定义 Topic & Default Qos 发送数据
*
* @param topic 自定义 Topic
* @param data string
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data);
/**
* 使用 自定义 Topic & 自定义 Qos 发送数据
*
* @param topic 自定义 Topic
* @param qos 自定义 Qos
* @param data string
*/
void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer qos, String data);
}
dc3-driver/dc3-driver-edge-gateway/src/main/resources/application.yml
+73
View File
@@ -0,0 +1,73 @@
driver:
name: EdgeGatewayDriver
tenant: default
type: gateway
project: @project.artifactId@
description: @project.description@
schedule:
status:
enable: true
corn: '0/10 * * * * ?'
read:
enable: false
corn: '0/30 * * * * ?'
custom:
enable: true
corn: '0/5 * * * * ?'
point-attribute:
- displayName: 指令Topic
name: commandTopic
type: string
value: commandTopic
description: 测点/设备接收下行指令的Mqtt主题
option:
type: input\select\checkox\switch\time...
required: true
data-type: static/url
data: jsonString
- displayName: 指令Qos
name: commandQos
type: int
value: 2
description: 测点/设备接收下行指令的Mqtt主题的Qos
option:
type: input\select\checkox\switch\time...
required: true
data-type: static/url
data: jsonString
mqtt:
url: ssl://10.49.28.100:8883
auth-type: X509
username: iiot
password: uYLZypvzhZWqX7Ih
ca-crt: classpath:/certs/ca.crt
client-key-pass: emqx-mqtt
client-key: classpath:/certs/client.key
client-crt: classpath:/certs/client.crt
qos:
- 0
topics:
- mqtt/group/device/#
client:
id: dc3-mqtt-client
default:
topic: dc3-mqtt-topic
qos: 1
receive:
enable: true
keep-alive: 15
completion-timeout: 3000
server:
port: 8702
spring:
application:
name: ${driver.project}_${driver.tenant}
logging:
level:
com.dc3.common.sdk: DEBUG
com.dc3: DEBUG
file:
name: dc3/logs/driver/mqtt/${spring.application.name}.log
dc3-driver/dc3-driver-edge-gateway/src/main/resources/certs/ca.crt
+21
View File
@@ -0,0 +1,21 @@
-----BEGIN CERTIFICATE-----
MIIDgTCCAmmgAwIBAgIUWY7+eHeW/LofTVlu7VXUI+7MtHkwDQYJKoZIhvcNAQEL
BQAwUDELMAkGA1UEBhMCQ04xCzAJBgNVBAgMAkJKMQswCQYDVQQHDAJCSjELMAkG
A1UECgwCTUkxCzAJBgNVBAsMAklNMQ0wCwYDVQQDDARJSU9UMB4XDTIxMDcyOTEz
MTkwOVoXDTMxMDcyNzEzMTkwOVowUDELMAkGA1UEBhMCQ04xCzAJBgNVBAgMAkJK
MQswCQYDVQQHDAJCSjELMAkGA1UECgwCTUkxCzAJBgNVBAsMAklNMQ0wCwYDVQQD
DARJSU9UMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAoJb1WPaosaaR
DkvxZL5tPtoHpAurtZ8DppHU2Ip91B/vHWKf+O3yQc0dGyLUr7ZVBLvsjIjZ4BTO
BPCrhXqgVHoZttL0tCBkr63K4Asb7dxqwxkcwKZdRIz3XdoGex40hgU0huEvqEHT
D23A26JCrzglH1n7iQ/mOZrVi2vawJhLcPKDlWX2IC31vuyqdnUuTFWmuAAxvJyT
WkIhho9rd+59ODT7ghCcMdvM/4NVP3dQvVBRL14Fn07QA7ZnHYsrONuJAXWGe6Co
tsbZI095B9BQvRoMGsXojCQIOwp2IxNNz9ZmvhJkU0AvI9OapJokCYW2EB+Nv1WK
hoMxZQobLQIDAQABo1MwUTAdBgNVHQ4EFgQUJ+fZa4zhELn9fHDhlgKFFoUzjrgw
HwYDVR0jBBgwFoAUJ+fZa4zhELn9fHDhlgKFFoUzjrgwDwYDVR0TAQH/BAUwAwEB
/zANBgkqhkiG9w0BAQsFAAOCAQEAOQLIewgHTQ02Vb8nW2Z6fP8nE6F4EFzE4NP0
g+6qovPbCZxqz65Ntj9tJCeg7j4kHQaULB61IUJTdv6ssRTDrj/Lv+xP6Fyd48g5
N0WUYv8UbXZKz7qmb44GadXw+EYdhklZVndwT4M055BdT2mzug70JQpoyrdmp364
dkXEILkEE/XhQVUsM7NsLl/OT76SV2jeBxU3qJXPTdmMZc11KVf66SrSH+rILNjg
Xansswzig6s19iGF/Ww/pa7IfKpuH0DqxArFJcRYGhNYV7kBs4aVGOclru7RANum
uL02qNFlysI5u8Q3e0Lohf1D1ulrKhWznC8c8ZYM71JglEU5tw==
-----END CERTIFICATE-----
dc3-driver/dc3-driver-edge-gateway/src/main/resources/certs/client.crt
+19
View File
@@ -0,0 +1,19 @@
-----BEGIN CERTIFICATE-----
MIIDJzCCAg8CFD0pKkrBkGYtm4DWzceRMuN+EdIrMA0GCSqGSIb3DQEBCwUAMFAx
CzAJBgNVBAYTAkNOMQswCQYDVQQIDAJCSjELMAkGA1UEBwwCQkoxCzAJBgNVBAoM
Ak1JMQswCQYDVQQLDAJJTTENMAsGA1UEAwwESUlPVDAeFw0yMTA3MjkxMzIyMTZa
Fw0zMTA3MjcxMzIyMTZaMFAxCzAJBgNVBAYTAkNOMQswCQYDVQQIDAJCSjELMAkG
A1UEBwwCQkoxCzAJBgNVBAoMAk1JMQswCQYDVQQLDAJJTTENMAsGA1UEAwwESUlP
VDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMBzV9Yk2abEocj1YN3c
wdlBI+wxinyD6sktZB1h2HbEJ0H+9n6Z4Z0LsuQTfm4nTNSeW2e+XuqTc5DXMw9Z
EX8c4NYH7G8Q2rpZuJlrH/cQ2QL/cr2wf9UBwfuBBqCQxexQjR3RHyJIaQemlQkX
x4xqGk/qcMbqj+kUn582WxthRiiAzzHuDpkMKemi4QHmw/qXaNkldIKzThARM5qR
6silBZ3yD5B8CpFFBrt4j4IvELBGW4yXtB5R4NqtU58QS1UaEX/0pAAqDScNoiYr
GUjQsyXEEM3qEICcYUSvPBlSo+YtXgiyaGdvwT9DQjIvSbTAvQS0L5z+0dvciCnK
XuMCAwEAATANBgkqhkiG9w0BAQsFAAOCAQEASaGCE9Kf+j6rhcBlxnytY1nd1JoA
ECjjvm6T/AzzrJJWubPcxF6dQviEWpe3FTBdHx+vrEC3FmDEUT+8WpBxgmJ+8b/3
TpqXXD+irY/Gu8Ba/NwbrSPOYCGFe8qDOu2FppuLWfl9yvf1Ljqi/bU/j5C2rxrv
xlsaJO0WZIv96ccmm+Bp2VfNVEid5BDvrGjMeOhDDLZySBlY09yGnAxi4/8jtgk2
EyniAfyj6Ahok1TNr8l9meRLOvFbwuzf8QwBUBg9qhOijCzRhZYtd7zpmsImUdNd
vIaucduOJHL/ft7HOCvE7NiAx84fvIyNC5dhH4PCwlatHjHVZlu0jI9mgg==
-----END CERTIFICATE-----
dc3-driver/dc3-driver-edge-gateway/src/main/resources/certs/client.key
+30
View File
@@ -0,0 +1,30 @@
-----BEGIN RSA PRIVATE KEY-----
Proc-Type: 4,ENCRYPTED
DEK-Info: DES-EDE3-CBC,A8175B17E8D40281
Onv8/P2DjEyQ7Wircom/f72P9bVVAVP7dmjj44EmspxSkxdJvogPdUmEurw4jU3G
zBW4HqyuhQaEicOTGXzTyyYvOw3e/222SJElZrobcaymUrjDsgNZLWeqsGUZjLQf
6qoNjwYXjfxEhy9a3DROAWqZCYzIgzN34DroKDY5vrSfDZYMVeacvwVE2sD3nXHi
gEkppvfHWR7X9bkeXuds51kItu4I4C2pURKR3CP7mOBmyzuC+IGJXjVaVJSAIgDd
slKGFxwEhsmPF4J0XjFPqPaG5p8uwBEjMrt+LM6GLQZv/uuFefCbAyRoEWHF1vp4
PNhFFi9c9u29HofZkcCAYwxx2OacajOJ3t/m4Yfne0e3wRNa8i9b7Tl3JEviLaEp
NqveunYov8AYYGQCR6S7zFiawwTJD8uvnDXhZXWnfQhbBROqJLf2zGDYNCAa65Sy
R/gyIGDBgAyLTjqtXHHHKxDhY6/Uxf8LV16q6ihJq+CG7WNG5ojxBdCZV5Jn3ELS
NN7SQfyycD1LE7/4t0bUXO5/XkhWRIsEb4419+GhO9NDqMudz4ntksFryHoafqsR
JCNfuV+HXwnA7nZoCyLYHDGrPmBq73UIORFAH76wDzv+9ffXZ+MfFbxWstNzLGDQ
2zm7dgIXe9T1gw4L0+dG161sm7XDHRskwBLoy3pxt9Oqh7EVSmS/lo9OiJ6DilOx
tUTA2cTI2WNPhdp+9px7O6bRMI5s7q7yy1+Gxf36T0ZdT0Cthqjk6pBvViJekLtm
pt5vGN9zozBHCQzoeQa+un9QOgPEt83/k9dRDGuCfPripPRF+lloP+LLtoAKWekn
WpBZAIZKD18UsCyvJj4GWCbhK++STjFn6piXoEmFKhqGcvNbiOq/wU47WAFxy9JZ
/QhUxMJIuM9gaPvciqIjef9RUyo5aqO5+ggBTLNRAP8Cp4LtPF3g9U3o9gFc0Gal
u8BRDtHeo5bAxDiHBNoJYGrikwAaizmtqbUUlOO8gjL1O717kcmGpf/+ZtXnqAF0
SVSJ9iOJzrXf0Tgg0mml20pL9iE80/kDBEQR9lqQYInzxutBDL6IHyOWnr5eMQK2
fTlJSnxmaBoCPPGmTlKhvKKHWXPGYxeeOGhSACMVkv2va44zH5jliwdhjw/+nMN0
OM001UjGk49c8sfjxkgnY00caBEB+cIcfxrBuqZrfC6yEgwxWd8KyDrTA/Ymu0CH
XoOKvXmOP1HrWaO1wrLEhy5E/N7Op3Q+6Nzuc155NmgzPrJi+Qd1J6JLZxDI15dI
JgDY2h9v44bVsQS5FpmYfvabyMzbGgscJabf8tQ1xnnb/HIf0asx8SdaikmZI7I5
PVqt5R+aocwghvLN9GHMwVNlTdqkif3TKqeUqTHTSTAsbh9HTC+5k5s7KWs93nNd
yBzorIGmEet7+XtXFW9w/IHtQYygJOVEFfe/SSjOOa7UE+SFO+Yyr9elMIGAfMvy
QykELDv6SB1rk6fJ35gsoONvOFyGXDkm68jMrSulY/u/RTTbZrE65c69QLaHaAaq
7rl70yv95LrzS8it8qduANjuoHQMcxD/3OnWsxe+UwoKcJccp6SDiUN5Awre01SZ
-----END RSA PRIVATE KEY-----
dc3-driver/dc3-driver-mqtt/src/main/resources/application.yml
+1 -1
View File
@@ -55,7 +55,7 @@ driver:
qos: 1
receive:
enable: true
keep-alive: 5
keep-alive: 15
completion-timeout: 3000
server:
dc3-driver/pom.xml
+1
View File
@@ -33,6 +33,7 @@
<module>dc3-driver-opc-ua</module>
<module>dc3-driver-mqtt</module>
<module>dc3-driver-modbus-tcp</module>
<module>dc3-driver-edge-gateway</module>
</modules>
<properties>
dc3/dependencies/mysql/iot-dc3.sql
+1 -1
View File
@@ -30,9 +30,9 @@ CREATE TABLE `dc3_driver`
`id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
`name` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '协议名称',
`service_name` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '协议服务名称',
`type` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 'driver' COMMENT '类型，deriver、gateway...',
`host` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主机IP',
`port` int(11) NOT NULL COMMENT '端口',
`type` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT 'driver' COMMENT '类型，deriver、gateway...',
`enable` tinyint(4) NULL DEFAULT 1 COMMENT '是否可用',
`tenant_id` bigint(20) NOT NULL COMMENT '租户ID',
`description` varchar(380) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT '' COMMENT '描述',
dc3/docker-compose.yml
+27
View File
@@ -387,6 +387,33 @@ services:
aliases:
- dc3-driver-modbus-tcp
edge-gateway:
build:
context: ../dc3-driver/dc3-driver-edge-gateway/
dockerfile: ./Dockerfile
image: pnoker/dc3-driver-edge-gateway:1.0.0
restart: always
ports:
- 8702:8702
environment:
- RABBITMQ_VIRTUAL_HOST=dc3
- RABBITMQ_HOST=dc3-rabbitmq
- RABBITMQ_PORT=5672
- RABBITMQ_USERNAME=dc3
- RABBITMQ_PASSWORD=dc3
- EUREKA_HOST=dc3-register
- EUREKA_PORT=8100
- SECURITY_USER_NAME=dc3
- SECURITY_USER_PASSWORD=dc3
container_name: dc3-driver-edge-gateway
hostname: dc3-driver-edge-gateway
volumes:
- logs:/dc3-driver/dc3-driver-edge-gateway/dc3/logs
networks:
dc3net:
aliases:
- dc3-driver-edge-gateway
rtmp:
build:
context: ../dc3-transfer/dc3-rtmp/