|
|
|
@ -1,24 +1,9 @@ |
|
|
|
package com.huaxing.test; |
|
|
|
|
|
|
|
import com.huaxing.data.storage.domain.DataAnalysisDTO; |
|
|
|
import com.huaxing.data.storage.service.IDeviceDataQueryDfsService; |
|
|
|
import com.huaxing.data.storage.service.IDeviceDataQueryStreamService; |
|
|
|
import com.huaxing.data.storage.service.IDeviceDataStoredService; |
|
|
|
import com.huaxing.data.tablemanagement.domain.TableColumnDTO; |
|
|
|
import com.huaxing.data.tablemanagement.domain.TableDTO; |
|
|
|
import com.huaxing.data.tablemanagement.service.ITableStructureService; |
|
|
|
import com.huaxing.common.util.JacksonUtil; |
|
|
|
import com.huaxing.data.tablemanagement.service.ITableTemplateService; |
|
|
|
import com.huaxing.mqtt.processor.MqttMessageSender; |
|
|
|
import com.huaxing.common.result.ResultVo; |
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.web.bind.annotation.GetMapping; |
|
|
|
import org.springframework.web.bind.annotation.RequestMapping; |
|
|
|
import org.springframework.web.bind.annotation.RestController; |
|
|
|
|
|
|
|
import java.util.*; |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
* @ProjectName: iot-data-bridge |
|
|
|
@ -35,206 +20,206 @@ import java.util.*; |
|
|
|
@SuppressWarnings("all") |
|
|
|
public class TestController { |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private MqttMessageSender messageSender; |
|
|
|
final IDeviceDataStoredService dataStoredService; |
|
|
|
final IDeviceDataQueryDfsService dataQueryDfsService; |
|
|
|
final IDeviceDataQueryStreamService dataQueryStreamService; |
|
|
|
final ITableStructureService tableStructureService; |
|
|
|
|
|
|
|
|
|
|
|
// =============================================== 测试构造器 ======================================
|
|
|
|
public TestController(IDeviceDataStoredService dataStoredService, IDeviceDataQueryDfsService dataQueryDfsService, IDeviceDataQueryStreamService dataQueryStreamService, ITableStructureService tableStructureService, ITableTemplateService tableTemplateService) { |
|
|
|
this.dataStoredService = dataStoredService; |
|
|
|
this.dataQueryDfsService = dataQueryDfsService; |
|
|
|
this.dataQueryStreamService = dataQueryStreamService; |
|
|
|
this.tableStructureService = tableStructureService; |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// =============================================== 测试插入数据 ======================================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试数据插入 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testInsert") |
|
|
|
public void testInsert() { |
|
|
|
String tableName = "MyTable2Stream"; |
|
|
|
String sql = "INSERT INTO " + tableName + " (time, projectId, deviceId, test) VALUES (2025.01.11 00:00:00,'48', '0jZU2102_0806_0011', 'testVal')"; |
|
|
|
dataStoredService.execute(sql); |
|
|
|
log.info("SUCCESS"); |
|
|
|
} |
|
|
|
|
|
|
|
// =============================================== 测试Dfs表查询 ======================================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试Dfs表查询 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testSelectDfs") |
|
|
|
public ResultVo<List<Map<String, Object>>> testSelectDfs() { |
|
|
|
String dbPath = "dfs://ZbDB"; |
|
|
|
String sql = String.format("select * from loadTable('%s','%s')", dbPath, "ZbWaterMeter1Dfs"); |
|
|
|
return ResultVo.ok(dataQueryDfsService.selectList(sql)); |
|
|
|
} |
|
|
|
|
|
|
|
// =============================================== 测试订阅流表查询 ======================================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试订阅流表查询 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testSelectStream") |
|
|
|
public ResultVo<List<Map<String, Object>>> testSelectStream() { |
|
|
|
String sql = "select * from WaterMeterTset1Stream"; |
|
|
|
return ResultVo.ok(dataQueryStreamService.selectList(sql)); |
|
|
|
} |
|
|
|
|
|
|
|
// =============================================== 给指定的流表增加列字段 ======================================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试给指定的流表增加列字段 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testStreamAddColumn") |
|
|
|
public void testStreamAddColumn() { |
|
|
|
String tableName = "WaterMeterTsetStream"; |
|
|
|
String columnName = "test"; |
|
|
|
String columnType = "STRING"; |
|
|
|
String columnDesc = "test"; |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setTableName(tableName); |
|
|
|
tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc))); |
|
|
|
try { |
|
|
|
tableStructureService.addStreamColumns(tableDTO); |
|
|
|
log.info("SUCCESS"); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("FAIL"); |
|
|
|
e.printStackTrace(); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
// =============================================== 给指定的Dfs表增加列字段 ======================================
|
|
|
|
/** |
|
|
|
* 测试给指定的Dfs表增加列字段 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testDfsAddColumn") |
|
|
|
public void testDfsAddColumn() { |
|
|
|
String tableName = "WaterMeterTsetDfs"; |
|
|
|
String columnName = "test"; |
|
|
|
String columnType = "STRING"; |
|
|
|
String columnDesc = "test描述"; |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setTableName(tableName); |
|
|
|
tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc))); |
|
|
|
try { |
|
|
|
tableStructureService.addDfsColumns(tableDTO); |
|
|
|
log.info("SUCCESS"); |
|
|
|
} catch (Exception e) { |
|
|
|
log.error("FAIL"); |
|
|
|
e.printStackTrace(); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// ===================================== mqtt消息发送-测试数据入库 START ==============================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试发送mqtt消息 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testSendMessage") |
|
|
|
public void testSendMessage() { |
|
|
|
for (int i = 0; i < 2; i++) { |
|
|
|
List<Map<String, Object>> dataList = new ArrayList<>(); |
|
|
|
dataList.add(handleMapByIndex(i)); |
|
|
|
DataAnalysisDTO dataAnalysisDTO = DataAnalysisDTO.builder().tableName( "WaterMeterTsetStream").dataList(dataList).build(); |
|
|
|
messageSender.send("iot/test1/in-storage", JacksonUtil.objectStr(dataAnalysisDTO)); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 根据索引生成测试数据 |
|
|
|
* @param index 索引 |
|
|
|
* @return |
|
|
|
*/ |
|
|
|
private Map<String, Object> handleMapByIndex(int index) { |
|
|
|
Map<String, Object> map = new HashMap<>(); |
|
|
|
map.put("time", new Date()); |
|
|
|
map.put("projectId", "0jZU2102"); |
|
|
|
map.put("deviceId", "0jZU2102_0806_0011"); |
|
|
|
map.put("WM_WFA", 124.656 + index); |
|
|
|
map.put("WM_WFA_Unit", "m³"); |
|
|
|
map.put("test", "test"); |
|
|
|
return map; |
|
|
|
} |
|
|
|
|
|
|
|
// =============================================== 创建Dfs表 Stream流表 取消订阅 添加订阅 创建 ======================================
|
|
|
|
|
|
|
|
/** |
|
|
|
* 测试创建Dfs表 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testCreateDfsTable") |
|
|
|
public void testCreateDfsTable() { |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setDatabaseName("ZbDB"); |
|
|
|
tableDTO.setTableName("Test1"); |
|
|
|
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|
|
|
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|
|
|
tableColumnList.add(new TableColumnDTO("test1", "STRING","test1")); |
|
|
|
tableColumnList.add(new TableColumnDTO("test2", "STRING","test2")); |
|
|
|
tableDTO.setTableColumnList(tableColumnList); |
|
|
|
tableStructureService.createDfsTable(tableDTO); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 测试创建流表 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testCreateStreamTable") |
|
|
|
public void testCreateStreamTable() { |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setDatabaseName("ZbDB"); |
|
|
|
tableDTO.setTableName("Test1"); |
|
|
|
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|
|
|
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|
|
|
tableColumnList.add(new TableColumnDTO("test1", "STRING","test1")); |
|
|
|
tableColumnList.add(new TableColumnDTO("test2", "STRING","test2")); |
|
|
|
tableDTO.setTableColumnList(tableColumnList); |
|
|
|
tableStructureService.createStreamTable(tableDTO); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 取消订阅流表 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testUnsubscribeStreamTable") |
|
|
|
public void testUnsubscribeStreamTable() { |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setDatabaseName("ZbDB"); |
|
|
|
tableDTO.setTableName("Test1"); |
|
|
|
tableStructureService.unsubscribeStreamTable(tableDTO); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 订阅流表 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testSubscribeStreamTable") |
|
|
|
public void testSubscribeStreamTable() { |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setDatabaseName("ZbDB"); |
|
|
|
tableDTO.setTableName("Test1"); |
|
|
|
tableStructureService.subscribeStreamTable(tableDTO); |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 创建Dfs和Stream表并添加订阅 |
|
|
|
*/ |
|
|
|
@GetMapping(value = "/testCreateDfsAndStreamTableSubscribe") |
|
|
|
public void testCreateDfsAndStreamTableSubscribe() { |
|
|
|
TableDTO tableDTO = new TableDTO(); |
|
|
|
tableDTO.setDatabaseName("ZbDB"); |
|
|
|
tableDTO.setTableName("MyTe2"); |
|
|
|
tableDTO.setTableDesc("MyTe2测试表"); |
|
|
|
List<TableColumnDTO> tableColumnList = new ArrayList<>(); |
|
|
|
tableColumnList.add(new TableColumnDTO("test", "STRING", "test")); |
|
|
|
tableDTO.setTableColumnList(tableColumnList); |
|
|
|
tableStructureService.createDfsAndStreamTableSubscribe(tableDTO); |
|
|
|
} |
|
|
|
// @Autowired
|
|
|
|
// private MqttMessageSender messageSender;
|
|
|
|
// final IDeviceDataStoredService dataStoredService;
|
|
|
|
// final IDeviceDataQueryDfsService dataQueryDfsService;
|
|
|
|
// final IDeviceDataQueryStreamService dataQueryStreamService;
|
|
|
|
// final ITableStructureService tableStructureService;
|
|
|
|
//
|
|
|
|
//
|
|
|
|
// // =============================================== 测试构造器 ======================================
|
|
|
|
// public TestController(IDeviceDataStoredService dataStoredService, IDeviceDataQueryDfsService dataQueryDfsService, IDeviceDataQueryStreamService dataQueryStreamService, ITableStructureService tableStructureService, ISqlTemplateService tableTemplateService) {
|
|
|
|
// this.dataStoredService = dataStoredService;
|
|
|
|
// this.dataQueryDfsService = dataQueryDfsService;
|
|
|
|
// this.dataQueryStreamService = dataQueryStreamService;
|
|
|
|
// this.tableStructureService = tableStructureService;
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
//
|
|
|
|
// // =============================================== 测试插入数据 ======================================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试数据插入
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testInsert")
|
|
|
|
// public void testInsert() {
|
|
|
|
// String tableName = "MyTable2Stream";
|
|
|
|
// String sql = "INSERT INTO " + tableName + " (time, projectId, deviceId, test) VALUES (2025.01.11 00:00:00,'48', '0jZU2102_0806_0011', 'testVal')";
|
|
|
|
// dataStoredService.execute(sql);
|
|
|
|
// log.info("SUCCESS");
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// // =============================================== 测试Dfs表查询 ======================================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试Dfs表查询
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testSelectDfs")
|
|
|
|
// public ResultVo<List<Map<String, Object>>> testSelectDfs() {
|
|
|
|
// String dbPath = "dfs://ZbDB";
|
|
|
|
// String sql = String.format("select * from loadTable('%s','%s')", dbPath, "ZbWaterMeter1Dfs");
|
|
|
|
// return ResultVo.ok(dataQueryDfsService.selectList(sql));
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// // =============================================== 测试订阅流表查询 ======================================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试订阅流表查询
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testSelectStream")
|
|
|
|
// public ResultVo<List<Map<String, Object>>> testSelectStream() {
|
|
|
|
// String sql = "select * from WaterMeterTset1Stream";
|
|
|
|
// return ResultVo.ok(dataQueryStreamService.selectList(sql));
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// // =============================================== 给指定的流表增加列字段 ======================================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试给指定的流表增加列字段
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testStreamAddColumn")
|
|
|
|
// public void testStreamAddColumn() {
|
|
|
|
// String tableName = "WaterMeterTsetStream";
|
|
|
|
// String columnName = "test";
|
|
|
|
// String columnType = "STRING";
|
|
|
|
// String columnDesc = "test";
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setTableName(tableName);
|
|
|
|
// tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc)));
|
|
|
|
// try {
|
|
|
|
// tableStructureService.addStreamColumns(tableDTO);
|
|
|
|
// log.info("SUCCESS");
|
|
|
|
// } catch (Exception e) {
|
|
|
|
// log.error("FAIL");
|
|
|
|
// e.printStackTrace();
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// // =============================================== 给指定的Dfs表增加列字段 ======================================
|
|
|
|
// /**
|
|
|
|
// * 测试给指定的Dfs表增加列字段
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testDfsAddColumn")
|
|
|
|
// public void testDfsAddColumn() {
|
|
|
|
// String tableName = "WaterMeterTsetDfs";
|
|
|
|
// String columnName = "test";
|
|
|
|
// String columnType = "STRING";
|
|
|
|
// String columnDesc = "test描述";
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setTableName(tableName);
|
|
|
|
// tableDTO.setTableColumnList(Arrays.asList(new TableColumnDTO(columnName, columnType, columnDesc)));
|
|
|
|
// try {
|
|
|
|
// tableStructureService.addDfsColumns(tableDTO);
|
|
|
|
// log.info("SUCCESS");
|
|
|
|
// } catch (Exception e) {
|
|
|
|
// log.error("FAIL");
|
|
|
|
// e.printStackTrace();
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
//
|
|
|
|
// // ===================================== mqtt消息发送-测试数据入库 START ==============================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试发送mqtt消息
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testSendMessage")
|
|
|
|
// public void testSendMessage() {
|
|
|
|
// for (int i = 0; i < 2; i++) {
|
|
|
|
// List<Map<String, Object>> dataList = new ArrayList<>();
|
|
|
|
// dataList.add(handleMapByIndex(i));
|
|
|
|
// DataAnalysisDTO dataAnalysisDTO = DataAnalysisDTO.builder().tableName( "WaterMeterTsetStream").dataList(dataList).build();
|
|
|
|
// messageSender.send("iot/test1/in-storage", JacksonUtil.objectStr(dataAnalysisDTO));
|
|
|
|
// }
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 根据索引生成测试数据
|
|
|
|
// * @param index 索引
|
|
|
|
// * @return
|
|
|
|
// */
|
|
|
|
// private Map<String, Object> handleMapByIndex(int index) {
|
|
|
|
// Map<String, Object> map = new HashMap<>();
|
|
|
|
// map.put("time", new Date());
|
|
|
|
// map.put("projectId", "0jZU2102");
|
|
|
|
// map.put("deviceId", "0jZU2102_0806_0011");
|
|
|
|
// map.put("WM_WFA", 124.656 + index);
|
|
|
|
// map.put("WM_WFA_Unit", "m³");
|
|
|
|
// map.put("test", "test");
|
|
|
|
// return map;
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// // =============================================== 创建Dfs表 Stream流表 取消订阅 添加订阅 创建 ======================================
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试创建Dfs表
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testCreateDfsTable")
|
|
|
|
// public void testCreateDfsTable() {
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setDatabaseName("ZbDB");
|
|
|
|
// tableDTO.setTableName("Test1");
|
|
|
|
// List<TableColumnDTO> tableColumnList = new ArrayList<>();
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test", "STRING", "test"));
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test1", "STRING","test1"));
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test2", "STRING","test2"));
|
|
|
|
// tableDTO.setTableColumnList(tableColumnList);
|
|
|
|
// tableStructureService.createDfsTable(tableDTO);
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 测试创建流表
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testCreateStreamTable")
|
|
|
|
// public void testCreateStreamTable() {
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setDatabaseName("ZbDB");
|
|
|
|
// tableDTO.setTableName("Test1");
|
|
|
|
// List<TableColumnDTO> tableColumnList = new ArrayList<>();
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test", "STRING", "test"));
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test1", "STRING","test1"));
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test2", "STRING","test2"));
|
|
|
|
// tableDTO.setTableColumnList(tableColumnList);
|
|
|
|
// tableStructureService.createStreamTable(tableDTO);
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 取消订阅流表
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testUnsubscribeStreamTable")
|
|
|
|
// public void testUnsubscribeStreamTable() {
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setDatabaseName("ZbDB");
|
|
|
|
// tableDTO.setTableName("Test1");
|
|
|
|
// tableStructureService.unsubscribeStreamTable(tableDTO);
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 订阅流表
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testSubscribeStreamTable")
|
|
|
|
// public void testSubscribeStreamTable() {
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setDatabaseName("ZbDB");
|
|
|
|
// tableDTO.setTableName("Test1");
|
|
|
|
// tableStructureService.subscribeStreamTable(tableDTO);
|
|
|
|
// }
|
|
|
|
//
|
|
|
|
// /**
|
|
|
|
// * 创建Dfs和Stream表并添加订阅
|
|
|
|
// */
|
|
|
|
// @GetMapping(value = "/testCreateDfsAndStreamTableSubscribe")
|
|
|
|
// public void testCreateDfsAndStreamTableSubscribe() {
|
|
|
|
// TableDTO tableDTO = new TableDTO();
|
|
|
|
// tableDTO.setDatabaseName("ZbDB");
|
|
|
|
// tableDTO.setTableName("MyTe2");
|
|
|
|
// tableDTO.setTableDesc("MyTe2测试表");
|
|
|
|
// List<TableColumnDTO> tableColumnList = new ArrayList<>();
|
|
|
|
// tableColumnList.add(new TableColumnDTO("test", "STRING", "test"));
|
|
|
|
// tableDTO.setTableColumnList(tableColumnList);
|
|
|
|
// tableStructureService.createDfsAndStreamTableSubscribe(tableDTO);
|
|
|
|
// }
|
|
|
|
|
|
|
|
|
|
|
|
} |
|
|
|
|
swordmeng
3 weeks ago