This commit is contained in:
pnoker
2024-05-26 23:38:09 +08:00
parent 09283a57d8
commit 939b5a1edc
52 changed files with 0 additions and 4842 deletions
-48
View File
@@ -1,48 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
# runtime
FROM registry.cn-beijing.aliyuncs.com/dc3/dc3-jre:17 AS runtime
LABEL dc3.author pnokers
LABEL dc3.author.email pnokers.icloud.com
# default env
ENV PARAMS ''
ENV NODE_ENV dev
ENV SERVER_PORT 8600
ENV GRPC_PORT 9600
# default jvm & gc env
ENV JAVA_OPS -server \
-Xms256m \
-Xmx1024m \
-Djava.security.egd=file:/dev/./urandom \
-XX:+UseG1GC
ENV GC_LOG -Xlog:gc*:file=dc3/logs/center/ekuiper/gc/gc.log:time,uptime,level,tags:filecount=20,filesize=10M
RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
WORKDIR /dc3-center/dc3-center-ekuiper
ADD ./target/dc3-center-ekuiper.jar ./
EXPOSE ${SERVER_PORT}
EXPOSE ${GRPC_PORT}
VOLUME /dc3-center/dc3-center-ekuiper/dc3/logs
CMD mkdir -p /dc3-center/dc3-center-ekuiper/dc3/logs/center/ekuiper/gc \
&& java ${JAVA_OPS} ${GC_LOG} -jar dc3-center-ekuiper.jar ${PARAMS}
-85
View File
@@ -1,85 +0,0 @@
<!--
~ Copyright 2016-present the IoT DC3 original author or authors.
~
~ 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
~
~ https://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:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
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>io.github.pnoker</groupId>
<artifactId>dc3-center</artifactId>
<version>2024.3.0</version>
</parent>
<artifactId>dc3-center-ekuiper</artifactId>
<packaging>jar</packaging>
<description>IOT DC3 平台 规则引擎服务</description>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-common-mysql</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- webClient -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- fastjson -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.54</version>
</dependency>
<dependency>
<groupId>org.glassfish</groupId>
<artifactId>javax.el</artifactId>
<version>3.0.1-b11</version>
</dependency>
<!-- Hutool -->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.8.3</version>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-common-repository</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
@@ -1,35 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author : Zhen
*/
@SpringBootApplication
public class EkuiperApplication {
private static final Logger log = LoggerFactory.getLogger(EkuiperApplication.class);
public static void main(String[] args) {
SpringApplication.run(EkuiperApplication.class, args);
log.info("Ekuiper Application Start...");
}
}
@@ -1,47 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.entity.dto.SqlConfigDto;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClient;
/**
* @author : Zhen
*/
@Configuration
public class EkuiperConfig {
@Bean
public WebClient webClient() {
return WebClient.create();
}
@Bean
public ObjectMapper objectMapper() {
return new ObjectMapper();
}
@Bean
public SqlConfigDto sqlConfigDto() {
return new SqlConfigDto();
}
}
@@ -1,108 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.constant;
/**
* 缓存相关
*
* @author pnoker
*/
public interface CacheConstant {
/**
* 超时相关
*/
interface Timeout {
/**
* salt 在 redis 中的失效时间, 分钟
*/
int SALT_CACHE_TIMEOUT = 5;
/**
* user 登陆限制失效时间, 分钟
*/
int USER_LIMIT_TIMEOUT = 5;
/**
* token 在 redis 中的失效时间, 小时
*/
int TOKEN_CACHE_TIMEOUT = 12;
}
/**
* 实体相关
*/
interface Entity {
String TENANT = "tenant";
String TENANT_BIND = "tenant_bind";
String BLACK_IP = "black_ip";
String RTMP = "rtmp";
String USER = "user";
String DRIVER = "driver";
String DRIVER_INFO = "driver_info";
String DRIVER_ATTRIBUTE = "driver_attribute";
String PROFILE = "profile";
String PROFILE_BIND = "profile_bind";
String GROUP = "group";
String DEVICE = "device";
String POINT = "point";
String POINT_ATTRIBUTE = "point_attribute";
String POINT_INFO = "point_info";
String LABEL = "label";
String LABEL_BIND = "label_bind";
}
/**
* 前缀相关
*/
interface Prefix {
String REAL_TIME_VALUE_KEY_PREFIX = Entity.POINT + Suffix.VALUE + CommonConstant.Symbol.SEPARATOR;
String DRIVER_STATUS_KEY_PREFIX = Entity.DRIVER + Suffix.STATUS + CommonConstant.Symbol.SEPARATOR;
String DEVICE_STATUS_KEY_PREFIX = Entity.DEVICE + Suffix.STATUS + CommonConstant.Symbol.SEPARATOR;
}
/**
* 后缀相关
*/
interface Suffix {
String ID = "_id";
String TENANT_ID = "_tenant_id";
String USER_ID = "_user_id";
String DRIVER_ID = "_driver_id";
String GROUP_ID = "_group_id";
String PROFILE_ID = "_profile_id";
String POINT_ID = "_point_id";
String DEVICE_ID = "_device_id";
String ATTRIBUTE_ID = "_attribute_id";
String ENTITY_ID = "_entity_id";
String NAME = "_name";
String PHONE = "_phone";
String EMAIL = "_email";
String SALT = "_salt";
String TOKEN = "_token";
String LIMIT = "_limit";
String VALUE = "_value";
String STATUS = "_status";
String TYPE = "_type";
String UNIT = "_unit";
String IP = "_ip";
String HOST_PORT = "_host_port";
String SERVICE_NAME = "_service_name";
String LIST = "_list";
String DIC = "_dic";
}
}
@@ -1,324 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.constant;
/**
* dc3平台常量
*
* @author pnoker
*/
public interface CommonConstant {
/**
* 符号相关
*/
interface Symbol {
/**
* 点
*/
String DOT = ".";
/**
* 下划线
*/
String UNDERSCORE = "_";
/**
* 星号
*/
String ASTERISK = "*";
/**
* 井号
*/
String HASHTAG = "#";
/**
* 分隔符
*/
String SEPARATOR = "::";
/**
* 斜线
*/
String SLASH = "/";
}
/**
* 算法相关
*/
interface Algorithm {
/**
* 默认密钥
*/
String DEFAULT_KEY = "io.github.pnoker.dc3";
/**
* 默认密码
*/
String DEFAULT_PASSWORD = "a12345";
/**
* 加密算法 对称AES
*/
String ALGORITHM_AES = "AES";
/**
* 加密算法 非对称RSA
*/
String ALGORITHM_RSA = "RSA";
/**
* 加密算法 SHA256withRSA
*/
String ALGORITHM_SHA256_RSA = "SHA256withRSA";
/**
* 证书类型 X.509
*/
String CERTIFICATE_X509 = "X.509";
/**
* 证书类型 PKCS12
*/
String CERTIFICATE_PKCS12 = "PKCS12";
/**
* 证书类型 JKS
*/
String CERTIFICATE_JKS = "jks";
}
/**
* 时间相关
*/
interface Time {
/**
* 时区
*/
String TIMEZONE = "GMT+8";
/**
* 时间格式化
*/
String DEFAULT_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
String COMPLETE_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS";
}
/**
* 文件夹相关
*/
interface Folder {
/**
* 用户主目录
*/
String USER_HOME_PATH = System.getProperty("user.home") + "/.dc3/";
/**
* 默认上传文件的缓存位置
*/
String TEMP_FILE_PATH = System.getProperty("java.io.tmpdir") + "/dc3/";
}
/**
* 响应相关
*/
interface Response {
String OK = "ok";
String ERROR = "error";
}
/**
* 消息相关
*/
interface Rabbit {
// Arguments
String MESSAGE_TTL = "x-message-ttl";
String TOPIC_EXCHANGE_EVENT = "dc3.exchange.event";
String ROUTING_DRIVER_EVENT_PREFIX = "dc3.routing.event.driver.";
String QUEUE_DRIVER_EVENT = "dc3.queue.event.driver";
// Event
String ROUTING_DEVICE_EVENT_PREFIX = "dc3.routing.event.device.";
String QUEUE_DEVICE_EVENT = "dc3.queue.event.device";
// Metadata
String TOPIC_EXCHANGE_METADATA = "dc3.exchange.metadata";
String ROUTING_DRIVER_METADATA_PREFIX = "dc3.routing.metadata.driver.";
String QUEUE_DRIVER_METADATA_PREFIX = "dc3.queue.metadata.driver.";
// Value
String TOPIC_EXCHANGE_VALUE = "dc3.exchange.value";
String ROUTING_POINT_VALUE_PREFIX = "dc3.routing.value.point.";
String QUEUE_POINT_VALUE = "dc3.queue.value.point";
}
/**
* 存储相关
*/
interface Storage {
/**
* 设备数据存储集合前缀
*/
String POINT_VALUE_PREFIX = CacheConstant.Entity.POINT + io.github.ponker.center.ekuiper.constant.CacheConstant.Suffix.VALUE + Symbol.UNDERSCORE;
}
/**
* 状态相关
*/
interface Status {
/**
* 注册状态相关
*/
String REGISTERING = "REGISTERING";
String UNREGISTERED = "UNREGISTERED";
/**
* 运行状态相关
*/
String ONLINE = "ONLINE";
String OFFLINE = "OFFLINE";
String MAINTAIN = "MAINTAIN";
String FAULT = "FAULT";
}
/**
* 驱动相关
*/
interface Driver {
int DEFAULT_MAX_REQUEST_SIZE = 100;
/**
* 事件相关
*/
interface Event {
/**
* 驱动注册握手事件, 该事件用于校验当前 dc3-center-manager 是否可用
*/
String DRIVER_HANDSHAKE = "driver_handshake";
String DRIVER_HANDSHAKE_BACK = "driver_handshake_back";
/**
* 驱动注册事件, 该事件用于向 dc3-center-manager 注册驱动配置信息
*/
String DRIVER_REGISTER = "driver_register";
String DRIVER_REGISTER_BACK = "driver_register_back";
/**
* 同步驱动元数据时间, 该事件用于向 dc3-center-manager 发送驱动元数据同步请求
*/
String DRIVER_METADATA_SYNC = "driver_metadata_sync";
String DRIVER_METADATA_SYNC_BACK = "driver_metadata_sync_back";
/**
* 驱动心跳事件, 该事件用于向 dc3-center-manager 发送驱动的当前状态
*/
String DRIVER_HEARTBEAT = "driver_heartbeat";
String ERROR = "error";
}
/**
* 类型
*/
interface Type {
String DRIVER = "driver";
String GATEWAY = "gateway";
String PROFILE = "profile";
String DEVICE = "device";
String POINT = "point";
String DRIVER_INFO = "driver_info";
String POINT_INFO = "point_info";
}
interface Profile {
String ADD = "add_profile";
String DELETE = "delete_profile";
String UPDATE = "update_profile";
}
interface Device {
String ADD = "add_device";
String DELETE = "delete_device";
String UPDATE = "update_device";
}
interface Point {
String ADD = "add_point";
String DELETE = "delete_point";
String UPDATE = "update_point";
}
interface DriverConfig {
String ADD = "add_driver_info";
String DELETE = "delete_driver_info";
String UPDATE = "update_driver_info";
}
interface PointConfig {
String ADD = "add_point_info";
String DELETE = "delete_point_info";
String UPDATE = "update_point_info";
}
}
/**
* 设备相关
*/
interface Device {
/**
* 设备事件
*/
interface Event {
/**
* 设备心跳事件
*/
String HEARTBEAT = "heartbeat";
/**
* 超出上限事件
*/
String OVER_UPPER_LIMIT = "over_upper_limit";
/**
* 超出下限事件
*/
String OVER_LOWER_LIMIT = "over_lower_limit";
/**
* 用于记录错误事件类型
*/
String ERROR = "error";
}
}
/**
* 策略工厂相关
*/
interface RepositoryStrategy {
String REPOSITORY_STRATEGY = "repository" + Symbol.SEPARATOR;
String REPOSITORY_STRATEGY_REDIS = "redis";
String REPOSITORY_STRATEGY_MONGO = "mongo";
String REPOSITORY_STRATEGY_INFLUXDB = "influxdb";
String REPOSITORY_STRATEGY_OPENTSDB = "opentsdb";
String REPOSITORY_STRATEGY_ELASTICSEARCH = "elasticsearch";
String REPOSITORY_STRATEGY_MQTT = "mqtt";
}
}
@@ -1,48 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.constant;
/**
* @author : Zhen
* <p>
* 规则引擎相关
*/
public interface EkuiperConstant {
/**
* 规则引擎url相关
*/
interface EKUIPERURL {
String STREAM_URL = "http://localhost:9081/streams";
String TABLE_URL = "http://localhost:9081/tables";
String CONFKEY_URL = "http://localhost:9081/metadata/sources";
String RULE_URL = "http://localhost:9081/rules";
}
interface ConKeyType {
String MQTT = "mqtt";
String REDIS = "redis";
String HTTPPULL = "httppull";
String HTTPPUSH = "httppush";
String SQL = "sql";
String REST = "rest";
}
}
@@ -1,38 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.constant;
/**
* 服务相关
*
* @author pnoker
*/
public interface ServiceConstant {
/**
* 规则引擎服务相关
*/
interface Ekuiper {
String SERVICE_NAME = "DC3-CENTER-EKUIPER";
String STREAM_URL_PREFIX = "/stream";
String TABLE_URL_PREFIX = "/table";
String RULE_URL_PREFIX = "/rule";
String CONFKEY_URL_PREFIX = "/confkey";
String SINKTEM_URL_PREFIX = "/sinktem";
}
}
@@ -1,151 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.controller;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.constant.ServiceConstant;
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.service.ApiService;
import io.github.ponker.center.ekuiper.service.ConfigService;
import io.github.ponker.center.ekuiper.service.SinkTemplateService;
import io.github.ponker.center.ekuiper.service.UrlService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpMethod;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
/**
* @author : Zhen
*/
@Slf4j
@RestController
@RequestMapping(ServiceConstant.Ekuiper.CONFKEY_URL_PREFIX)
public class ConKeyController {
@Resource
private ApiService apiService;
@Resource
private ConfigService configService;
@Resource
private ObjectMapper objectMapper;
@Resource
private UrlService urlService;
@Resource
private SinkTemplateService sinkTemplateService;
/**
* 注册, 更新ConfKey
*/
@PutMapping("/create")
public Mono<R<String>> createConf(@Validated @RequestBody Object form, @RequestParam(name = "name") String name, @RequestParam(name = "confKey") String confKey) {
String url = urlService.getConfigUrl() + "/sources/" + name + "/confKeys/" + confKey;
Mono<String> stringMono = configService.callApiConfig(HttpMethod.PUT, url, form, name);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call create ConfKey API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 列出{name}类型的全部Confkey
*/
@GetMapping("/list")
public Mono<R<Map<String, Map<String, Object>>>> listConf(@RequestParam(name = "name") String name) {
String url = urlService.getConfigUrl() + "/sources/yaml/" + name;
Mono<String> stringMono = apiService.callApi(HttpMethod.GET, url);
return stringMono.flatMap(jsonString -> {
try {
TypeReference<Map<String, Map<String, Object>>> typeRef = new TypeReference<Map<String, Map<String, Object>>>() {
};
Map<String, Map<String, Object>> configurations = objectMapper.readValue(jsonString, typeRef);
return Mono.just(R.ok(configurations, "successful"));
} catch (Exception e) {
log.error("Failed to call list ConfKey API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除ConfKey
*/
@DeleteMapping("/delete")
public Mono<R<String>> deleteConf(@RequestParam(name = "name") String name, @RequestParam(name = "confKey") String confKey) {
String url = urlService.getConfigUrl() + "/sources/" + name + "/confKeys/" + confKey;
Mono<String> stringMono = configService.callApiDelConf(HttpMethod.DELETE, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call delete ConfKey API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 测试配置连接
*/
@PostMapping("/connection")
public Mono<R<String>> connectConfig(@Validated @RequestBody Object form, @RequestParam(name = "name") String name) {
String url = urlService.getConfigUrl() + "/sources/connection/" + name;
Mono<String> stringMono = configService.callApiConfig(HttpMethod.POST, url, form, name);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call connect ConfKey API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 显示所有Config和Sinktemplate
*/
@GetMapping("/resources")
public Mono<R<Map<String, List<Map<String, String>>>>> listAllConfig() {
String url = urlService.getConfigUrl() + "/resources";
Mono<String> stringMono = sinkTemplateService.callConfigAndSink(HttpMethod.GET, url);
return stringMono.flatMap(jsonString -> {
try {
TypeReference<Map<String, List<Map<String, String>>>> typeRef = new TypeReference<Map<String, List<Map<String, String>>>>() {
};
Map<String, List<Map<String, String>>> configurations = objectMapper.readValue(jsonString, typeRef);
return Mono.just(R.ok(configurations, "successful"));
} catch (Exception e) {
log.error("Failed to call list All ConfKey API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
}
@@ -1,348 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.controller;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.github.ponker.center.ekuiper.constant.ServiceConstant;
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.entity.vo.DetailRuleVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleDataVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleStatusVO;
import io.github.ponker.center.ekuiper.service.ApiService;
import io.github.ponker.center.ekuiper.service.RuleService;
import io.github.ponker.center.ekuiper.service.UrlService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpMethod;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
/**
* @author : Zhen
*/
@Slf4j
@RestController
@RequestMapping(ServiceConstant.Ekuiper.RULE_URL_PREFIX)
public class RuleController {
@Resource
private ApiService apiService;
@Resource
private RuleService ruleService;
@Resource
private UrlService urlService;
/**
* 新增rule
*/
@PostMapping("/create")
public Mono<R<String>> createRule(@Validated @RequestBody Object form) throws Exception {
Mono<String> stringMono = ruleService.callRuleApiWithData(form, HttpMethod.POST, urlService.getRuleUrl());
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call create Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 更新rule
*/
@PutMapping("/update")
public Mono<R<String>> updateRule(@Validated @RequestBody Object form, @RequestParam(name = "id") String id) throws Exception {
String url = urlService.getRuleUrl() + "/" + id;
Mono<String> stringMono = ruleService.callRuleApiWithData(form, HttpMethod.PUT, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call update Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 展示所有规则
*/
@GetMapping("/list")
public Mono<R<Page<RuleDataVO>>> listAllRule(@RequestParam(required = false, defaultValue = "1") Integer pageNum,
@RequestParam(required = false, defaultValue = "10") Integer pageSize) {
Mono<Page<RuleDataVO>> pageMono = ruleService.callApiWithRulePage(HttpMethod.GET, urlService.getRuleUrl(), pageNum, pageSize);
return pageMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call listAllRule Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 描述当前规则
*/
@GetMapping("/listOne")
public Mono<R<DetailRuleVO>> listOneRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id;
Mono<DetailRuleVO> dynamicEntityMono = ruleService.callApilRule(HttpMethod.GET, url);
return dynamicEntityMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call listOneRule Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 查看规则状态
*/
@GetMapping("/status")
public Mono<R<RuleStatusVO>> lookStatusRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id + "/status";
Mono<RuleStatusVO> ruleStatusMono = ruleService.callApiRuleStatus(HttpMethod.GET, url);
return ruleStatusMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call lookStatusRule Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 启动规则
*/
@PostMapping("/start")
public Mono<R<String>> startRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id + "/" + "start";
Mono<String> stringMono = apiService.callApi(HttpMethod.POST, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call start Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 停止规则
*/
@PostMapping("/stop")
public Mono<R<String>> stopRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id + "/" + "stop";
Mono<String> stringMono = apiService.callApi(HttpMethod.POST, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call stop Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 重启规则
*/
@PostMapping("/restart")
public Mono<R<String>> restartRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id + "/" + "restart";
Mono<String> stringMono = apiService.callApi(HttpMethod.POST, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to restart stop Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除规则
*/
@DeleteMapping("/delete")
public Mono<R<String>> deleteRule(@RequestParam(name = "id") String id) {
String url = urlService.getRuleUrl() + "/" + id;
Mono<String> stringMono = apiService.callApi(HttpMethod.DELETE, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to delete stop Rule API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 添加actionNodes
*/
@PostMapping("/createAction")
public Mono<R<String>> createActions(@RequestBody Object data) {
Mono<String> stringMono = ruleService.addActions(data);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to create Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 显示actionNodes
*/
@GetMapping("/listAction")
public Mono<Object> listActions() {
Mono<List<Map<String, Map<String, Object>>>> listMono = ruleService.listActions();
return listMono.flatMap(actions -> {
try {
return Mono.just(R.ok(actions));
} catch (Exception e) {
log.error("Failed to list Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除actionNodes
*/
@DeleteMapping("/deleteAction3")
public Mono<R<String>> deleteActions(@RequestParam(name = "index") Integer index) {
Mono<String> stringMono = ruleService.deleteActions(index);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to delete Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除所有actionNodes
*/
@DeleteMapping("/deleteAllAction")
public Mono<R<String>> deleteAllActions() {
Mono<String> stringMono = ruleService.deleteAllActions();
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to delete AllActions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 编辑actionNodes
*/
@PutMapping("/editAction3")
public Mono<R<String>> editActions(@Validated @RequestBody Object data, @RequestParam(name = "index") Integer index) {
Mono<String> stringMono = ruleService.editActions(data, index);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to edit Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
@DeleteMapping("/deleteAction")
public Mono<R<String>> deleteActions(@RequestParam(name = "actionType") String actionType) {
Mono<String> stringMono = ruleService.deleteActions(actionType);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to delete Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
@DeleteMapping("/deleteAction2")
public Mono<R<String>> deleteActions(@RequestParam(name = "actionType") String actionType, @RequestParam(name = "index") Integer index) {
Mono<String> stringMono = ruleService.deleteActions(actionType, index);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to delete Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
@PutMapping("/editAction")
public Mono<R<String>> editActions(@RequestParam(name = "type") String type, @Validated @RequestBody Object data) {
Mono<String> stringMono = ruleService.editActions(data, type);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to edit Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
@PutMapping("/editAction2")
public Mono<R<String>> editActions(@RequestParam(name = "type") String type, @Validated @RequestBody Object data, @RequestParam(name = "index") Integer index) {
Mono<String> stringMono = ruleService.editActions(data, type, index);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to edit Actions API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
}
@@ -1,130 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.controller;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.constant.ServiceConstant;
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.service.ApiService;
import io.github.ponker.center.ekuiper.service.ConfigService;
import io.github.ponker.center.ekuiper.service.SinkTemplateService;
import io.github.ponker.center.ekuiper.service.UrlService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpMethod;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
import java.util.Map;
/**
* @author : Zhen
*/
@Slf4j
@RestController
@RequestMapping(ServiceConstant.Ekuiper.SINKTEM_URL_PREFIX)
public class SinkTemplateController {
@Resource
private ApiService apiService;
@Resource
private ConfigService configService;
@Resource
private SinkTemplateService sinkTemplateService;
@Resource
private ObjectMapper objectMapper;
@Resource
private UrlService urlService;
/**
* 注册, 更新SinkTemplate
*/
@PutMapping("/create")
public Mono<R<String>> createSink(@Validated @RequestBody Object form, @RequestParam(name = "name") String name, @RequestParam(name = "confKey") String confKey) {
String url = urlService.getConfigUrl() + "/sinks/" + name + "/confKeys/" + confKey;
Mono<String> stringMono = sinkTemplateService.callApiSinktem(HttpMethod.PUT, url, form, name);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call createSink API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 列出{name}类型的全部SinkTemplate
*/
@GetMapping("/list")
public Mono<R<Map<String, Map<String, Object>>>> listSink(@RequestParam(name = "name") String name) {
String url = urlService.getConfigUrl() + "/sinks/yaml/" + name;
Mono<String> stringMono = apiService.callApi(HttpMethod.GET, url);
return stringMono.flatMap(jsonString -> {
try {
TypeReference<Map<String, Map<String, Object>>> typeRef = new TypeReference<Map<String, Map<String, Object>>>() {
};
Map<String, Map<String, Object>> configurations = objectMapper.readValue(jsonString, typeRef);
return Mono.just(R.ok(configurations, "successful"));
} catch (Exception e) {
log.error("Failed to call listSink API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除SinkTemplate
*/
@DeleteMapping("/delete")
public Mono<R<String>> deleteSink(@RequestParam(name = "name") String name, @RequestParam(name = "confKey") String confKey) {
String url = urlService.getConfigUrl() + "/sinks/" + name + "/confKeys/" + confKey;
Mono<String> stringMono = configService.callApiDelConf(HttpMethod.DELETE, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call deleteSink API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 测试SinkTemplate连接
*/
@PostMapping("/connection")
public Mono<R<String>> connectSink(@Validated @RequestBody Object form, @RequestParam(name = "name") String name) {
String url = urlService.getConfigUrl() + "/sinks/connection/" + name;
Mono<String> stringMono = sinkTemplateService.callApiSinktem(HttpMethod.POST, url, form, name);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call connectSink API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
}
@@ -1,136 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.controller;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.github.ponker.center.ekuiper.constant.ServiceConstant;
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.entity.dto.RecordDto;
import io.github.ponker.center.ekuiper.entity.dto.StreamFormDto;
import io.github.ponker.center.ekuiper.entity.vo.DetailStreamVO;
import io.github.ponker.center.ekuiper.service.ApiService;
import io.github.ponker.center.ekuiper.service.StreamTableService;
import io.github.ponker.center.ekuiper.service.UrlService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpMethod;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
*/
@Slf4j
@RestController
@RequestMapping(ServiceConstant.Ekuiper.STREAM_URL_PREFIX)
public class StreamController {
@Resource
private ApiService apiService;
@Resource
private StreamTableService streamTableService;
@Resource
private UrlService urlService;
/**
* 创建流
*/
@PostMapping("/create")
public Mono<R<String>> createStream(@Validated @RequestBody StreamFormDto form) {
Mono<String> stringMono = apiService.callApiWithData(form, HttpMethod.POST, urlService.getStreamUrl());
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call create Stream API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 显示所有流
*/
@GetMapping("/list")
public Mono<R<Page<RecordDto>>> listStream(@RequestParam(required = false, defaultValue = "1") Integer pageNum,
@RequestParam(required = false, defaultValue = "10") Integer pageSize) {
Mono<Page<RecordDto>> pageMono = apiService.callApiWithPage(HttpMethod.GET, urlService.getStreamUrl(), pageNum, pageSize);
return pageMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call list Stream API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除流
*/
@DeleteMapping("/delete")
public Mono<R<String>> deleteStream(@RequestParam(name = "id") String id) {
String url = urlService.getStreamUrl() + "/" + id;
Mono<String> stringMono = apiService.callApi(HttpMethod.DELETE, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call delete Stream API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 描述流
*/
@GetMapping("/detail")
public Mono<R<DetailStreamVO>> detailStream(@RequestParam(name = "id") String id) {
String url = urlService.getStreamUrl() + "/" + id;
Mono<DetailStreamVO> detailStreamMono = streamTableService.callApiStream(HttpMethod.GET, url);
return detailStreamMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call detail Stream API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 更新流
*/
@PutMapping("/update")
public Mono<R<String>> updateStream(@Validated @RequestBody StreamFormDto form, @RequestParam(name = "id") String id) {
String url = urlService.getStreamUrl() + "/" + id;
Mono<String> stringMono = apiService.callApiWithData(form, HttpMethod.PUT, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call update Stream API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
}
@@ -1,137 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.controller;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.github.ponker.center.ekuiper.constant.ServiceConstant;
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.entity.dto.RecordDto;
import io.github.ponker.center.ekuiper.entity.dto.StreamFormDto;
import io.github.ponker.center.ekuiper.entity.vo.DetailTableVO;
import io.github.ponker.center.ekuiper.service.ApiService;
import io.github.ponker.center.ekuiper.service.StreamTableService;
import io.github.ponker.center.ekuiper.service.UrlService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpMethod;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
*/
@Slf4j
@RestController
@RequestMapping(ServiceConstant.Ekuiper.TABLE_URL_PREFIX)
public class TableController {
@Resource
private ApiService apiService;
@Resource
private StreamTableService streamTableService;
@Resource
private UrlService urlService;
/**
* 创建表
*/
@PostMapping("/create")
public Mono<R<String>> createTable(@Validated @RequestBody StreamFormDto form) {
Mono<String> stringMono = apiService.callApiWithData(form, HttpMethod.POST, urlService.getTableUrl());
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call create Table API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 列出所有表
*/
@GetMapping("/list")
public Mono<R<Page<RecordDto>>> listTable(@RequestParam(required = false, defaultValue = "1") Integer pageNum,
@RequestParam(required = false, defaultValue = "10") Integer pageSize) {
Mono<Page<RecordDto>> pageMono = apiService.callApiWithPage(HttpMethod.GET, urlService.getTableUrl(), pageNum, pageSize);
return pageMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call list Table API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 删除表
*/
@DeleteMapping("/delete")
public Mono<R<String>> deleteTable(@RequestParam(name = "id") String id) {
String url = urlService.getTableUrl() + "/" + id;
Mono<String> stringMono = apiService.callApi(HttpMethod.DELETE, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call delete Table API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 描述表
*/
@GetMapping("/detail")
public Mono<R<DetailTableVO>> detailTable(@RequestParam(name = "id") String id) {
String url = urlService.getTableUrl() + "/" + id;
Mono<DetailTableVO> detailTableMono = streamTableService.callApiTable(HttpMethod.GET, url);
return detailTableMono.flatMap(s -> {
try {
return Mono.just(R.ok(s, "successful"));
} catch (Exception e) {
log.error("Failed to call detail Table API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
/**
* 更新表
*/
@PutMapping("/update")
public Mono<R<String>> updateTable(@Validated @RequestBody StreamFormDto form, @RequestParam(name = "id") String id) {
String url = urlService.getTableUrl() + "/" + id;
Mono<String> stringMono = apiService.callApiWithData(form, HttpMethod.PUT, url);
return stringMono.flatMap(s -> {
try {
return Mono.just(R.ok(s));
} catch (Exception e) {
log.error("Failed to call update Table API", e);
return Mono.just(R.fail(e.getMessage()));
}
});
}
}
@@ -1,252 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.github.ponker.center.ekuiper.constant.CommonConstant;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import java.io.Serial;
import java.io.Serializable;
/**
* Response
*
* @author pnoker
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class R<T> implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private boolean ok = false;
private int code = Code.OK.getCode();
@JsonInclude(JsonInclude.Include.NON_NULL)
private String message = CommonConstant.Response.ERROR;
@JsonInclude(JsonInclude.Include.NON_NULL)
private T data;
/**
* 构造函数
*
* @param data 数据
*/
private R(T data) {
this.data = data;
}
/**
* 成功
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> ok() {
return new R().success();
}
/**
* 成功 自定义提示信息
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> ok(String message) {
return new R().success(message);
}
/**
* 成功 自定义 Code & 提示信息
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> ok(Code code, String message) {
return new R().success(code.getCode(), message);
}
/**
* 成功 返回结果
*
* @param data 返回结果
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> ok(T data) {
return new R(data).success();
}
/**
* 成功 返回结果 & 自定义提示信息
*
* @param data 返回结果
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> ok(T data, String message) {
return new R(data).success(message);
}
/**
* 失败
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> fail() {
return new R().failure();
}
/**
* 失败 自定义提示信息
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> fail(String message) {
return new R().failure(message);
}
/**
* 失败 自定义 Code & 提示信息
*
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> fail(Code code, String message) {
return new R().failure(code.getCode(), message);
}
/**
* 失败 返回结果
*
* @param data 返回结果
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> fail(T data) {
return new R(data).failure();
}
/**
* 失败 返回结果 & 自定义提示信息
*
* @param data 返回结果
* @return Response
*/
@SuppressWarnings("unchecked")
public static <T> R<T> fail(T data, String message) {
return new R(data).failure(message);
}
/**
* 成功
*
* @return Response
*/
private R success() {
this.ok = true;
this.code = Code.OK.getCode();
this.message = CommonConstant.Response.OK;
return this;
}
/**
* 成功 自定义提示信息
*
* @param message 成功提示信息
* @return Response
*/
private R success(String message) {
this.ok = true;
this.code = Code.OK.getCode();
this.message = message;
return this;
}
/**
* 成功 自定义提示信息
*
* @param code Code
* @param message 成功提示信息
* @return Response
*/
private R success(int code, String message) {
this.ok = true;
this.code = code;
this.message = message;
return this;
}
/**
* 失败
*
* @return Response
*/
private R failure() {
this.ok = false;
this.code = Code.FAILURE.getCode();
this.message = CommonConstant.Response.ERROR;
return this;
}
/**
* 失败 自定义提示信息
*
* @param message 错误提示信息
* @return Response
*/
private R failure(String message) {
this.ok = false;
this.code = Code.FAILURE.getCode();
this.message = message;
return this;
}
/**
* 失败 自定义提示信息
*
* @param code Code
* @param message 错误提示信息
* @return Response
*/
private R failure(int code, String message) {
this.ok = false;
this.code = code;
this.message = message;
return this;
}
public enum Code {
OK(200), FAILURE(500), NotFound(404);
@Getter
private int code;
Code(int code) {
this.code = code;
}
}
}
@@ -1,54 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
import java.util.Map;
/**
* @author : Zhen
*/
@Data
public class HttppullConfigDto {
@NotBlank
private String url;
@NotBlank
private String method;
private Integer interval;
private Integer timeout;
private Boolean incremental;
@NotBlank
private String bodyType;
private Boolean insecureSkipVerify;
@NotBlank
private String responseType;
private Map<String, String> headers;
private String body;
}
@@ -1,31 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class HttppushConfigDto {
@NotBlank
private String method;
}
@@ -1,47 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class MqttConfigDto {
private Integer qos;
@NotBlank
private String server;
private Boolean insecureSkipVerify;
@JsonProperty("username")
private String username;
@JsonProperty("password")
private String password;
private String protocolVersion;
@JsonProperty("clientid")
private String clientid;
}
@@ -1,64 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class MqttSinkDto {
@NotBlank
private String format;
private Boolean sendSingle;
private Boolean omitIfEmpty;
@NotBlank
private String server;
@NotBlank
private String topic;
private Integer qos;
@JsonProperty("clientId")
private String clientId;
@JsonProperty("username")
private String username;
@JsonProperty("password")
private String password;
private String protocolVersion;
private Boolean insecureSkipVerify;
@JsonProperty("retained")
private Boolean retained;
// @JsonProperty("resourceId")
// private String resourceId;
}
@@ -1,33 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* @author : Zhen
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class RecordDto {
private String name;
}
@@ -1,40 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class RedisConfigDto {
@NotBlank
private String dataType;
@NotBlank
private String addr;
@JsonProperty("username")
private String username;
@JsonProperty("password")
private String password;
}
@@ -1,60 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class RedisSinkDto {
@NotBlank
private String format;
private Boolean sendSingle;
private Boolean omitIfEmpty;
@NotBlank
private String addr;
private String key;
private String dataType;
private Integer db;
private Integer expiration;
@JsonProperty("rowkindField")
private String rowkindField;
@JsonProperty("password")
private String password;
private String field;
private String keyType;
// @JsonProperty("resourceId")
// private String resourceId;
}
@@ -1,61 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
import java.util.Map;
/**
* @author : Zhen
*/
@Data
public class RestSinkDto {
@NotBlank
private String format;
private Boolean sendSingle;
private Boolean omitIfEmpty;
@NotBlank
private String url;
@NotBlank
private String method;
private String bodyType;
private Map<String, String> headers;
private String responseType;
private Boolean insecureSkipVerify;
@JsonProperty("timeout")
private Integer timeout;
@JsonProperty("debugResp")
private Boolean debugResp;
// @JsonProperty("resourceId")
// private String resourceId;
}
@@ -1,50 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
* @date : 2024/2/23
* 对象聚合
*/
@Data
public class SqlConfigDto {
@NotBlank
private String url;
private Integer interval;
private action internalSqlQueryCfg;
}
@Data
class action {
@NotBlank
private String indexField;
private Integer indexValue;
private Integer limit;
@NotBlank
private String table;
}
@@ -1,52 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class SqlSinkDto {
@NotBlank
private String url;
@NotBlank
private String table;
private String keyField;
@JsonProperty("rowkindField")
private String rowkindField;
@NotBlank
private String format;
private Boolean sendSingle;
private Boolean omitIfEmpty;
private String tableDataField;
// @JsonProperty("resourceId")
// private String resourceId;
}
@@ -1,31 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.dto;
import jakarta.validation.constraints.NotBlank;
import lombok.Data;
/**
* @author : Zhen
*/
@Data
public class StreamFormDto {
@NotBlank(message = "SQL不能为空")
private String sql;
}
@@ -1,47 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.vo;
import lombok.Data;
import java.util.List;
import java.util.Map;
/**
* @author : Zhen
*/
@Data
//@JsonIgnoreProperties(ignoreUnknown = true)
public class DetailRuleVO {
private String id;
private String name;
private String sql;
private List<Map<String, Object>> actions;
private Map<String, Object> options;
//private Map<String, Object> dynamicProperties = new HashMap<>();
// @JsonAnySetter
// public void setDynamicProperty(String name, Object value) {
// dynamicProperties.put(name, value);
// }
}
@@ -1,103 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.vo;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.List;
/**
* @author : Zhen
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class DetailStreamVO {
@JsonProperty("Name")
private String Name;
@JsonProperty("StreamFields")
private List<StreamField> StreamFields;
@JsonProperty("Options")
private Options Options;
@JsonProperty("StreamType")
private int StreamType;
@JsonProperty("Statement")
private String Statement;
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class StreamField {
@JsonProperty("Name")
private String Name;
@JsonProperty("FieldType")
private String FieldType;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class FieldType {
private int type;
}
@Data
@NoArgsConstructor
public static class Options {
private String datasource;
private String format;
private String confKey;
private String type;
private Boolean shared;
public Options(String datasource, String format, String confKey, String type, Boolean shared) {
this.datasource = datasource;
this.format = format;
this.confKey = confKey;
this.type = type;
// 如果提供了 shared 属性, 使用提供的值, 否则默认为 false
this.shared = shared != null ? shared : false;
}
// 添加 getter 和 setter 方法, 以便在接收 JSON 数据时进行处理
public Boolean getShared() {
return shared != null ? shared : false;
}
public void setShared(Boolean shared) {
this.shared = shared != null ? shared : false;
}
}
}
@@ -1,85 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.vo;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.List;
/**
* @author : Zhen
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class DetailTableVO {
@JsonProperty("Name")
private String Name;
@JsonProperty("StreamFields")
private List<StreamField> StreamFields;
@JsonProperty("Options")
private Options Options;
@JsonProperty("StreamType")
private int StreamType;
@JsonProperty("Statement")
private String Statement;
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class StreamField {
@JsonProperty("Name")
private String Name;
@JsonProperty("FieldType")
private String FieldType;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class FieldType {
private int type;
}
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class Options {
private String datasource;
private String format;
private String confKey;
private String type;
private String key;
private String kind;
}
}
@@ -1,38 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.vo;
import jakarta.validation.constraints.NotBlank;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
/**
* @author : Zhen
*/
@lombok.Data
@AllArgsConstructor
@NoArgsConstructor
public class RuleDataVO {
@NotBlank
private String id;
private String name;
@NotBlank
private String status;
}
@@ -1,40 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.entity.vo;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
import java.util.HashMap;
import java.util.Map;
/**
* @author : Zhen
*/
@JsonIgnoreProperties(ignoreUnknown = true)
@Data
public class RuleStatusVO {
private String status;
private Map<String, Object> dynamicProperties = new HashMap<>();
@JsonAnySetter
public void setDynamicProperty(String name, Object value) {
dynamicProperties.put(name, value);
}
}
@@ -1,31 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.exception;
import cn.hutool.core.util.StrUtil;
/**
* 自定义 规则引擎 异常
*
* @author : Zhen
*/
public class EkuiperException extends RuntimeException {
public EkuiperException(CharSequence template, Object... params) {
super(StrUtil.format(template, params));
}
}
@@ -1,104 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.exception;/*
* Copyright 2022 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
*
* https://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.
*/
import io.github.ponker.center.ekuiper.entity.R;
import io.github.ponker.center.ekuiper.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.validation.BindException;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import java.util.HashMap;
import java.util.List;
/**
* 全局异常处理
*
* @author pnoker
*/
@Slf4j
@RestControllerAdvice
public class GlobalExceptionAdvice {
/**
* Global Exception
*
* @param exception Exception
* @return R
*/
@ExceptionHandler(Exception.class)
@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
public R globalException(Exception exception) {
log.error("Global Exception Handler: {}", exception.getMessage(), exception);
return R.fail(exception.getMessage());
}
/**
* UnAuthorized Exception
*
* @param unAuthorizedException UnAuthorizedException
* @return R
*/
@ExceptionHandler(UnAuthorizedException.class)
@ResponseStatus(HttpStatus.UNAUTHORIZED)
public R unAuthorizedException(UnAuthorizedException unAuthorizedException) {
log.warn("UnAuthorized Exception Handler: {}", unAuthorizedException.getMessage(), unAuthorizedException);
return R.fail(unAuthorizedException.getMessage());
}
/**
* Validation Exception
*
* @param exception MethodArgumentNotValidException
* @return R
*/
@ExceptionHandler({
BindException.class,
MethodArgumentNotValidException.class
})
@ResponseStatus(HttpStatus.PRECONDITION_FAILED)
public R methodArgumentNotValidException(MethodArgumentNotValidException exception) {
HashMap<String, String> map = new HashMap<>(16);
List<FieldError> errorList = exception.getBindingResult().getFieldErrors();
errorList.forEach(error -> {
log.warn("Method Argument Not Valid Exception Handler: {}[{}]", error.getField(), error.getDefaultMessage());
map.put(error.getField(), error.getDefaultMessage());
});
return R.fail(JsonUtil.toJsonString(map));
}
}
@@ -1,45 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.exception;/*
* Copyright 2022 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
*
* https://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.
*/
import cn.hutool.core.util.StrUtil;
/**
* 自定义 未授权 异常
*
* @author pnoker
*/
public class UnAuthorizedException extends RuntimeException {
public UnAuthorizedException(CharSequence template, Object... params) {
super(StrUtil.format(template, params));
}
}
@@ -1,41 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.github.ponker.center.ekuiper.entity.dto.RecordDto;
import org.springframework.http.HttpMethod;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
* @date : 2024/3/11
*/
public interface ApiService {
Mono<String> callApi(HttpMethod method, String url);
Mono<String> callApiWithData(Object data, HttpMethod method, String url);
Mono<Page<RecordDto>> callApiWithPage(HttpMethod method, String url, Integer current, Integer size);
}
@@ -1,33 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
import org.springframework.http.HttpMethod;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
* @date : 2024/3/11
*/
public interface ConfigService {
Mono<String> callApiDelConf(HttpMethod method, String url);
Mono<String> callApiConfig(HttpMethod method, String url, Object data, String name);
}
@@ -1,62 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.github.ponker.center.ekuiper.entity.vo.DetailRuleVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleDataVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleStatusVO;
import org.springframework.http.HttpMethod;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
/**
* @author : Zhen
* @date : 2024/3/11
*/
public interface RuleService {
Mono<DetailRuleVO> callApilRule(HttpMethod method, String url);
Mono<RuleStatusVO> callApiRuleStatus(HttpMethod method, String url);
Mono<String> callRuleApiWithData(Object data, HttpMethod method, String url) throws Exception;
Mono<Page<RuleDataVO>> callApiWithRulePage(HttpMethod method, String url, Integer current, Integer size);
Mono<String> addActions(Object data);
Mono<List<Map<String, Map<String, Object>>>> listActions();
Mono<String> deleteActions(String actionType);
Mono<String> deleteActions(String actionType, Integer index);
Mono<String> deleteActions(Integer index);
Mono<String> editActions(Object data, String type);
Mono<String> editActions(Object data, Integer index);
Mono<String> editActions(Object data, String type, Integer index);
Mono<String> deleteAllActions();
}
@@ -1,32 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
import org.springframework.http.HttpMethod;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
* @date : 2024/3/12
*/
public interface SinkTemplateService {
Mono<String> callApiSinktem(HttpMethod method, String url, Object data, String name);
Mono<String> callConfigAndSink(HttpMethod method, String url);
}
@@ -1,34 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
import io.github.ponker.center.ekuiper.entity.vo.DetailStreamVO;
import io.github.ponker.center.ekuiper.entity.vo.DetailTableVO;
import org.springframework.http.HttpMethod;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
* @date : 2024/3/11
*/
public interface StreamTableService {
Mono<DetailStreamVO> callApiStream(HttpMethod method, String url);
Mono<DetailTableVO> callApiTable(HttpMethod method, String url);
}
@@ -1,32 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service;
/**
* @author : Zhen
* @date : 2024/2/20
*/
public interface UrlService {
String getStreamUrl();
String getTableUrl();
String getConfigUrl();
String getRuleUrl();
}
@@ -1,142 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.entity.dto.RecordDto;
import io.github.ponker.center.ekuiper.service.ApiService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
/**
* @author : Zhen
* @date : 2024/3/11
*/
@Slf4j
@Service
public class ApiServiceImpl implements ApiService {
@Autowired
private WebClient webClient;
@Autowired
private ObjectMapper objectMapper;
@Override
public Mono<String> callApi(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(String.class)
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用API时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用API时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
@Override
public Mono<String> callApiWithData(Object data, HttpMethod method, String url) {
WebClient.RequestBodySpec request = webClient.method(method).uri(url);
if (Objects.nonNull(data)) {
try {
String jsonBody = objectMapper.writeValueAsString(data);
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(jsonBody));
} catch (Exception e) {
log.info("将对象序列化为JSON失败");
return Mono.error(e);
}
}
return request
.retrieve()
.bodyToMono(String.class)
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callApiWithData时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callApiWithData时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
@Override
public Mono<Page<RecordDto>> callApiWithPage(HttpMethod method, String url, Integer current, Integer size) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(String.class)
.flatMap(response -> {
//获取响应后进行MyBatisPlus分页查询
List<RecordDto> dataList = getDataList(response); // 修改此处
Page<RecordDto> page = getPageSubset(current, size, dataList); // 修改此处
return Mono.just(page);
});
}
private List<RecordDto> getDataList(String response) {
List<RecordDto> dataList = new ArrayList<>();
try {
// 解析成JSON数组
JSONArray jsonArray = JSONArray.parseArray(response);
// 将 JSON 数组中的元素逐个添加到 dataList 中
for (int i = 0; i < jsonArray.size(); i++) {
String recordName = jsonArray.getString(i);
dataList.add(new RecordDto(recordName));
}
} catch (JSONException e) {
e.printStackTrace();
}
return dataList;
}
private Page<RecordDto> getPageSubset(int current, int size, List<RecordDto> dataList) {
int fromIndex = Math.max((current - 1) * size, 0);
int toIndex = Math.min(current * size, dataList.size());
List<RecordDto> subset = dataList.subList(fromIndex, toIndex);
Page<RecordDto> page = new Page<>(current, size, dataList.size());
page.setRecords(subset);
return page;
}
}
@@ -1,240 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.entity.dto.*;
import io.github.ponker.center.ekuiper.service.ConfigService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import java.util.HashSet;
import java.util.Set;
/**
* @author : Zhen
* @date : 2024/3/11
*/
@Slf4j
@Service
public class ConfigServiceImpl implements ConfigService {
@Autowired
private WebClient webClient;
@Autowired
private ObjectMapper objectMapper;
@Autowired
private SqlConfigDto SqlConfigDto;
@Override
public Mono<String> callApiDelConf(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
//.bodyToMono(String.class);
//.toBodilessEntity().map(responseEntity -> responseEntity.getStatusCode().is2xxSuccessful());
.toBodilessEntity()
.flatMap(responseEntity -> {
if (responseEntity.getStatusCode().is2xxSuccessful()) {
return Mono.just("true"); // 返回成功消息 "true"
} else {
return Mono.error(new RuntimeException("请求失败--状态码错误: " + responseEntity.getStatusCode()));
}
})
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callApiDelConf时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callApiDelConf时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
@Override
public Mono<String> callApiConfig(HttpMethod method, String url, Object data, String name) {
WebClient.RequestBodySpec request = webClient.method(method).uri(url);
try {
String dataJsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = objectMapper.readTree(dataJsonBody);
switch (name) {
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.MQTT:
handleMqttRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.HTTPPUSH:
handleHttpPushRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.SQL:
handleSqlRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.HTTPPULL:
handleHttpPullRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.REDIS:
handleRedisRequest(request, dataJsonNode, dataJsonBody);
break;
default:
throw new IllegalArgumentException("无效的名称");
}
} catch (JsonProcessingException e) {
log.info("对象序列化为JSON失败");
e.printStackTrace();
}
return request.retrieve()
//.bodyToMono
// .toBodilessEntity().map(responseEntity -> responseEntity.getStatusCode().is2xxSuccessful());
//.bodyToMono(String.class)
.toBodilessEntity()
.flatMap(responseEntity -> {
if (responseEntity.getStatusCode().is2xxSuccessful()) {
return Mono.just("true"); // 返回成功消息 "true"
} else {
return Mono.error(new RuntimeException("请求失败--状态码错误: " + responseEntity.getStatusCode()));
}
})
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callApiConfig时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callApiConfig时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
private void handleMqttRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
MqttConfigDto mqttConfigDto = new MqttConfigDto();
JsonNode mqttJsonNode = generateJsonNode(mqttConfigDto);
if (containsAllFields(dataJsonNode, mqttJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from mqttConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from mqttConfigForm");
}
} catch (Exception e) {
log.info("将对象序列化为JSON失败");
e.printStackTrace();
}
}
private void handleHttpPushRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
HttppushConfigDto httppushConfigDto = new HttppushConfigDto();
JsonNode httppushJsonNode = generateJsonNode(httppushConfigDto);
if (containsAllFields(dataJsonNode, httppushJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from httpPushConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from httpPushConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleSqlRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
//SqlConfigDto sqlConfigDto = new SqlConfigDto();
JsonNode sqlJsonNode = generateJsonNode(SqlConfigDto);
if (containsAllFields(dataJsonNode, sqlJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from sqlConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from sqlConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleHttpPullRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
HttppullConfigDto httppullConfigDto = new HttppullConfigDto();
JsonNode httppulJsonNode = generateJsonNode(httppullConfigDto);
if (containsAllFields(dataJsonNode, httppulJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from httpPullConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from httpPullConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleRedisRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
RedisConfigDto redisConfigDto = new RedisConfigDto();
JsonNode redisJsonNode = generateJsonNode(redisConfigDto);
if (containsAllFields(dataJsonNode, redisJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from redisConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from redisConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private boolean containsAllFields(JsonNode node1, JsonNode node2) {
Set<String> keys1 = new HashSet<>();
Set<String> keys2 = new HashSet<>();
node1.fieldNames().forEachRemaining(keys1::add);
node2.fieldNames().forEachRemaining(keys2::add);
return keys1.equals(keys2);
}
private JsonNode generateJsonNode(Object data) throws JsonProcessingException {
String dataJsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = objectMapper.readTree(dataJsonBody);
return dataJsonNode;
}
}
@@ -1,428 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.github.ponker.center.ekuiper.entity.dto.MqttSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.RedisSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.RestSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.SqlSinkDto;
import io.github.ponker.center.ekuiper.entity.vo.DetailRuleVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleDataVO;
import io.github.ponker.center.ekuiper.entity.vo.RuleStatusVO;
import io.github.ponker.center.ekuiper.exception.EkuiperException;
import io.github.ponker.center.ekuiper.service.RuleService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import java.util.*;
/**
* @author : Zhen
* @date : 2024/3/11
*/
@Slf4j
@Service
public class RuleServiceImpl implements RuleService {
@Autowired
private WebClient webClient;
@Autowired
private ObjectMapper objectMapper;
private List<Map<String, Map<String, Object>>> actions = new ArrayList<>();
public static Page<RuleDataVO> getPageSubset2(int current, int size, List<RuleDataVO> ruleDataVOList) {
int fromIndex = Math.max((current - 1) * size, 0);
int toIndex = Math.min(current * size, ruleDataVOList.size());
List<RuleDataVO> subset = ruleDataVOList.subList(fromIndex, toIndex);
Page<RuleDataVO> page = new Page<>(current, size, ruleDataVOList.size());
page.setRecords(subset);
return page;
}
@Override
public Mono<DetailRuleVO> callApilRule(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(DetailRuleVO.class);
}
@Override
public Mono<RuleStatusVO> callApiRuleStatus(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(RuleStatusVO.class);
}
@Override
public Mono<String> callRuleApiWithData(Object data, HttpMethod method, String url) throws Exception {
WebClient.RequestBodySpec request = webClient.method(method).uri(url);
if (Objects.nonNull(data)) {
try {
String jsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = generateJsonNode(data);
validateSinkStructure(dataJsonNode);
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(jsonBody));
} catch (JsonProcessingException e) {
log.error("将对象序列化为JSON失败: {}", e.getMessage());
throw new EkuiperException("JSON序列化失败");
}
}
return request.retrieve()
.bodyToMono(String.class)
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callRuleApiWithData时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callRuleApiWithData时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
@Override
public Mono<Page<RuleDataVO>> callApiWithRulePage(HttpMethod method, String url, Integer current, Integer size) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(String.class)
.flatMap(response -> {
//获取响应后进行MyBatisPlus分页查询
List<RuleDataVO> ruleDataVOList = getDataList2(response);
Page<RuleDataVO> page = getPageSubset2(current, size, ruleDataVOList);
return Mono.just(page);
});
}
@Override
public Mono<String> addActions(Object data) {
try {
String jsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = generateJsonNode(data);
//validateActionNodes(dataJsonNode);
} catch (Exception e) {
log.error("将对象序列化为JSON失败: {}", e.getMessage());
throw new EkuiperException("JSON序列化失败");
}
//actions.clear(); // 清空 actions 集合
actions.add((Map<String, Map<String, Object>>) data); // 添加新元素
return Mono.just("successful");
}
public Mono<List<Map<String, Map<String, Object>>>> listActions() {
return Mono.just(actions);
}
public Mono<String> deleteActions(String actionType) {
try {
// 遍历 actions 列表
Iterator<Map<String, Map<String, Object>>> iterator = actions.iterator();
while (iterator.hasNext()) {
Map<String, Map<String, Object>> action = iterator.next();
// 如果 action 包含指定的 actionType, 则删除
if (action.containsKey(actionType)) {
iterator.remove();
}
}
// 操作完成后返回 "successful"
return Mono.just("successful");
} catch (Exception e) {
// 如果发生异常, 则返回 Mono.error
return Mono.error(e);
}
}
@Override
public Mono<String> deleteActions(String actionType, Integer index) {
try {
int currentIndex = 0; // 当前数据索引
// 遍历 actions 列表
Iterator<Map<String, Map<String, Object>>> iterator = actions.iterator();
while (iterator.hasNext()) {
Map<String, Map<String, Object>> action = iterator.next();
// 检查当前 action 是否包含指定的 actionType
if (action.containsKey(actionType)) {
// 如果索引等于指定的 index, 则删除
if (currentIndex == index) {
iterator.remove();
return Mono.just("删除成功");
}
currentIndex++; // 增加索引
}
}
// 如果没有找到匹配的数据, 则返回错误信息
return Mono.error(new RuntimeException("找不到类型为: " + actionType + ", 索引为: " + index + " 的数据"));
} catch (Exception e) {
// 如果发生异常, 则返回 Mono.error
return Mono.error(e);
}
}
@Override
public Mono<String> deleteActions(Integer index) {
try {
int currentIndex = 0; // 当前数据索引
// 遍历 actions 列表
Iterator<Map<String, Map<String, Object>>> iterator = actions.iterator();
while (iterator.hasNext()) {
Map<String, Map<String, Object>> action = iterator.next();
// 如果当前 action 不为空
if (!action.isEmpty()) {
// 如果索引等于指定的 index, 则删除
if (currentIndex == index) {
iterator.remove();
return Mono.just("删除成功");
}
currentIndex++; // 增加索引
}
}
// 如果没有找到匹配的数据, 则返回错误信息
return Mono.error(new RuntimeException("索引超出范围"));
} catch (Exception e) {
// 如果发生异常, 则返回 Mono.error
return Mono.error(e);
}
}
@Override
public Mono<String> editActions(Object data, String type) {
// 解析传入的数据为Map对象
Map<String, Map<String, Object>> newData = (Map<String, Map<String, Object>>) data;
// 遍历actions列表中的每个Map对象
for (Map<String, Map<String, Object>> actionMap : actions) {
// 获取指定类型的结构
Map<String, Object> targetConfig = actionMap.get(type);
// 如果存在指定类型的结构, 则更新配置
if (targetConfig != null) {
updateConfig(targetConfig, newData.get(type));
return Mono.just(type + "编辑成功");
}
}
// 如果actions列表中没有指定类型的结构, 则返回错误信息
return Mono.error(new RuntimeException("No configuration found for type: " + type));
}
@Override
public Mono<String> editActions(Object data, String type, Integer index) {
// 解析传入的数据为Map对象
Map<String, Map<String, Object>> newData = (Map<String, Map<String, Object>>) data;
int currentIndex = 0; // 当前 MQTT 数据索引
// 遍历 actions 列表中的每个 Map 对象
for (Map<String, Map<String, Object>> actionMap : actions) {
// 检查当前 Map 对象是否包含指定类型的结构
if (actionMap.containsKey(type)) {
// 如果是 MQTT 类型, 并且索引等于指定的 mqttIndex, 则更新配置并返回成功消息
if (currentIndex == index) {
updateConfig(actionMap.get(type), newData.get(type));
return Mono.just(type + "[" + index + "]" + "编辑成功");
}
currentIndex++; // 仅当类型为 MQTT 时增加索引
}
}
// 如果 actions 列表中没有指定类型和索引的结构, 则返回错误信息
return Mono.error(new RuntimeException("找不到类型为: " + type + ", 索引为: " + index + " 的配置"));
}
@Override
public Mono<String> editActions(Object data, Integer index) {
// 解析传入的数据为Map<String, Map<String, Object>>对象
Map<String, Map<String, Object>> newDataMap = (Map<String, Map<String, Object>>) data;
// 检查 index 是否在有效范围内
if (index >= 0 && index < actions.size()) {
Map<String, Map<String, Object>> newData = newDataMap;
// 遍历 actions 列表中的每个 Map 对象
for (int i = 0; i < actions.size(); i++) {
// 检查当前 Map 对象是否与 newData 相等
if (actions.get(index) != null) {
// 转换原始配置和新配置为Map<String, Object>类型
Map<String, Object> originalConfig = convertToMap(actions.get(i));
Map<String, Object> newConfig = convertToMap(newData);
// 调用updateConfig方法更新配置
updateConfig(originalConfig, newConfig);
// 更新actions列表中的数据
actions.set(index, newData);
return Mono.just("actions中第" + (index + 1) + "个sink编辑成功");
}
}
// 如果 actions 列表中没有与 newData 匹配的结构, 则返回错误信息
return Mono.error(new RuntimeException("找不到要编辑的数据"));
} else {
return Mono.error(new RuntimeException("索引越界"));
}
}
// 更新配置方法
private void updateConfig(Map<String, Object> originalConfig, Map<String, Object> newConfig) {
for (Map.Entry<String, Object> entry : newConfig.entrySet()) {
// 检查新传入的字段是否有更新
if (originalConfig.containsKey(entry.getKey())) {
// 如果新传入的字段在原始配置中存在, 则更新原始配置中的对应字段
originalConfig.put(entry.getKey(), entry.getValue());
}
}
}
// 辅助方法: 将Map<String, Map<String, Object>>类型转换为Map<String, Object>类型
private Map<String, Object> convertToMap(Map<String, Map<String, Object>> map) {
Map<String, Object> result = new HashMap<>();
for (Map.Entry<String, Map<String, Object>> entry : map.entrySet()) {
result.put(entry.getKey(), entry.getValue());
}
return result;
}
@Override
public Mono<String> deleteAllActions() {
actions.clear(); // 清空 actions 集合
return Mono.just("successful");
}
private JsonNode generateJsonNode(Object data) throws JsonProcessingException {
String dataJsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = objectMapper.readTree(dataJsonBody);
return dataJsonNode;
}
private boolean containsAllFields(JsonNode node1, JsonNode node2) {
Set<String> keys1 = new HashSet<>();
Set<String> keys2 = new HashSet<>();
node1.fieldNames().forEachRemaining(keys1::add);
node2.fieldNames().forEachRemaining(keys2::add);
return keys1.equals(keys2);
}
private List<RuleDataVO> getDataList2(String response) {
List<RuleDataVO> ruleDataVOList = new ArrayList<>();
try {
// 解析成 JSON 数组
JSONArray jsonArray = JSONArray.parseArray(response);
// 遍历 JSON 数组中的每个元素, 将其转换为 Data 对象, 并添加到 dataList 中
for (int i = 0; i < jsonArray.size(); i++) {
// 获取 JSON 对象
JSONObject jsonObject = jsonArray.getJSONObject(i);
// 解析 JSON 对象的属性
String id = jsonObject.getString("id");
String name = jsonObject.getString("name");
String status = jsonObject.getString("status");
// 创建 Data 对象, 并添加到 dataList 中
RuleDataVO data = new RuleDataVO(id, name, status);
ruleDataVOList.add(data);
}
} catch (JSONException e) {
e.printStackTrace();
}
return ruleDataVOList;
}
private void validateSinkStructure(JsonNode dataJsonNode) throws Exception {
if (!dataJsonNode.has("actions") || !dataJsonNode.has("id") || !dataJsonNode.has("sql")) {
log.error("RuleSink结构错误");
throw new EkuiperException("RuleSink结构错误");
}
JsonNode actionsNode = dataJsonNode.get("actions");
if (!actionsNode.isArray()) {
log.error("actionsNode结构错误");
throw new EkuiperException("actionsNode结构错误");
}
for (JsonNode action : actionsNode) {
String actionName = action.fieldNames().next();
log.info("发现action: " + actionName);
JsonNode actionNode = action.get(actionName);
if (!isValidSinkType(actionName, actionNode)) {
log.info("Sink中{}表单结构不符合规范", actionName.toUpperCase());
throw new EkuiperException("Sink中" + actionName.toUpperCase() + "表单结构不符合规范");
}
}
}
private void validateActionNodes(JsonNode dataJsonNode) throws Exception {
if (!dataJsonNode.has("actions")) {
log.error("actionNodes结构错误");
throw new EkuiperException("actionNodes结构错误");
}
JsonNode actionsNode = dataJsonNode.get("actions");
if (!actionsNode.isArray()) {
log.error("actionsNode结构错误");
throw new EkuiperException("actionsNode结构错误");
}
for (JsonNode action : actionsNode) {
String actionName = action.fieldNames().next();
log.info("发现action: " + actionName);
JsonNode actionNode = action.get(actionName);
if (!isValidSinkType(actionName, actionNode)) {
log.info("Sink中{}表单结构不符合规范", actionName.toUpperCase());
throw new EkuiperException("Sink中" + actionName.toUpperCase() + "表单结构不符合规范");
}
}
}
private boolean isValidSinkType(String actionName, JsonNode actionNode) throws JsonProcessingException {
switch (actionName) {
case "mqtt":
return containsAllFields(generateJsonNode(new MqttSinkDto()), actionNode);
case "redis":
return containsAllFields(generateJsonNode(new RedisSinkDto()), actionNode);
case "rest":
return containsAllFields(generateJsonNode(new RestSinkDto()), actionNode);
case "sql":
return containsAllFields(generateJsonNode(new SqlSinkDto()), actionNode);
default:
return false;
}
}
}
@@ -1,262 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.github.ponker.center.ekuiper.entity.dto.MqttSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.RedisSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.RestSinkDto;
import io.github.ponker.center.ekuiper.entity.dto.SqlSinkDto;
import io.github.ponker.center.ekuiper.service.SinkTemplateService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Mono;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
/**
* @author : Zhen
* @date : 2024/3/12
*/
@Slf4j
@Service
public class SinkTemplateServiceImpl implements SinkTemplateService {
@Autowired
private WebClient webClient;
@Autowired
private ObjectMapper objectMapper;
@Override
public Mono<String> callApiSinktem(HttpMethod method, String url, Object data, String name) {
WebClient.RequestBodySpec request = webClient.method(method).uri(url);
try {
String dataJsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = objectMapper.readTree(dataJsonBody);
switch (name) {
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.MQTT:
handleMqttRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.REST:
handleRestRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.SQL:
handleSqlRequest(request, dataJsonNode, dataJsonBody);
break;
case io.github.ponker.center.ekuiper.constant.EkuiperConstant.ConKeyType.REDIS:
handleRedisRequest(request, dataJsonNode, dataJsonBody);
break;
default:
throw new IllegalArgumentException("无效的名称");
}
} catch (JsonProcessingException e) {
log.info("将对象序列化为JSON失败");
e.printStackTrace();
}
return request.retrieve()
//.bodyToMono
// .toBodilessEntity().map(responseEntity -> responseEntity.getStatusCode().is2xxSuccessful());
//.bodyToMono(String.class)
.toBodilessEntity()
.flatMap(responseEntity -> {
if (responseEntity.getStatusCode().is2xxSuccessful()) {
return Mono.just("true"); // 返回成功消息 "true"
} else {
return Mono.error(new RuntimeException("请求失败--状态码错误: " + responseEntity.getStatusCode()));
}
})
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callApiSinktem时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callApiSinktem时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
@Override
public Mono<String> callConfigAndSink(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(String.class)
.map(response -> {
try {
// 解析原始 JSON
JsonNode root = objectMapper.readTree(response);
// 构造新的 JSON 结构
ArrayNode newSinks = objectMapper.createArrayNode();
ArrayNode newSources = objectMapper.createArrayNode();
// 处理 sinks
JsonNode sinks = root.get("sinks");
for (JsonNode sinkGroup : sinks) {
Iterator<Map.Entry<String, JsonNode>> fields = sinkGroup.fields();
while (fields.hasNext()) {
Map.Entry<String, JsonNode> entry = fields.next();
ObjectNode newSink = objectMapper.createObjectNode();
newSink.put("name", entry.getKey());
newSink.put("type", entry.getValue().asText());
newSinks.add(newSink);
}
}
// 处理 sources
JsonNode sources = root.get("sources");
for (JsonNode sourceGroup : sources) {
Iterator<Map.Entry<String, JsonNode>> fields = sourceGroup.fields();
while (fields.hasNext()) {
Map.Entry<String, JsonNode> entry = fields.next();
ObjectNode newSource = objectMapper.createObjectNode();
newSource.put("name", entry.getKey());
newSource.put("type", entry.getValue().asText());
newSources.add(newSource);
}
}
// 构造新的 JSON 结构
ObjectNode newRoot = objectMapper.createObjectNode();
newRoot.set("sinks", newSinks);
newRoot.set("sources", newSources);
// 将修改后的 JSON 结构转换为字符串
return objectMapper.writeValueAsString(newRoot);
} catch (Exception e) {
log.error("处理JSON时发生错误: {}", e.getMessage());
throw new RuntimeException("处理JSON时发生错误", e);
}
})
.onErrorResume(error -> {
if (error instanceof WebClientResponseException) {
WebClientResponseException responseException = (WebClientResponseException) error;
log.error("调用callConfigAndSink时发生错误. Status code: {}, Response body: {}",
responseException.getRawStatusCode(), responseException.getResponseBodyAsString());
return Mono.error(new RuntimeException(responseException.getResponseBodyAsString()));
} else {
log.error("调用callConfigAndSink时发生错误: {}", error.getMessage());
return Mono.error(error);
}
});
}
private void handleMqttRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
MqttSinkDto mqttSinkDto = new MqttSinkDto();
JsonNode mqttJsonNode = generateJsonNode(mqttSinkDto);
if (containsAllFields(dataJsonNode, mqttJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from mqttSink");
throw new IllegalArgumentException("The data doesn't contain all fields from mqttSink");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleRestRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
RestSinkDto restSinkDto = new RestSinkDto();
JsonNode httppushJsonNode = generateJsonNode(restSinkDto);
if (containsAllFields(dataJsonNode, httppushJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from httpPushConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from httpPushConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleSqlRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
SqlSinkDto sqlSinkDto = new SqlSinkDto();
JsonNode sqlJsonNode = generateJsonNode(sqlSinkDto);
if (containsAllFields(dataJsonNode, sqlJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from sqlConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from sqlConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private void handleRedisRequest(WebClient.RequestBodySpec request, JsonNode dataJsonNode, String dataJsonBody) {
try {
RedisSinkDto redisSinkDto = new RedisSinkDto();
JsonNode redisJsonNode = generateJsonNode(redisSinkDto);
if (containsAllFields(dataJsonNode, redisJsonNode)) {
request = (WebClient.RequestBodySpec) request.body(BodyInserters.fromValue(dataJsonBody));
} else {
log.info("The data doesn't contain all fields from redisConfigForm");
throw new IllegalArgumentException("The data doesn't contain all fields from redisConfigForm");
}
} catch (Exception e) {
log.info("Failed to serialize object to JSON:");
e.printStackTrace();
}
}
private boolean containsAllFields(JsonNode node1, JsonNode node2) {
Set<String> keys1 = new HashSet<>();
Set<String> keys2 = new HashSet<>();
node1.fieldNames().forEachRemaining(keys1::add);
node2.fieldNames().forEachRemaining(keys2::add);
return keys1.equals(keys2);
}
private JsonNode generateJsonNode(Object data) throws JsonProcessingException {
String dataJsonBody = objectMapper.writeValueAsString(data);
JsonNode dataJsonNode = objectMapper.readTree(dataJsonBody);
return dataJsonNode;
}
}
@@ -1,57 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import io.github.ponker.center.ekuiper.entity.vo.DetailStreamVO;
import io.github.ponker.center.ekuiper.entity.vo.DetailTableVO;
import io.github.ponker.center.ekuiper.service.StreamTableService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
/**
* @author : Zhen
* @date : 2024/3/11
*/
@Slf4j
@Service
public class StreamTableServiceImpl implements StreamTableService {
@Autowired
private WebClient webClient;
@Override
public Mono<DetailStreamVO> callApiStream(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(DetailStreamVO.class);
}
@Override
public Mono<DetailTableVO> callApiTable(HttpMethod method, String url) {
return webClient
.method(method)
.uri(url)
.retrieve()
.bodyToMono(DetailTableVO.class);
}
}
@@ -1,64 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.service.impl;
import io.github.ponker.center.ekuiper.service.UrlService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
/**
* @author : Zhen
* @date : 2024/2/20
*/
@Slf4j
@Service
public class UrlServiceImpl implements UrlService {
@Value("${driver.custom.STREAM_URL}")
private String streamUrl;
@Value("${driver.custom.TABLE_URL}")
private String tableUrl;
@Value("${driver.custom.CONFKEY_URL}")
private String confKeyUrl;
@Value("${driver.custom.RULE_URL}")
private String ruleUrl;
@Override
public String getStreamUrl() {
return streamUrl;
}
@Override
public String getTableUrl() {
return tableUrl;
}
@Override
public String getConfigUrl() {
return confKeyUrl;
}
@Override
public String getRuleUrl() {
return ruleUrl;
}
}
@@ -1,375 +0,0 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* 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
*
* https://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 io.github.ponker.center.ekuiper.utils;/*
* Copyright 2022 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
*
* https://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.
*/
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.json.JsonMapper;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
/**
* Json 工具类
* 基于 Jackson 实现
*
* @author pnoker
*/
public final class JsonUtil {
private static final ObjectMapper OBJECT_MAPPER = JsonMapper.builder()
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, Boolean.FALSE)
.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, Boolean.FALSE)
.serializationInclusion(JsonInclude.Include.NON_NULL).build();
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param text Json String
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(String text, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(text, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param bytes Json Byte Array
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(byte[] bytes, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(bytes, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param jsonParser Json Parser
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(JsonParser jsonParser, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(jsonParser, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param dataInput Json DataInput
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(DataInput dataInput, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(dataInput, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param inputStream Json InputStream
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(InputStream inputStream, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(inputStream, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param reader Json Reader
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(Reader reader, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(reader, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象
*
* @param file Json File
* @param valueType Java 对象
* @return Java 对象
*/
public static <T> T parseObject(File file, Class<T> valueType) {
try {
return OBJECT_MAPPER.readValue(file, valueType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将Json 字符串转化为指定对象,并将指定对象中的泛型转为指定的T类型
*
* @param text Json String
* @param typeReference TypeReference, T 为指定要转化为的数据类型
* @return Java 对象
*/
public static <T> T parseObject(String text, TypeReference<T> typeReference) {
try {
return OBJECT_MAPPER.readValue(text, typeReference);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param text Json String
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(String text, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(text, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param bytes Json Byte Array
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(byte[] bytes, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(bytes, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param jsonParser Json JsonParser
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(JsonParser jsonParser, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(jsonParser, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param dataInput Json DataInput
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(DataInput dataInput, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(dataInput, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param inputStream Json InputStream
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(InputStream inputStream, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(inputStream, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param reader Json Reader
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(Reader reader, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(reader, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 反序列化, 将 Json 字符串解析为 Java 对象集合
*
* @param file Json File
* @param valueType Java 对象
* @return Java 对象集合
*/
public static <T> List<T> parseArray(File file, Class<T> valueType) {
try {
JavaType javaType = OBJECT_MAPPER.getTypeFactory().constructParametricType(ArrayList.class, valueType);
return OBJECT_MAPPER.readValue(file, javaType);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* 序列化, 将一个 Java 对象或 Java 对象集合转化为 Json 字符串
*
* @param type Java 对象引用
* @return Json 字符串
*/
public static <T> String toJsonString(T type) {
try {
return OBJECT_MAPPER.writeValueAsString(type);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
/**
* 序列化, 将一个 Java 对象转化为 Json 字符串, 并定制 Json 展示内容
*
* @param type Java 对象引用
* @param serializationView 定制现实内容的 Java 对象
* @return Json 字符串
*/
public static <T> String toJsonString(T type, Class<?> serializationView) {
try {
return OBJECT_MAPPER.writerWithView(serializationView).writeValueAsString(type);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
/**
* 美化序列化, 将一个 Java 对象或 Java 对象集合转化为 Json 字符串
*
* @param type Java 对象引用
* @return Json 字符串
*/
public static <T> String toPrettyJsonString(T type) {
try {
return OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(type);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
/**
* 美化序列化, 将一个Java 对象转化为 Json 字符串, 并定制 Json 展示内容
*
* @param type Java 对象引用
* @param serializationView 定制现实内容的Java 对象
* @return Json 字符串
*/
public static <T> String toPrettyJsonString(T type, Class<?> serializationView) {
try {
return OBJECT_MAPPER.writerWithView(serializationView).withDefaultPrettyPrinter().writeValueAsString(type);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
/**
* 序列化, 将一个 Java 对象或 Java 对象集合转化为 Json Byte Array
*
* @param type Java 对象引用
* @return Json 字符串
*/
public static <T> byte[] toJsonBytes(T type) {
try {
return OBJECT_MAPPER.writeValueAsString(type).getBytes(StandardCharsets.UTF_8);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
}
/**
* 返回一个ObjectMapper对象
*
* @return ObjectMapper
*/
public static ObjectMapper getObjectMapper() {
return OBJECT_MAPPER;
}
}
@@ -1,28 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
# 开发环境配置, 会覆盖 application-*.yml 中配置的内容
spring:
env: dev
group: ${user.name}
datasource:
url: jdbc:mysql://${MYSQL_HOST:dc3-mysql}:${MYSQL_PORT:33306}/${MYSQL_DB:dc3_data}?allowPublicKeyRetrieval=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
username: ${MYSQL_USERNAME:root}
password: ${MYSQL_PASSWORD:dc3}
logging:
level:
io.github.pnoker: DEBUG
@@ -1,22 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
# 预发布环境配置, 会覆盖 application-*.yml 中配置的内容
spring:
datasource:
url: jdbc:mysql://${MYSQL_HOST:dc3-mysql}:${MYSQL_PORT:3306}/${MYSQL_DB:dc3_data}?allowPublicKeyRetrieval=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
username: ${MYSQL_USERNAME:root}
password: ${MYSQL_PASSWORD:dc3}
@@ -1,22 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
# 生产环境配置, 会覆盖 application-*.yml 中配置的内容
spring:
datasource:
url: jdbc:mysql://${MYSQL_HOST:dc3-mysql}:${MYSQL_PORT:3306}/${MYSQL_DB:dc3_data}?allowPublicKeyRetrieval=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
username: ${MYSQL_USERNAME:root}
password: ${MYSQL_PASSWORD:dc3}
@@ -1,22 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
# 测试环境配置, 会覆盖 application-*.yml 中配置的内容
spring:
datasource:
url: jdbc:mysql://${MYSQL_HOST:dc3-mysql}:${MYSQL_PORT:3306}/${MYSQL_DB:dc3_data}?allowPublicKeyRetrieval=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
username: ${MYSQL_USERNAME:root}
password: ${MYSQL_PASSWORD:dc3}
@@ -1,42 +0,0 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# 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
#
# https://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.
#
server:
port: ${SERVER_PORT:8600}
servlet:
context-path: /ekuiper
undertow:
accesslog:
dir: dc3/logs/center/ekuiper/undertow
spring:
application:
name: @project.artifactId@
profiles:
active:
- ${NODE_ENV:dev}
driver:
custom:
url: http://10.6.0.107:9081
STREAM_URL: ${driver.custom.url}/streams
TABLE_URL: ${driver.custom.url}/tables
CONFKEY_URL: ${driver.custom.url}/metadata
RULE_URL: ${driver.custom.url}/rules
logging:
file:
name: dc3/logs/center/ekuiper/${spring.application.name}.log
-1
View File
@@ -66,7 +66,6 @@
<module>dc3-center-auth</module>
<module>dc3-center-manager</module>
<module>dc3-center-data</module>
<module>dc3-center-ekuiper</module>
</modules>
<dependencies>