zcy
/
iotkit-parent
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'dev'

V0.5.x
xiwa 2022-05-24 16:40:04 +08:00
parent 09c3a0dfc8 a9234f6773
commit 69e7983e90
29 changed files with 1189 additions and 337 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -4,7 +4,7 @@
此仓库为奇特物联(iotkit)物联网平台开源项目。
奇特物联是一个开源的物联网基础开发平台，提供了物联网及相关业务开发的常见基础功能, 能帮助你快速搭建自己的物联网相关业务平台。
系统包含了品类、物模型、消息转换、通讯组件（mqtt通讯组件、小度音箱接入组件、onenet Studio接入组件）、云端低代码设备开发、设备管理、规则引擎、第三方平台接入、数据流转、数据可视化、报警中心等模块和智能家居APP（小程序），集成了[Sa-Token](https://gitee.com/dromara/sa-token) 认证框架。
系统包含了品类、物模型、消息转换、通讯组件（mqtt/EMQX通讯组件、小度音箱接入组件、onenet Studio接入组件）、云端低代码设备开发、设备管理、规则引擎、第三方平台接入、数据流转、数据可视化、报警中心等模块和智能家居APP（小程序），集成了[Sa-Token](https://gitee.com/dromara/sa-token) 认证框架。
**前端项目见：** https://gitee.com/iotkit-open-source/iot-console-web

4
common/src/main/java/cc/iotkit/common/Constants.java
View File

@ -160,4 +160,8 @@ public interface Constants {
*/
String GET_DEVICE = "/device/{deviceId}";
}
interface MQTT {
String DEVICE_SUBSCRIBE_TOPIC = "^/sys/.+/.+/c/#$";
}
}

1
manager/src/main/java/cc/iotkit/manager/Application.java
View File

@ -19,6 +19,7 @@ public class Application {
if (EmbeddedRedisConfig.embeddedEnable()) {
EmbeddedRedisConfig.startEmbeddedRedisServer();
}
System.setProperty("nashorn.args","--no-deprecation-warning");
SpringApplication.run(Application.class, args);
}

54
manager/src/main/java/cc/iotkit/manager/config/SaTokenConfigure.java
View File

@ -3,6 +3,7 @@ package cc.iotkit.manager.config;
import cn.dev33.satoken.interceptor.SaAnnotationInterceptor;
import cn.dev33.satoken.interceptor.SaRouteInterceptor;
import cn.dev33.satoken.router.SaRouter;
import cn.dev33.satoken.router.SaRouterStaff;
import cn.dev33.satoken.stp.StpUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
@ -20,28 +21,35 @@ public class SaTokenConfigure implements WebMvcConfigurer {
// 注册路由拦截器，自定义认证规则
registry.addInterceptor(new SaRouteInterceptor((req, res, handler) -> {
log.info("resource role check,path:{}", req.getRequestPath());
SaRouter
//管理员、系统、客户端用户角色能使用的功能
.match("/space/addSpace/**",
"/space/saveSpace/**",
"/space/delSpace/**",
"/space/saveHome/**",
"/space/currentHome/**",
"/space/myRecentDevices/**",
"/space/spaces/**",
"/space/myDevices/**",
"/space/findDevice/**",
"/space/addDevice/**",
"/space/saveDevice",
"/space/removeDevice",
"/space/device/*",
"/device/*/consumer/*",
"/device/*/service/property/set",
"/device/*/service/*/invoke"
)
.check(c -> StpUtil.checkRoleOr("iot_admin", "iot_system", "iot_client"));
//客户端角色能使用的功能
if (StpUtil.hasRole("iot_client")) {
if (SaRouter
.match("/space/addSpace/**",
"/space/saveSpace/**",
"/space/delSpace/**",
"/space/saveHome/**",
"/space/currentHome/**",
"/space/myRecentDevices/**",
"/space/spaces/**",
"/space/myDevices/**",
"/space/findDevice/**",
"/space/addDevice/**",
"/space/saveDevice",
"/space/removeDevice",
"/space/device/*",
"/device/*/consumer/*",
"/device/*/service/property/set",
"/device/*/service/*/invoke"
).isHit()) {
return;
}
}
SaRouter
//除了以上所有功能都需要 管理员或系统用户角色
.match("/**")
.check(c -> StpUtil.checkRoleOr("iot_admin", "iot_system"))
//需要有可写权限的功能
.match(
"/**/save*/**",
@ -54,12 +62,6 @@ public class SaTokenConfigure implements WebMvcConfigurer {
"/**/invoke"
).check(c -> StpUtil.checkPermission("write"));
SaRouter
//管理员、系统用户角色能使用的功能
.match("/**")
.check(c -> StpUtil.checkRoleOr("iot_admin", "iot_system", "iot_client"))
;
})).addPathPatterns("/**")
.excludePathPatterns(
"/*.png",

154
manager/src/main/java/cc/iotkit/manager/service/KeycloakAdminService.java1
View File

@ -1,154 +0,0 @@
package cc.iotkit.manager.service;
import cc.iotkit.common.exception.BizException;
import cc.iotkit.common.utils.JsonUtil;
import cc.iotkit.model.UserInfo;
import lombok.extern.slf4j.Slf4j;
import org.keycloak.admin.client.Keycloak;
import org.keycloak.admin.client.KeycloakBuilder;
import org.keycloak.admin.client.resource.UserResource;
import org.keycloak.admin.client.resource.UsersResource;
import org.keycloak.representations.idm.CredentialRepresentation;
import org.keycloak.representations.idm.UserRepresentation;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@Slf4j
@Service
public class KeycloakAdminService {
@Value("${keycloak.realm}")
private String realm;
@Value("${keycloak.auth-server-url}")
private String authServerUrl;
@Value("${keycloak-admin-user}")
private String adminUser;
@Value("${keycloak-admin-password}")
private String adminPassword;
@Value("${keycloak-admin-clientid}")
private String adminClientId;
private Keycloak keycloak;
private Keycloak getKeycloak() {
if (keycloak == null) {
keycloak = KeycloakBuilder.builder()
.serverUrl(authServerUrl)
.username(adminUser)
.password(adminPassword)
.clientId(adminClientId)
.realm(realm)
.build();
}
return keycloak;
}
public String createUser(UserInfo user, String pwd) {
Keycloak keycloak = getKeycloak();
UsersResource usersResource = keycloak.realm(realm)
.users();
UserRepresentation userRepresentation = new UserRepresentation();
userRepresentation.setUsername(user.getUid());
userRepresentation.setGroups(Collections.singletonList(getGroup(user.getType())));
userRepresentation.setRealmRoles(user.getRoles());
if (user.getEmail() != null) {
userRepresentation.setEmail(user.getEmail());
}
userRepresentation.setEnabled(true);
userRepresentation.setFirstName(user.getNickName());
CredentialRepresentation credentialRepresentation = new CredentialRepresentation();
credentialRepresentation.setType(CredentialRepresentation.PASSWORD);
credentialRepresentation.setValue(pwd);
credentialRepresentation.setTemporary(false);
userRepresentation.setCredentials(Collections.singletonList(credentialRepresentation));
javax.ws.rs.core.Response response = usersResource.create(userRepresentation);
String url = response.getLocation().getPath();
String newUid = url.substring(url.lastIndexOf("/") + 1);
if (response.getStatus() >= 300) {
log.error("create userRepresentation response:{}", JsonUtil.toJsonString(response));
throw new BizException("create keycloak user failed");
}
return newUid;
}
public void updateUser(UserInfo user) {
Keycloak keycloak = getKeycloak();
UserResource userResource = keycloak.realm(realm)
.users().get(user.getId());
UserRepresentation userRepresentation = userResource.toRepresentation();
if (user.getUid() != null) {
userRepresentation.setUsername(user.getUid());
}
if (user.getEmail() != null) {
userRepresentation.setEmail(user.getEmail());
}
if (user.getType() != null) {
userRepresentation.setGroups(Arrays.asList(getGroup(user.getType())));
}
if (user.getRoles() != null) {
userRepresentation.setRealmRoles(user.getRoles());
}
userResource.update(userRepresentation);
}
public UserInfo getUser(String uid) {
Keycloak keycloak = getKeycloak();
List<UserRepresentation> users = keycloak.realm(realm)
.users().search(uid);
if (users.size() == 0) {
return null;
}
UserRepresentation user = users.get(0);
return UserInfo.builder()
.id(user.getId())
.uid(uid)
.build();
}
public void resetUserPwd(String id, String pwd) {
Keycloak keycloak = getKeycloak();
UserResource userResource = keycloak.realm(realm)
.users().get(id);
UserRepresentation userRepresentation = userResource.toRepresentation();
CredentialRepresentation credentialRepresentation = new CredentialRepresentation();
credentialRepresentation.setType(CredentialRepresentation.PASSWORD);
credentialRepresentation.setValue(pwd);
credentialRepresentation.setTemporary(false);
userRepresentation.setCredentials(Arrays.asList(credentialRepresentation));
userResource.update(userRepresentation);
}
public void deleteUser(String id) {
Keycloak keycloak = getKeycloak();
UserResource userResource = keycloak.realm(realm)
.users().get(id);
try {
userResource.remove();
} catch (javax.ws.rs.NotFoundException e) {
log.warn("user does not exist");
}
}
private String getGroup(Integer type) {
if (type == null) {
return "";
}
return type == UserInfo.USER_TYPE_PLATFORM
? "platform" : "client";
}
}

46
oauth2-server/src/main/java/cc/iotkit/oauth/controller/AuthClientController.java
View File

@ -17,13 +17,15 @@ import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cglib.beans.BeanMap;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.ModelAndView;
import org.springframework.web.servlet.view.RedirectView;
import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.Map;
@Slf4j
@RestController
@ -38,14 +40,9 @@ public class AuthClientController {
@Autowired
private UserInfoCache userInfoCache;
// 进入首页
@RequestMapping("/")
public Object index(HttpServletRequest request) {
request.setAttribute("uid", StpUtil.getLoginIdDefaultNull());
return new ModelAndView("index.html");
}
// 根据Code码进行登录，获取 Access-Token 和 openid
/**
* 根据Code码进行登录，获取 Access-Token 和 用户信息
*/
@RequestMapping("/codeLogin")
public SaResult codeLogin(String code, String clientId) {
OauthClient oauthClient = oauthClientCache.getClient(clientId);
@ -71,19 +68,24 @@ public class AuthClientController {
}
// 根据openid获取其对应的userId
SoMap data = new SoMap();
String uid = getUserIdByOpenid(so.getString("openid"));
String access_token = so.getString("access_token");
UserInfo userInfo = userInfoCache.getUserInfo(uid);
data.put("name", userInfo.getNickName());
data.put("uid", uid);
UserInfoVo userVo = getUserInfo(uid);
BeanMap beanMap = BeanMap.create(userVo);
Map<String,Object> data=new HashMap<>();
beanMap.forEach((key,value)->{
data.put(key.toString(),value);
});
data.put("access_token", access_token);
// 返回相关参数
StpUtil.login(uid, SaLoginConfig.setToken(access_token));
return SaResult.data(data);
}
// 注销登录
/**
* 注销登录
*/
@RequestMapping("/logout")
public RedirectView logout(String accessToken, String redirect_uri) {
//先注销client中cookie的token
@ -94,13 +96,14 @@ public class AuthClientController {
return new RedirectView(redirect_uri);
}
/**
* 登录验证
*/
@GetMapping("/checkLogin")
public SaResult checkLogin() {
try {
String uid = StpUtil.getLoginId().toString();
UserInfo userInfo = userInfoCache.getUserInfo(uid);
UserInfoVo userVo = new UserInfoVo();
ReflectUtil.copyNoNulls(userInfo, userVo);
UserInfoVo userVo = getUserInfo(uid);
return SaResult.ok().setData(userVo);
} catch (Throwable e) {
return SaResult.error("no login");
@ -113,4 +116,11 @@ public class AuthClientController {
return clientIdLoginId.split(":")[1];
}
private UserInfoVo getUserInfo(String uid) {
UserInfo userInfo = userInfoCache.getUserInfo(uid);
UserInfoVo userVo = new UserInfoVo();
ReflectUtil.copyNoNulls(userInfo, userVo);
return userVo;
}
}

37
oauth2-server/src/main/java/cc/iotkit/oauth/controller/AuthServerController.java
View File

@ -5,9 +5,7 @@ import cc.iotkit.dao.UserInfoRepository;
import cc.iotkit.model.UserInfo;
import cc.iotkit.oauth.service.TokenRequestHandler;
import cc.iotkit.utils.AuthUtil;
import cn.dev33.satoken.context.SaHolder;
import cn.dev33.satoken.oauth2.config.SaOAuth2Config;
import cn.dev33.satoken.oauth2.logic.SaOAuth2Util;
import cn.dev33.satoken.stp.StpUtil;
import cn.dev33.satoken.util.SaResult;
import lombok.extern.slf4j.Slf4j;
@ -19,7 +17,6 @@ import org.springframework.web.servlet.ModelAndView;
import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
@Slf4j
@ -29,7 +26,9 @@ public class AuthServerController {
@Autowired
private UserInfoRepository userInfoRepository;
// 处理所有OAuth相关请求
/**
* 处理所有OAuth相关请求
*/
@RequestMapping("/oauth2/*")
public Object request(HttpServletRequest request) {
Object result = TokenRequestHandler.serverRequest();
@ -37,7 +36,9 @@ public class AuthServerController {
return result;
}
// Sa-OAuth2 定制化配置
/**
* Sa-OAuth2 自定义配置
*/
@Autowired
public void setSaOAuth2Config(SaOAuth2Config cfg) {
cfg.
@ -67,6 +68,11 @@ public class AuthServerController {
return new ModelAndView("confirm.html", map);
})
;
//开启密码授权、刷新token和client授权模式
cfg.setIsPassword(true);
cfg.setIsNewRefresh(true);
cfg.setIsClient(true);
}
// 全局异常拦截
@ -76,28 +82,9 @@ public class AuthServerController {
return SaResult.error(e.getMessage());
}
// ---------- 开放相关资源接口： Client端根据 Access-Token ，置换相关资源 ------------
// 获取Userinfo信息：昵称、头像、性别等等
@RequestMapping("/oauth2/userinfo")
public SaResult userinfo() {
// 获取 Access-Token 对应的账号id
String accessToken = SaHolder.getRequest().getParamNotNull("access_token");
Object loginId = SaOAuth2Util.getLoginIdByAccessToken(accessToken);
System.out.println("-------- 此Access-Token对应的账号id: " + loginId);
// 校验 Access-Token 是否具有权限: userinfo
SaOAuth2Util.checkScope(accessToken, "userinfo");
// 模拟账号信息 （真实环境需要查询数据库获取信息）
Map<String, Object> map = new LinkedHashMap<>();
map.put("nickname", "shengzhang_");
map.put("avatar", "http://xxx.com/1.jpg");
map.put("age", "18");
map.put("sex", "男");
map.put("address", "山东省 青岛市 城阳区");
return SaResult.data(map);
return SaResult.ok();
}
}

4
oauth2-server/src/test/java/GenPwdSecret.java
View File

@ -6,9 +6,9 @@ public class GenPwdSecret {
@Test
public void gen() throws Exception {
//生成密码加密内容
String secret = AuthUtil.enCryptPwd("c123456");
String secret = AuthUtil.enCryptPwd("guest123");
System.out.println(secret);
System.out.println(AuthUtil.checkPwd("c123456", secret));
System.out.println(AuthUtil.checkPwd("guest123", secret));
}
}

62
protocol-gateway/component-server/src/main/java/cc/iotkit/comps/DeviceComponentManager.java
View File

@ -5,6 +5,7 @@ import cc.iotkit.common.ComponentClassLoader;
import cc.iotkit.common.exception.BizException;
import cc.iotkit.common.utils.JsonUtil;
import cc.iotkit.comp.CompConfig;
import cc.iotkit.comp.IComponent;
import cc.iotkit.comp.IDeviceComponent;
import cc.iotkit.comps.config.CacheKey;
import cc.iotkit.comps.config.ComponentConfig;
@ -55,6 +56,8 @@ public class DeviceComponentManager {
private DeviceCache deviceCache;
@Autowired
ProductCache productCache;
@Autowired
private DeviceRouter deviceRouter;
@PostConstruct
public void init() {
@ -126,9 +129,10 @@ public class DeviceComponentManager {
if (component == null) {
return;
}
DeviceMessageHandler messageHandler = new DeviceMessageHandler(this, component,
DeviceMessageHandler messageHandler = new DeviceMessageHandler(
this, component,
component.getScript(), component.getConverter(),
deviceBehaviourService);
deviceBehaviourService, deviceRouter);
messageHandler.putScriptEnv("apiTool", new ApiTool());
messageHandler.putScriptEnv("deviceBehaviour", deviceBehaviourService);
@ -168,34 +172,34 @@ public class DeviceComponentManager {
linkDn = parent.getDeviceName();
}
for (IDeviceComponent com : components.values()) {
if (com.exist(linkPk, linkDn)) {
Device device = new Device(deviceInfo.getDeviceId(), deviceInfo.getModel(), product.isTransparent());
//对下发消息进行编码转换
DeviceMessage message = com.getConverter().encode(service, device);
if (message == null) {
throw new BizException("encode send message failed");
}
//保存设备端mid与平台mid对应关系
redisTemplate.opsForValue().set(
CacheKey.getKeyCmdMid(message.getDeviceName(), message.getMid()),
service.getMid(), com.getConfig().getCmdTimeout(), TimeUnit.SECONDS);
com.send(message);
ThingModelMessage thingModelMessage = ThingModelMessage.builder()
.mid(service.getMid())
.productKey(service.getProductKey())
.deviceName(service.getDeviceName())
.identifier(service.getIdentifier())
.type(service.getType())
.data(service.getParams())
.build();
deviceBehaviourService.reportMessage(thingModelMessage);
return;
}
IComponent component = deviceRouter.getRouter(linkPk, linkDn);
if (!(component instanceof IDeviceComponent)) {
throw new BizException("send destination does not exist");
}
throw new BizException("send destination not found");
IDeviceComponent deviceComponent = (IDeviceComponent) component;
Device device = new Device(deviceInfo.getDeviceId(), deviceInfo.getModel(), product.isTransparent());
//对下发消息进行编码转换
DeviceMessage message = deviceComponent.getConverter().encode(service, device);
if (message == null) {
throw new BizException("encode send message failed");
}
//保存设备端mid与平台mid对应关系
redisTemplate.opsForValue().set(
CacheKey.getKeyCmdMid(message.getDeviceName(), message.getMid()),
service.getMid(), deviceComponent.getConfig().getCmdTimeout(), TimeUnit.SECONDS);
//发送消息给设备
deviceComponent.send(message);
ThingModelMessage thingModelMessage = ThingModelMessage.builder()
.mid(service.getMid())
.productKey(service.getProductKey())
.deviceName(service.getDeviceName())
.identifier(service.getIdentifier())
.type(service.getType())
.data(service.getParams())
.build();
deviceBehaviourService.reportMessage(thingModelMessage);
}
public String getPlatformMid(String deviceName, String mid) {

22
protocol-gateway/component-server/src/main/java/cc/iotkit/comps/DeviceMessageHandler.java
View File

@ -39,15 +39,22 @@ public class DeviceMessageHandler implements IMessageHandler {
private final IDeviceComponent component;
private final DeviceRouter deviceRouter;
@SneakyThrows
public DeviceMessageHandler(DeviceComponentManager deviceComponentManager,
IDeviceComponent component,
String script, IConverter converter,
DeviceBehaviourService deviceBehaviourService) {
DeviceBehaviourService deviceBehaviourService,
DeviceRouter deviceRouter
) {
this.deviceComponentManager = deviceComponentManager;
this.component = component;
this.converter = converter;
this.deviceBehaviourService = deviceBehaviourService;
this.deviceRouter = deviceRouter;
engine.put("component", component);
scriptObj = engine.eval(String.format("new (function () {\n%s})()", script));
}
@ -64,6 +71,7 @@ public class DeviceMessageHandler implements IMessageHandler {
if (!(data instanceof Map)) {
throw new BizException("script result data is incorrect");
}
Map<String, Object> dataMap = (Map) data;
//获取动作数据
Action action = getAction(result.get("action"));
@ -142,10 +150,16 @@ public class DeviceMessageHandler implements IMessageHandler {
private void doStateChange(DeviceState state) {
try {
String pk = state.getProductKey();
String dn = state.getDeviceName();
boolean isOnline = DeviceState.STATE_ONLINE.equals(state.getState());
if (isOnline) {
deviceRouter.putRouter(pk, dn, component);
} else {
deviceRouter.removeRouter(pk, dn);
}
component.onDeviceStateChange(state);
deviceBehaviourService.deviceStateChange(state.getProductKey(),
state.getDeviceName(),
DeviceState.STATE_ONLINE.equals(state.getState()));
deviceBehaviourService.deviceStateChange(pk, dn, isOnline);
} catch (Throwable e) {
log.error("device state change error", e);
}

47
protocol-gateway/component-server/src/main/java/cc/iotkit/comps/DeviceRouter.java Normal file
View File

@ -0,0 +1,47 @@
package cc.iotkit.comps;
import cc.iotkit.comp.IComponent;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
/**
* 设备路由
*/
@Component
public class DeviceRouter {
private static final String DEVICE_ROUTER = "str:device:router:%s:%s";
private static final Map<String, IComponent> components = new HashMap<>();
@Autowired
private StringRedisTemplate redisTemplate;
private String getDeviceRouter(String productKey, String deviceName) {
return String.format(DEVICE_ROUTER, productKey, deviceName);
}
public void putRouter(String productKey, String deviceName, IComponent component) {
String comId = component.getId();
components.put(comId, component);
redisTemplate.opsForValue().set(getDeviceRouter(productKey, deviceName), component.getId());
}
public void removeRouter(String productKey, String deviceName) {
redisTemplate.delete(getDeviceRouter(productKey, deviceName));
}
public IComponent getRouter(String productKey, String deviceName) {
String comId = redisTemplate.opsForValue().get(getDeviceRouter(productKey, deviceName));
if (StringUtils.isBlank(comId)) {
return null;
}
return components.get(comId);
}
}

8
protocol-gateway/component-server/src/main/java/cc/iotkit/comps/config/ComponentConfig.java
View File

@ -1,7 +1,10 @@
package cc.iotkit.comps.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.nio.file.Path;
@ -26,4 +29,9 @@ public class ComponentConfig {
return Paths.get(converterDir, conId)
.toAbsolutePath().normalize();
}
@Bean("objectMapper")
public ObjectMapper myMapper() {
return new ObjectMapper().disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
}
}

36
protocol-gateway/component-server/src/main/java/cc/iotkit/comps/service/DeviceBehaviourService.java
View File

@ -15,14 +15,13 @@ import cc.iotkit.model.product.Product;
import cc.iotkit.model.product.ProductModel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;
@ -46,7 +45,10 @@ public class DeviceBehaviourService {
// @Autowired
private DeviceStateHolder deviceStateHolder;
private Producer<ThingModelMessage> deviceMessageProducer;
//旧实现，ThingModelMessage序列化失败
//private Producer<ThingModelMessage> deviceMessageProducer;
private Producer<byte[]> deviceMessageProducer;
@PostConstruct
public void init() throws PulsarClientException {
@ -54,9 +56,16 @@ public class DeviceBehaviourService {
PulsarClient client = PulsarClient.builder()
.serviceUrl(serverConfig.getPulsarBrokerUrl())
.build();
/**
旧实现，ThingModelMessage序列化失败
deviceMessageProducer = client.newProducer(JSONSchema.of(ThingModelMessage.class))
.topic("persistent://iotkit/default/" + Constants.THING_MODEL_MESSAGE_TOPIC)
.create();
*/
deviceMessageProducer = client.newProducer()
.topic("persistent://iotkit/default/" + Constants.THING_MODEL_MESSAGE_TOPIC)
.create();
}
public void register(RegisterInfo info) {
@ -125,6 +134,10 @@ public class DeviceBehaviourService {
device.setState(new DeviceInfo.State(false, null, null));
device.setCreateAt(System.currentTimeMillis());
reportMsg = true;
//auth、acl
}
//透传设备，默认在线
@ -186,6 +199,7 @@ public class DeviceBehaviourService {
boolean online) {
DeviceInfo device = deviceRepository.findByProductKeyAndDeviceName(productKey, deviceName);
if (device == null) {
log.warn(String.format("productKey: %s,device: %s,online: %s",productKey,device,online));
throw new BizException("device does not exist");
}
deviceStateChange(device, online);
@ -245,8 +259,18 @@ public class DeviceBehaviourService {
message.setTime(System.currentTimeMillis());
}
message.setDeviceId(device.getDeviceId());
deviceMessageProducer.send(message);
} catch (PulsarClientException e) {
// 旧实现，ThingModelMessage序列化失败
//deviceMessageProducer.send(message);
// 新实现，用JsonUtil.toJsonString序列化ThingModelMessage，解决 ThingModelMessage序列化失败的问题
TypedMessageBuilder<byte[]> builder = deviceMessageProducer.newMessage();
builder.value(JsonUtil.toJsonString(message).getBytes(StandardCharsets.UTF_8));
builder.send();
}
catch (PulsarClientException e) {
log.error("send thing model message error", e);
}
}

5
protocol-gateway/component/src/main/java/cc/iotkit/comp/AbstractDeviceComponent.java
View File

@ -6,6 +6,8 @@ import cc.iotkit.comp.model.RegisterInfo;
import cc.iotkit.converter.IConverter;
import lombok.Data;
import java.util.UUID;
@Data
public abstract class AbstractDeviceComponent implements IDeviceComponent {
@ -17,9 +19,12 @@ public abstract class AbstractDeviceComponent implements IDeviceComponent {
protected String script;
private String id;
@Override
public void create(CompConfig config) {
this.config = config;
this.id = UUID.randomUUID().toString();
}
@Override

2
protocol-gateway/component/src/main/java/cc/iotkit/comp/IComponent.java
View File

@ -2,6 +2,8 @@ package cc.iotkit.comp;
public interface IComponent {
String getId();
void create(CompConfig config);
void start();

2
protocol-gateway/component/src/main/java/cc/iotkit/comp/IDeviceComponent.java
View File

@ -16,8 +16,6 @@ public interface IDeviceComponent extends IComponent {
void send(DeviceMessage message);
boolean exist(String productKey, String deviceName);
void setHandler(IMessageHandler handler);
void setConverter(IConverter converter);

133
protocol-gateway/component/src/main/java/cc/iotkit/comp/utils/SpringUtils.java Normal file
View File

@ -0,0 +1,133 @@
package cc.iotkit.comp.utils;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
/**
* spring工具类 方便在非spring管理环境中获取bean
*/
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware
{
/** Spring应用上下文环境 */
private static ConfigurableListableBeanFactory beanFactory;
private static ApplicationContext applicationContext;
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException
{
SpringUtils.beanFactory = beanFactory;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException
{
SpringUtils.applicationContext = applicationContext;
}
/**
* 获取对象
*
* @param name
* @return Object 一个以所给名字注册的bean的实例
* @throws org.springframework.beans.BeansException
*
*/
@SuppressWarnings("unchecked")
public static <T> T getBean(String name) throws BeansException
{
return (T) beanFactory.getBean(name);
}
/**
* 获取类型为requiredType的对象
*
* @param clz
* @return
* @throws org.springframework.beans.BeansException
*
*/
public static <T> T getBean(Class<T> clz) throws BeansException
{
T result = (T) beanFactory.getBean(clz);
return result;
}
/**
* 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
*
* @param name
* @return boolean
*/
public static boolean containsBean(String name)
{
return beanFactory.containsBean(name);
}
/**
* 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
*
* @param name
* @return boolean
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.isSingleton(name);
}
/**
* @param name
* @return Class 注册对象的类型
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static Class<?> getType(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.getType(name);
}
/**
* 如果给定的bean名字在bean定义中有别名，则返回这些别名
*
* @param name
* @return
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.getAliases(name);
}
/**
* 获取aop代理对象
*
* @param invoker
* @return
*/
@SuppressWarnings("unchecked")
public static <T> T getAopProxy(T invoker)
{
return (T) AopContext.currentProxy();
}
/**
* 获取当前的环境配置，无配置返回null
*
* @return 当前的环境配置
*/
public static String[] getActiveProfiles()
{
return applicationContext.getEnvironment().getActiveProfiles();
}
}

27
protocol-gateway/emqx-component/dependency-reduced-pom.xml
View File

@ -26,6 +26,21 @@
<include>io.vertx:vertx-core</include>
<include>io.vertx:vertx-web-proxy</include>
<include>io.vertx:vertx-mqtt</include>
<include>io.vertx:vertx-web</include>
<include>io.vertx:vertx-http-proxy</include>
<include>org.luaj:luaj-jse</include>
<include>io.netty:netty-common</include>
<include>io.netty:netty-transport</include>
<include>io.netty:netty-handler</include>
<include>io.netty:netty-resolver</include>
<include>io.netty:netty-buffer</include>
<include>io.netty:netty-handler</include>
<include>io.netty:netty-proxy</include>
<include>io.netty:netty-codec</include>
<include>io.netty:netty-codec-mqtt</include>
<include>io.netty:netty-codec-dns</include>
<include>io.netty:netty-resolver-dns</include>
<include>io.netty:netty-tcnative-boringssl-static</include>
</includes>
</artifactSet>
</configuration>
@ -76,5 +91,17 @@
<version>0.2.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.luaj</groupId>
<artifactId>luaj-jse</artifactId>
<version>3.0.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>cc.iotkit</groupId>
<artifactId>dao</artifactId>
<version>0.2.0-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

31
protocol-gateway/emqx-component/pom.xml
View File

@ -43,6 +43,15 @@
<artifactId>component</artifactId>
</dependency>
<dependency>
<groupId>org.luaj</groupId>
<artifactId>luaj-jse</artifactId>
</dependency>
<dependency>
<groupId>cc.iotkit</groupId>
<artifactId>dao</artifactId>
</dependency>
</dependencies>
<build>
@ -62,9 +71,31 @@
<configuration>
<artifactSet>
<includes>
<!-- <include>io.vertx:vertx-core</include>-->
<!-- <include>io.vertx:vertx-web-proxy</include>-->
<!-- <include>io.vertx:vertx-mqtt</include>-->
<include>io.vertx:vertx-core</include>
<include>io.vertx:vertx-web-proxy</include>
<include>io.vertx:vertx-mqtt</include>
<include>io.vertx:vertx-web</include>
<include>io.vertx:vertx-http-proxy</include>
<include>org.luaj:luaj-jse</include>
<include>io.netty:netty-common</include>
<include>io.netty:netty-transport</include>
<include>io.netty:netty-handler</include>
<include>io.netty:netty-resolver</include>
<include>io.netty:netty-buffer</include>
<include>io.netty:netty-handler</include>
<include>io.netty:netty-proxy</include>
<include>io.netty:netty-codec</include>
<include>io.netty:netty-codec-mqtt</include>
<include>io.netty:netty-codec-dns</include>
<include>io.netty:netty-resolver-dns</include>
<include>io.netty:netty-tcnative-boringssl-static</include>
</includes>
</artifactSet>
</configuration>

24
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/AuthVerticle.java
View File

@ -4,8 +4,8 @@ import cc.iotkit.comp.IMessageHandler;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServer;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.handler.BodyHandler;
import lombok.extern.slf4j.Slf4j;
import java.util.HashMap;
@ -31,12 +31,20 @@ public class AuthVerticle extends AbstractVerticle {
@Override
public void start() throws Exception {
backendServer = vertx.createHttpServer();
Router backendRouter = Router.router(vertx);
//第一步 声明Router&初始化Router
Router backendRouter = Router.router(vertx);
//获取body参数，得先添加这句
backendRouter.route().handler(BodyHandler.create());
//第二步 配置Router解析url
backendRouter.route(HttpMethod.POST, "/mqtt/auth").handler(rc -> {
JsonObject json = rc.getBodyAsJson();
String json = rc.getBodyAsString();
log.info("mqtt auth:{}", json);
try {
executor.onReceive(new HashMap<>(), "auth", json.toString());
Map<String, Object> head = new HashMap<>();
head.put("topic", "/mqtt/auth");
executor.onReceive(head, "auth", json);
rc.response().setStatusCode(200)
.end();
} catch (Throwable e) {
@ -46,11 +54,13 @@ public class AuthVerticle extends AbstractVerticle {
}
});
backendRouter.route(HttpMethod.POST, "/mqtt/acl").handler(rc -> {
JsonObject json = rc.getBodyAsJson();
String json = rc.getBodyAsString();
log.info("mqtt acl:{}", json);
try {
Map<String, Object> head = new HashMap<>();
head.put("topic", json.getString("topic"));
executor.onReceive(head, "subscribe", json.toString());
head.put("topic", "/mqtt/acl");
executor.onReceive(head, "acl", json);
rc.response().setStatusCode(200)
.end();
} catch (Throwable e) {

191
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/EmqxDeviceComponent.java
View File

@ -4,29 +4,45 @@ import cc.iotkit.common.exception.BizException;
import cc.iotkit.common.utils.JsonUtil;
import cc.iotkit.comp.AbstractDeviceComponent;
import cc.iotkit.comp.CompConfig;
import cc.iotkit.comp.IMessageHandler;
import cc.iotkit.comp.model.DeviceState;
import cc.iotkit.comp.utils.SpringUtils;
import cc.iotkit.converter.DeviceMessage;
import cc.iotkit.converter.ThingService;
import cc.iotkit.dao.DeviceRepository;
import cc.iotkit.model.device.DeviceInfo;
import cc.iotkit.model.device.message.ThingModelMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import io.vertx.mqtt.messages.MqttConnAckMessage;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import lombok.*;
import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.CountDownLatch;
@Slf4j
public class EmqxDeviceComponent extends AbstractDeviceComponent {
private static final Logger log = LoggerFactory.getLogger(EmqxDeviceComponent.class);
private Vertx vertx;
private AuthVerticle authVerticle;
private CountDownLatch countDownLatch;
private String deployedId;
private EmqxConfig mqttConfig;
private MqttClient client;
//组件mqtt clientId，默认通过mqtt auth / acl验证。
private final Set<String> compMqttClientIdList = new HashSet<>();
private final TransparentConverter transparentConverter = new TransparentConverter();
public void create(CompConfig config) {
super.create(config);
@ -38,17 +54,21 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
@Override
public void start() {
try {
compMqttClientIdList.add(mqttConfig.getClientId());
authVerticle.setExecutor(getHandler());
countDownLatch = new CountDownLatch(1);
Future<String> future = vertx.deployVerticle(authVerticle);
future.onSuccess((s -> {
deployedId = s;
countDownLatch.countDown();
log.error("start emqx auth component success");
}));
future.onFailure((e) -> {
countDownLatch.countDown();
log.error("start emqx auth component failed", e);
});
countDownLatch.await();
MqttClientOptions options = new MqttClientOptions()
@ -58,58 +78,64 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
.setCleanSession(true)
.setKeepAliveInterval(60);
if (mqttConfig.isSsl()) {
options.setSsl(true)
.setTrustAll(true);
}
MqttClient client = MqttClient.create(vertx, options);
Future<MqttConnAckMessage> connFuture =
client.connect(mqttConfig.getPort(), mqttConfig.getBroker());
connFuture.onSuccess(ack -> log.info("connect emqx broker success"))
.onFailure(e -> log.error("connect emqx broker failed", e));
client = MqttClient.create(vertx, options);
List<String> topics = mqttConfig.getSubscribeTopics();
Map<String, Integer> subscribes = new HashMap<>();
for (String topic : topics) {
subscribes.put(topic, 1);
}
client.publishHandler(s -> {
String topic = s.topicName();
String payload = s.payload().toString();
log.info("receive message,topic:{},payload:{}", topic, payload);
//
// //取消订阅
// if (topic.equals("/sys/session/topic/unsubscribed")) {
// topicUnsubscribed(payload);
// return;
// }
//
// //连接断开
// if (topic.equals("/sys/client/disconnected")) {
// disconnectedHandler.handler(payload);
// return;
// }
//
// String[] parts = topic.split("/");
// if (parts.length < 5) {
// log.error("message topic is illegal.");
// return;
// }
// String productKey = parts[2];
// String deviceName = parts[3];
//
// //子设备注册
// if (topic.endsWith("/register")) {
/*subscribes.put("/sys/+/+/s/#", 1);
subscribes.put("/sys/client/connected", 1);
subscribes.put("/sys/client/disconnected", 1);
subscribes.put("/sys/session/subscribed", 1);
subscribes.put("/sys/session/unsubscribed", 1);*/
// handler will be called when we have a message in topic we subscribe for
client.publishHandler(p -> {
log.info("Client received message on [{}] payload [{}] with QoS [{}]", p.topicName(), p.payload().toString(Charset.defaultCharset()), p.qosLevel());
String topic = p.topicName();
String payload = p.payload().toString();
try {
IMessageHandler messageHandler = getHandler();
if (messageHandler != null) {
Map<String, Object> head = new HashMap<>();
head.put("topic", topic);
getHandler().onReceive(head, "", payload);
}).subscribe(subscribes).onSuccess(a -> log.info("subscribe topic success"))
.onFailure(e -> log.error("subscribe topic failed", e));
messageHandler.onReceive(head, "", payload);
}
} catch (Exception e) {
log.error("message is illegal.", e);
}
});
client.connect(mqttConfig.getPort(), mqttConfig.getBroker(), s -> {
if (s.succeeded()) {
log.info("client connect success.");
client.subscribe(subscribes, e -> {
if (e.succeeded()) {
log.info("===>subscribe success: {}", e.result());
} else {
log.error("===>subscribe fail: ", e.cause());
}
});
} else {
log.error("client connect fail: ", s.cause());
}
}).exceptionHandler(event -> {
log.error("client fail: ", event.getCause());
});
} catch (Throwable e) {
throw new BizException("start emqx auth component error", e);
@ -122,6 +148,9 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
authVerticle.stop();
Future<Void> future = vertx.undeploy(deployedId);
future.onSuccess(unused -> log.info("stop emqx auth component success"));
client.disconnect()
.onSuccess(unused -> log.info("stop emqx component success"))
.onFailure(unused -> log.info("stop emqx component failure"));
}
@Override
@ -131,17 +160,85 @@ public class EmqxDeviceComponent extends AbstractDeviceComponent {
@Override
public void onDeviceStateChange(DeviceState state) {
DeviceState.Parent parent = state.getParent();
if (parent == null) {
return;
}
DeviceRepository deviceRepository = SpringUtils.getBean(DeviceRepository.class);
DeviceInfo deviceInfo = deviceRepository.findByProductKeyAndDeviceName(state.getProductKey(), state.getDeviceName());
if (deviceInfo != null) {
boolean isOnline = DeviceState.STATE_ONLINE.equals(state.getState());
deviceInfo.getState().setOnline(isOnline);
if (!isOnline) {
deviceInfo.getState().setOfflineTime(System.currentTimeMillis());
}
if (isOnline) {
deviceInfo.getState().setOnlineTime(System.currentTimeMillis());
}
deviceRepository.save(deviceInfo);
}
}
@Override
public void send(DeviceMessage message) {
Object obj = message.getContent();
if (!(obj instanceof Map)) {
throw new BizException("message content is not Map");
}
Message msg = new Message();
try {
//obj中的key,如果bean中有这个属性，就把这个key对应的value值赋给msg的属性
BeanUtils.populate(msg, (Map<String, ? extends Object>) obj);
} catch (Throwable e) {
throw new BizException("message content is incorrect");
}
log.info("publish topic:{},payload:{}", msg.getTopic(), msg.getPayload());
client.publish(msg.getTopic(),
Buffer.buffer(msg.getPayload()),
MqttQoS.AT_LEAST_ONCE,
false,
false);
}
@Override
public boolean exist(String productKey, String deviceName) {
return false;
/**
* 透传解码
*/
public ThingModelMessage transparentDecode(Map<String, Object> msg) throws InvocationTargetException, IllegalAccessException {
TransparentMsg transparentMsg = new TransparentMsg();
BeanUtils.populate(transparentMsg, msg);
return transparentConverter.decode(transparentMsg);
}
/**
* 透传编码
*/
public DeviceMessage transparentEncode(ThingService<?> service, cc.iotkit.converter.Device device) {
return transparentConverter.encode(service, device);
}
/**
* 提供js调用
*/
public Object getCompMqttClientIdList() {
String[] result = compMqttClientIdList.toArray(new String[0]);
return JsonUtil.toJsonString(result);
}
@Data
public static class Message {
private String topic;
private String payload;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public static class Device {
private String productKey;
private String deviceName;
}
}

19
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/IScripter.java Normal file
View File

@ -0,0 +1,19 @@
package cc.iotkit.comp.emqx;
import cc.iotkit.converter.ThingService;
import cc.iotkit.model.device.message.ThingModelMessage;
public interface IScripter {
void setScript(String script);
/**
* 透传解码
*/
ThingModelMessage decode(TransparentMsg msg);
/**
* 透传编码
*/
TransparentMsg encode(ThingService<?> service);
}

19
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/JsScripter.java Normal file
View File

@ -0,0 +1,19 @@
package cc.iotkit.comp.emqx;
import cc.iotkit.converter.ThingService;
import cc.iotkit.model.device.message.ThingModelMessage;
public class JsScripter implements IScripter {
@Override
public void setScript(String script) {
}
public ThingModelMessage decode(TransparentMsg msg) {
return null;
}
public TransparentMsg encode(ThingService<?> service) {
return null;
}
}

126
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/LuaScripter.java Normal file
View File

@ -0,0 +1,126 @@
package cc.iotkit.comp.emqx;
import cc.iotkit.converter.ThingService;
import cc.iotkit.model.device.message.ThingModelMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.luaj.vm2.LuaTable;
import org.luaj.vm2.LuaValue;
import org.luaj.vm2.script.LuaScriptEngine;
import javax.script.Compilable;
import javax.script.CompiledScript;
import javax.script.ScriptEngineManager;
import javax.script.SimpleBindings;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Slf4j
public class LuaScripter implements IScripter {
private final LuaScriptEngine engine = (LuaScriptEngine) (
new ScriptEngineManager().getEngineByName("luaj"));
private LuaValue decoder;
private LuaValue encoder;
@Override
public void setScript(String script) {
try {
CompiledScript compiledScript = ((Compilable) engine).compile(script);
SimpleBindings bindings = new SimpleBindings();
compiledScript.eval(bindings);
decoder = (LuaValue) bindings.get("decode");
encoder = (LuaValue) bindings.get("encode");
} catch (Throwable e) {
log.error("compile script error", e);
}
}
public ThingModelMessage decode(TransparentMsg msg) {
try {
LuaTable table = new LuaTable();
table.set("model", msg.getModel());
table.set("mac", msg.getMac());
table.set("data", msg.getData());
Map result = (Map) parse(decoder.call(table));
ThingModelMessage modelMessage = new ThingModelMessage();
BeanUtils.populate(modelMessage, result);
modelMessage.setProductKey(msg.getProductKey());
modelMessage.setDeviceName(msg.getMac());
return modelMessage;
} catch (Throwable e) {
log.error("execute decode script error", e);
}
return null;
}
public TransparentMsg encode(ThingService<?> service) {
try {
LuaTable table = new LuaTable();
table.set("identifier", service.getIdentifier());
table.set("type", service.getType());
table.set("productKey", service.getProductKey());
table.set("deviceName", service.getDeviceName());
table.set("mid", service.getMid());
Object params = service.getParams();
LuaTable tableParams = new LuaTable();
if (params instanceof Map) {
((Map<?, ?>) params).forEach((key, val) -> tableParams.set(key.toString(), parse(val)));
}
table.set("params", tableParams);
LuaValue result = encoder.call(table);
Map map = (Map) parse(result);
TransparentMsg message = new TransparentMsg();
BeanUtils.populate(message, map);
return message;
} catch (Throwable e) {
log.error("execute encode script error", e);
}
return null;
}
private Object parse(LuaValue value) {
String type = value.typename();
switch (type) {
case "string":
return value.toString();
case "number":
case "int":
return value.toint();
case "table":
Map<String, Object> data = new HashMap<>();
LuaTable table = (LuaTable) value;
int arrLen = table.rawlen();
if (arrLen > 0) {
//数组转换
List<Object> list = new ArrayList<>();
for (LuaValue key : table.keys()) {
list.add(parse(table.get(key)));
}
return list;
} else {
//map转换
for (LuaValue key : table.keys()) {
data.put(key.toString(), parse(table.get(key)));
}
}
return data;
}
return null;
}
private LuaValue parse(Object value) {
if (value instanceof String) {
return LuaValue.valueOf(value.toString());
}
if (value instanceof Integer) {
return LuaValue.valueOf((Integer) value);
}
return new LuaTable();
}
}

88
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/TransparentConverter.java Normal file
View File

@ -0,0 +1,88 @@
package cc.iotkit.comp.emqx;
import cc.iotkit.converter.Device;
import cc.iotkit.converter.DeviceMessage;
import cc.iotkit.converter.ThingService;
import cc.iotkit.dao.DeviceCache;
import cc.iotkit.dao.ProductCache;
import cc.iotkit.model.device.DeviceInfo;
import cc.iotkit.model.device.message.ThingModelMessage;
import cc.iotkit.model.product.ProductModel;
import lombok.extern.slf4j.Slf4j;
import java.util.HashMap;
import java.util.Map;
@Slf4j
public class TransparentConverter {
private final Map<String, IScripter> scripters = new HashMap<>();
private final Map<String, String> scripts = new HashMap<>();
/**
* 透传解码
*/
public ThingModelMessage decode(TransparentMsg msg) {
//通过上报消息中的model取得对应的产品
String productKey = checkScriptUpdate(msg.getModel());
msg.setProductKey(productKey);
return scripters.get(productKey).decode(msg);
}
/**
* 透传编码
*/
public DeviceMessage encode(ThingService<?> service, Device device) {
String productKey = service.getProductKey();
checkScriptUpdate(device.getModel());
TransparentMsg transparentMsg = scripters.get(productKey).encode(service);
//转换成网关消息
String deviceName = service.getDeviceName();
DeviceInfo gateway = getGatewayInfo(productKey, deviceName);
DeviceMessage message = new DeviceMessage();
message.setProductKey(gateway.getProductKey());
message.setDeviceName(gateway.getDeviceName());
message.setMid(transparentMsg.getMid());
//透传格式消息内容,mac、model、data
message.setContent(transparentMsg);
return message;
}
private ProductModel getScript(String model) {
return ProductCache.getInstance().getProductScriptByModel(model);
}
private DeviceInfo getGatewayInfo(String subPk, String subDn) {
String parentId = DeviceCache.getInstance().getDeviceInfo(subPk, subDn).getParentId();
return DeviceCache.getInstance().get(parentId);
}
/**
* 检查产品脚本是否更新
*/
private String checkScriptUpdate(String model) {
ProductModel productModel = getScript(model);
String productKey = productModel.getProductKey();
String script = productModel.getScript();
String oldScript = scripts.get(productKey);
if (script.equals(oldScript)) {
return productKey;
}
String type = productModel.getType();
if (ProductModel.TYPE_LUA.equals(type)) {
scripters.putIfAbsent(productKey, new LuaScripter());
} else if (ProductModel.TYPE_JS.equals(type)) {
scripters.putIfAbsent(productKey, new JsScripter());
}
//更新脚本
IScripter scripter = scripters.get(productKey);
scripter.setScript(script);
scripts.put(productKey, script);
return productKey;
}
}

21
protocol-gateway/emqx-component/src/main/java/cc/iotkit/comp/emqx/TransparentMsg.java Normal file
View File

@ -0,0 +1,21 @@
package cc.iotkit.comp.emqx;
import lombok.Data;
@Data
public class TransparentMsg {
private String productKey;
/**
* 生成给设备端的消息id
*/
private String mid;
private String model;
private String mac;
private String data;
}

340
protocol-gateway/emqx-component/src/main/resources/component.js Normal file
View File

@ -0,0 +1,340 @@
!function(n){"use strict";function d(n,t){var r=(65535&n)+(65535&t);return(n>>16)+(t>>16)+(r>>16)<<16|65535&r}function f(n,t,r,e,o,u){return d((u=d(d(t,n),d(e,u)))<<o|u>>>32-o,r)}function l(n,t,r,e,o,u,c){return f(t&r|~t&e,n,t,o,u,c)}function g(n,t,r,e,o,u,c){return f(t&e|r&~e,n,t,o,u,c)}function v(n,t,r,e,o,u,c){return f(t^r^e,n,t,o,u,c)}function m(n,t,r,e,o,u,c){return f(r^(t|~e),n,t,o,u,c)}function c(n,t){var r,e,o,u;n[t>>5]|=128<<t%32,n[14+(t+64>>>9<<4)]=t;for(var c=1732584193,f=-271733879,i=-1732584194,a=271733878,h=0;h<n.length;h+=16)c=l(r=c,e=f,o=i,u=a,n[h],7,-680876936),a=l(a,c,f,i,n[h+1],12,-389564586),i=l(i,a,c,f,n[h+2],17,606105819),f=l(f,i,a,c,n[h+3],22,-1044525330),c=l(c,f,i,a,n[h+4],7,-176418897),a=l(a,c,f,i,n[h+5],12,1200080426),i=l(i,a,c,f,n[h+6],17,-1473231341),f=l(f,i,a,c,n[h+7],22,-45705983),c=l(c,f,i,a,n[h+8],7,1770035416),a=l(a,c,f,i,n[h+9],12,-1958414417),i=l(i,a,c,f,n[h+10],17,-42063),f=l(f,i,a,c,n[h+11],22,-1990404162),c=l(c,f,i,a,n[h+12],7,1804603682),a=l(a,c,f,i,n[h+13],12,-40341101),i=l(i,a,c,f,n[h+14],17,-1502002290),c=g(c,f=l(f,i,a,c,n[h+15],22,1236535329),i,a,n[h+1],5,-165796510),a=g(a,c,f,i,n[h+6],9,-1069501632),i=g(i,a,c,f,n[h+11],14,643717713),f=g(f,i,a,c,n[h],20,-373897302),c=g(c,f,i,a,n[h+5],5,-701558691),a=g(a,c,f,i,n[h+10],9,38016083),i=g(i,a,c,f,n[h+15],14,-660478335),f=g(f,i,a,c,n[h+4],20,-405537848),c=g(c,f,i,a,n[h+9],5,568446438),a=g(a,c,f,i,n[h+14],9,-1019803690),i=g(i,a,c,f,n[h+3],14,-187363961),f=g(f,i,a,c,n[h+8],20,1163531501),c=g(c,f,i,a,n[h+13],5,-1444681467),a=g(a,c,f,i,n[h+2],9,-51403784),i=g(i,a,c,f,n[h+7],14,1735328473),c=v(c,f=g(f,i,a,c,n[h+12],20,-1926607734),i,a,n[h+5],4,-378558),a=v(a,c,f,i,n[h+8],11,-2022574463),i=v(i,a,c,f,n[h+11],16,1839030562),f=v(f,i,a,c,n[h+14],23,-35309556),c=v(c,f,i,a,n[h+1],4,-1530992060),a=v(a,c,f,i,n[h+4],11,1272893353),i=v(i,a,c,f,n[h+7],16,-155497632),f=v(f,i,a,c,n[h+10],23,-1094730640),c=v(c,f,i,a,n[h+13],4,681279174),a=v(a,c,f,i,n[h],11,-358537222),i=v(i,a,c,f,n[h+3],16,-722521979),f=v(f,i,a,c,n[h+6],23,76029189),c=v(c,f,i,a,n[h+9],4,-640364487),a=v(a,c,f,i,n[h+12],11,-421815835),i=v(i,a,c,f,n[h+15],16,530742520),c=m(c,f=v(f,i,a,c,n[h+2],23,-995338651),i,a,n[h],6,-198630844),a=m(a,c,f,i,n[h+7],10,1126891415),i=m(i,a,c,f,n[h+14],15,-1416354905),f=m(f,i,a,c,n[h+5],21,-57434055),c=m(c,f,i,a,n[h+12],6,1700485571),a=m(a,c,f,i,n[h+3],10,-1894986606),i=m(i,a,c,f,n[h+10],15,-1051523),f=m(f,i,a,c,n[h+1],21,-2054922799),c=m(c,f,i,a,n[h+8],6,1873313359),a=m(a,c,f,i,n[h+15],10,-30611744),i=m(i,a,c,f,n[h+6],15,-1560198380),f=m(f,i,a,c,n[h+13],21,1309151649),c=m(c,f,i,a,n[h+4],6,-145523070),a=m(a,c,f,i,n[h+11],10,-1120210379),i=m(i,a,c,f,n[h+2],15,718787259),f=m(f,i,a,c,n[h+9],21,-343485551),c=d(c,r),f=d(f,e),i=d(i,o),a=d(a,u);return[c,f,i,a]}function i(n){for(var t="",r=32*n.length,e=0;e<r;e+=8)t+=String.fromCharCode(n[e>>5]>>>e%32&255);return t}function a(n){var t=[];for(t[(n.length>>2)-1]=void 0,e=0;e<t.length;e+=1)t[e]=0;for(var r=8*n.length,e=0;e<r;e+=8)t[e>>5]|=(255&n.charCodeAt(e/8))<<e%32;return t}function e(n){for(var t,r="0123456789abcdef",e="",o=0;o<n.length;o+=1)t=n.charCodeAt(o),e+=r.charAt(t>>>4&15)+r.charAt(15&t);return e}function r(n){return unescape(encodeURIComponent(n))}function o(n){return i(c(a(n=r(n)),8*n.length))}function u(n,t){return function(n,t){var r,e=a(n),o=[],u=[];for(o[15]=u[15]=void 0,16<e.length&&(e=c(e,8*n.length)),r=0;r<16;r+=1)o[r]=909522486^e[r],u[r]=1549556828^e[r];return t=c(o.concat(a(t)),512+8*t.length),i(c(u.concat(t),640))}(r(n),r(t))}function t(n,t,r){return t?r?u(t,n):e(u(t,n)):r?o(n):e(o(n))}"function"==typeof define&&define.amd?define(function(){return t}):"object"==typeof module&&module.exports?module.exports=t:n.md5=t}(this);
var md5=this.md5;
function isServerId(clientId){
return JSON.parse(component.getCompMqttClientIdList()).indexOf(clientId) > -1
}
function getPkDn(clientId){
var arr=clientId.split("_");
return {
pk:arr[0],
dn:arr[1]
};
}
function auth(head,type,payload){
if(isServerId(payload.clientid) > -1){
return {
type:"serverAuth",
data:{
productKey:"pd",
deviceName:"dn",
}
};
}
var arr= payload.clientid.split("_");
if(arr.length<3){
throw new Error("incorrect clientid：" + payload.clientid);
}
var pk=arr[0];
var dn=arr[1];
var model=arr[2];
var pwd=md5("xdkKUymrEGSCYWswqCvSPyRSFvH5j7CU"+payload.clientid);
if(pwd!=payload.password){
throw new Error("incorrect password：" + pwd);
}
return {
type:"register",
data:{
productKey:pk,
deviceName:dn,
model:model
}
};
}
function acl(head,type,payload){
var _topic = payload.topic;
if(isServerId(payload.clientid)){
return {
type: "acl",
data:{
productKey:"pd",
deviceName:"dn",
}
};
}
/**
* /sys/N523nWsCiG3CAn6X/AA:BB:CC:EE:01/c/#
* /sys/N523nWsCiG3CAn6X/AA:BB:CC:EE:01/c/register_reply
*/
if (/^\/sys\/.+\/.+\/c\/#?$/i.test(_topic)) {
if (payload.access == 1) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/c\/register_reply$/i.test(_topic)) {
if (payload.access == 1) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/s\/register$/i.test(_topic)) {
if (payload.access == 2) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/s\/service\/allowJoin_reply$/i.test(_topic)) {
if (payload.access == 2) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/s\/service\/rawSend_reply$/i.test(_topic)) {
if (payload.access == 2) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/s\/event\/property\/post$/i.test(_topic)) {
if (payload.access == 2) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
if (/^\/sys\/.+\/.+\/s\/service\/.+_reply$/i.test(_topic)) {
if (payload.access == 2) { // 1、subscribe 2、public
return subscribe(head,type,payload);
}
}
}
function register(head,type,payload){
var auth= payload;
var arr= auth.clientid.split("_");
if(arr.length<3){
throw new Error("incorrect clientid：" + auth.clientid);
}
var pk=arr[0];
var dn=arr[1];
var model=arr[2];
var pwd=md5("xdkKUymrEGSCYWswqCvSPyRSFvH5j7CU"+auth.clientid);
if(pwd!=auth.password){
throw new Error("incorrect password：" + pwd);
}
return {
type:"register",
data:{
productKey:pk,
deviceName:dn,
model:model
}
};
}
function subRegister(topic,parent, payload){
if(!topic){
throw new Error("topic is blank")
}
var params= payload.params;
var reply=
{
productKey: parent.productKey,
deviceName: parent.deviceName,
mid: "0",
content:{
topic: topic.replace("/s/","/c/")+"_reply",
payload: JSON.stringify({
id: payload.id,
code: 0,
data: {
"productKey":params.productKey,
"deviceName":params.deviceName
}
})
}
};
return {
type:"register",
data:{
productKey:parent.productKey,
deviceName:parent.deviceName,
subDevices:[{
productKey:params.productKey,
deviceName:params.deviceName,
model:params.model
}]
},
action:{
type:"ack",
content:JSON.stringify(reply)
}
};
}
function disconnect(head,type,payload){
var clientId = payload.clientid
var device=getPkDn(clientId);
return {
type:"state",
data:{
productKey:device.pk,
deviceName:device.dn,
state:"offline"
}
}
}
function connect(head,type,payload) {
var clientId = payload.clientid
var device = getPkDn(clientId);
return {
type: "state",
data: {
productKey: device.pk,
deviceName: device.dn,
state: "online"
}
}
}
function unsubscribe(head,type,payload){
var topic = payload.topic;
if(isServerId(payload.clientid)){
return {
type: "acl",
data:{
productKey:"pd",
deviceName:"dn",
}
};
}
var arr= topic.split('/');
if(arr.length<6){
throw new Error("incorrect topic: "+topic)
}
var pk=arr[2];
var dn=arr[3];
return {
type:"state",
data:{
productKey: pk,
deviceName: dn,
state:"offline"
}
}
}
function subscribe(head,type,payload){
var topic = payload.topic;
if(isServerId(payload.clientid)){
return {
type: "acl",
data:{
productKey:"pd",
deviceName:"dn",
}
};
}
var arr= topic.split('/');
if(arr.length<6){
throw new Error("incorrect topic: "+topic)
}
var pk=arr[2];
var dn=arr[3];
return {
type: "state",
data: {
productKey: pk,
deviceName: dn,
state: "online"
}
}
}
var messageHandler = {
"/sys/client/connected":connect,
"/sys/client/disconnected":disconnect,
"/mqtt/auth":auth,
"/mqtt/acl":acl,
"/sys/session/subscribed":subscribe,
"/sys/session/unsubscribed":unsubscribe
}
//必须提供onReceive方法
this.onReceive=function(head,type,payload){
payload=JSON.parse(payload);
print("======================================================================= ");
print("【message from】: " + (isServerId(payload.clientid)?"Server":"Device") );
print("onReceive head: "+JSON.stringify(head));
print("onReceive type: "+JSON.stringify(type));
print("onReceive payload: "+ JSON.stringify(payload));
//print("onReceive compMqttClientIdList: "+ component.getCompMqttClientIdList());
var result = {};
var topic = head.topic;
if(!topic) {
print("【result】: " + JSON.stringify(result));
print("======================================================================= ");
return result;
}
var fun = messageHandler[topic];
if(fun){
result = fun(head,type,payload)
}
else{
var arr= topic.split('/');
if(arr.length<6){
throw new Error("incorrect topic: "+topic)
}
var pk=arr[2];
var dn=arr[3];
//子设备注册
if(topic.endsWith('/register')){
result = subRegister(topic,{productKey:pk,deviceName:dn}, payload);
}
else {
//数据上报
result = {
type: "report",
data: {
productKey: pk,
deviceName: dn,
mid: payload.id,
content: {
topic: topic,
payload: payload
}
}
}
}
}
print("【result】: " + JSON.stringify(result));
print("======================================================================= ");
return result;
}
this.onRegistered=function(regInfo,result){
}

8
protocol-gateway/http-biz-component/src/main/java/cc/iotkit/comp/biz/HttpBizComponent.java
View File

@ -18,10 +18,7 @@ import lombok.extern.slf4j.Slf4j;
import javax.script.ScriptEngineManager;
import javax.script.ScriptException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
@Data
@Slf4j
@ -41,8 +38,11 @@ public class HttpBizComponent implements IComponent {
private HttpServer backendServer;
private String id;
@Override
public void create(CompConfig config) {
this.id = UUID.randomUUID().toString();
this.httpConfig = JsonUtil.parse(config.getOther(), HttpConfig.class);
try {
scriptObj = engine.eval(String.format("new (function () {\n%s})()", script));

13
protocol-gateway/mqtt-component/src/main/java/cc/iotkit/comp/mqtt/MqttDeviceComponent.java
View File

@ -27,7 +27,7 @@ public class MqttDeviceComponent extends AbstractDeviceComponent {
private String deployedId;
private MqttVerticle mqttVerticle;
private final Map<String, Device> deviceChildToParent = new HashMap<>();
private TransparentConverter transparentConverter = new TransparentConverter();
private final TransparentConverter transparentConverter = new TransparentConverter();
public void create(CompConfig config) {
super.create(config);
@ -110,17 +110,6 @@ public class MqttDeviceComponent extends AbstractDeviceComponent {
msg.getTopic(), msg.getPayload());
}
@Override
public boolean exist(String productKey, String deviceName) {
//先作为子设备查找是否存在父设备
Device device = deviceChildToParent.get(new Device(productKey, deviceName).toString());
if (device != null) {
return true;
}
return mqttVerticle.exist(productKey, deviceName);
}
@Override
public CompConfig getConfig() {
return config;