admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加EG验证

Browse Source
This commit is contained in:
zren25
2025-05-30 18:08:03 +08:00
committed by lxu75
parent 8adc0e0575
commit 5a6704bab4
13 changed files with 1433 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/HuaWeiEGConfig.java Normal file
View File

@@ -0,0 +1,59 @@
package com.volvo.ai.analytic.center.config;
import com.huaweicloud.sdk.core.auth.BasicCredentials;
import com.huaweicloud.sdk.core.auth.ICredential;
import com.huaweicloud.sdk.core.http.HttpConfig;
import com.huaweicloud.sdk.eg.v1.EgClient;
import com.huaweicloud.sdk.eg.v1.region.EgRegion;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* HuaWeiConfig 配置类
*/
@Data
@Configuration
@RefreshScope
public class HuaWeiEGConfig {
/**
* 创建 EventProperties 对象
* @return EventProperties 对象
*/
@Value("${huawei.cloud.EG.ak}")
private String ak;
@Value("${huawei.cloud.EG.sk}")
private String sk;
@Value("${huawei.cloud.EG.projectId}")
private String projectId;
@Value("${huawei.cloud.EG.endpoint}")
private String endpoint;
@Value("${huawei.cloud.EG.region}")
private String region;
/**
* 创建 EgClient 对象
* @return EgClient 对象
*/
@Bean
public EgClient egClient() {
// 创建 BasicCredentials 对象，并设置 accessKey 和 secretKey
ICredential auth = new BasicCredentials()
.withAk(ak)
.withSk(sk);
// 创建 HttpConfig 对象
HttpConfig httpConfig = HttpConfig.getDefaultHttpConfig();
// 创建 EgClient 对象，并设置认证信息、Http配置和区域
return EgClient.newBuilder()
.withCredential(auth)
.withHttpConfig(httpConfig)
.withRegion(EgRegion.valueOf(region))
.build();
}
}

179
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/EGEventSubscriptionController.java Normal file
View File

@@ -0,0 +1,179 @@
package com.volvo.ai.analytic.center.controller;
import com.huaweicloud.sdk.eg.v1.model.*;
import com.volvo.ai.analytic.center.dto.event.ListSubscriptionsRequestDto;
import com.volvo.ai.analytic.center.dto.event.SendEventDto;
import com.volvo.ai.analytic.center.dto.event.SubscriptionCreateReqDto;
import com.volvo.ai.analytic.center.dto.event.SubscriptionOperateReqDto;
import com.volvo.ai.analytic.center.dto.resp.ResultDTO;
import com.volvo.ai.analytic.center.service.HuaWeiEGService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
/**
* 事件订阅管理
*/
@Tag(name = "/event", description = "事件订阅管理")
@RestController
@RequestMapping("/event")
@Slf4j
public class EGEventSubscriptionController {
@Resource
HuaWeiEGService huaWeiService;
/**
* 发布事件到事件通道
*/
@Operation(summary = "发布事件到事件通道")
@PostMapping("/sendEvent")
public ResultDTO<PutEventsResponse> sendEvent(@RequestBody CloudEvents req, @RequestParam("channel") String channel) throws ExecutionException {
return ResultDTO.success(huaWeiService.sendEvent(req, "channel"));
}
/**
* 创建事件订阅
*
* @param req req
* @return 事件订阅ID
*/
@Operation(summary = "创建事件订阅")
@PostMapping("/subscription")
public ResultDTO<CreateSubscriptionResponse> createSubscription(@RequestBody SubscriptionCreateReqDto req) {
return ResultDTO.success(huaWeiService.createSubscription(req));
}
/**
* 创建事件订阅目标
*
* @param req req
* @return 事件订阅目标ID
*/
@Operation(summary = "创建事件订阅目标")
@PostMapping("/subscriptionTarget")
public ResultDTO<CreateSubscriptionTargetResponse> createSubscriptionTarget(@RequestBody CreateSubscriptionTargetRequest req) {
return ResultDTO.success(huaWeiService.createSubscriptionTarget(req));
}
/**
* 删除事件订阅
*
* @param subscriptionId 事件订阅ID
* @return
*/
@Operation(summary = "删除事件订阅")
@DeleteMapping("/subscription/{subscriptionId}")
public ResultDTO deleteSubscription(@PathVariable("subscriptionId") String subscriptionId) {
huaWeiService.deleteSubscription(subscriptionId);
return ResultDTO.success();
}
/**
* 删除事件订阅目标
*
* @param subscriptionId 事件订阅ID
* @param targetId 事件订阅目标ID
* @return
*/
@Operation(summary = "删除事件订阅目标")
@DeleteMapping("/subscriptionTarget/{subscriptionId}/{targetId}")
public ResultDTO deleteSubscriptionTarget(@PathVariable("subscriptionId") String subscriptionId, @PathVariable("targetId") String targetId) {
huaWeiService.deleteSubscriptionTarget(subscriptionId, targetId);
return ResultDTO.success();
}
/**
* 查询事件订阅列表
*
* @param req req
* @return
*/
@Operation(summary = "查询事件订阅列表")
@PostMapping("/subscription/list")
public ResultDTO<ListSubscriptionsResponse> listSubscriptions(@RequestBody ListSubscriptionsRequestDto req) {
return ResultDTO.success(huaWeiService.listSubscriptions(req));
}
/**
* 启用禁用事件订阅
*
* @param req req
* @return
*/
@Operation(summary = "启用禁用事件订阅")
@PostMapping("/subscription/enabled")
public ResultDTO operateSubscription(@RequestBody SubscriptionOperateReqDto req) {
huaWeiService.operateSubscription(req);
return ResultDTO.success();
}
/**
* 查询事件订阅详情
*
* @param subscriptionId 事件订阅ID
* @return
*/
@Operation(summary = "查询事件订阅详情")
@GetMapping("/subscription/{subscriptionId}")
public ResultDTO<ShowDetailOfSubscriptionResponse> showDetailOfSubscription(@PathVariable("subscriptionId") String subscriptionId) {
return ResultDTO.success(huaWeiService.showDetailOfSubscription(subscriptionId));
}
/**
* 查询事件订阅目标详情
*
* @param subscriptionId 事件订阅ID
* @param targetId 事件订阅目标ID
* @return
*/
@Operation(summary = "查询事件订阅目标详情")
@GetMapping("/subscriptionTarget/{subscriptionId}/{targetId}")
public ResultDTO<ShowDetailOfSubscriptionTargetResponse> showDetailOfSubscriptionTarget(@PathVariable("subscriptionId") String subscriptionId,
@PathVariable("targetId") String targetId) {
return ResultDTO.success(huaWeiService.showDetailOfSubscriptionTarget(subscriptionId, targetId));
}
/**
* 更新事件订阅
*
* @param req req
* @return
*/
@Operation(summary = "更新事件订阅")
@PutMapping("/subscription")
public ResultDTO updateSubscription(@RequestBody UpdateSubscriptionRequest req) {
huaWeiService.updateSubscription(req);
return ResultDTO.success();
}
/**
* 更新事件订阅源
*
* @param req req
* @return
*/
@Operation(summary = "更新事件订阅源")
@PutMapping("/subscriptionSource")
public ResultDTO updateSubscriptionSource(@RequestBody UpdateSubscriptionSourceRequest req) {
huaWeiService.updateSubscriptionSource(req);
return ResultDTO.success();
}
/**
* 更新事件订阅目标
*
* @param req req
* @return
*/
@Operation(summary = "更新事件订阅目标")
@PutMapping("/subscriptionTarget")
public ResultDTO updateSubscriptionTarget(@RequestBody UpdateSubscriptionTargetRequest req) {
huaWeiService.updateSubscriptionTarget(req);
return ResultDTO.success();
}
}

285
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/CloudService.java Normal file
View File

@@ -0,0 +1,285 @@
package com.volvo.ai.analytic.center.service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.huaweicloud.sdk.core.exception.ClientRequestException;
import com.huaweicloud.sdk.eg.v1.model.*;
import com.volvo.ai.analytic.center.config.HuaWeiEGConfig;
import com.volvo.ai.analytic.center.dto.event.SendEventDto;
import com.volvo.ai.analytic.center.dto.event.SourceChannelDto;
import com.volvo.ai.analytic.center.dto.resp.ResultDTO;
import com.volvo.ai.analytic.center.utils.Signer;
import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.data.PojoCloudEventData;
import io.cloudevents.jackson.JsonFormat;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@Slf4j
@Service
public class CloudService implements InitializingBean {
@Resource
RestTemplate restTemplate;
@Resource
HuaWeiEGService huaWeiService;
@Resource
RedisTemplate<Object, Object> redisTemplate;
@Resource
HuaWeiEGConfig huaWeiEGConfig;
private static final String HA_WEI_REQUEST = "hwReq={}";
private static final String HA_WEI_RESPONSE = "hwRes={}";
private static final String SEND_EVENT_MSG = "发布事件到事件通道异常";
private static final String HTTPS = "https://";
private static final String V1 = "/v1/";
private static final String CHANNELS = "/channels/";
private static final String EVENTS = "/events";
private static final String BODY_PRE = "{\"events\":[";
private static final String BODY_END = "]}";
private static final String APPLICATION = "APPLICATION";
private static final String COMMA = ",";
/**
* 事件通道缓存
*/
private static final String SOURCE_KEY = "eSource";
/**
* 发送事件
*
* @param req 发送事件的负载内容
* @return 发送结果
* @throws ExecutionException 如果执行异常
*/
public ResultDTO<String> sendEvent(SendEventDto<Object> req) throws ExecutionException {
String eventId = UUID.randomUUID().toString();
return send(req, eventId);
}
/**
* 发送事件
*
* @param req 发送事件的请求数据
* @param eventId 事件ID
* @return 发送事件的响应数据
* @throws ExecutionException 如果执行异常
*/
private ResultDTO<String> send(SendEventDto<Object> req, String eventId) throws ExecutionException {
// 根据sourceId或sourceName获取通道数据
SourceChannelDto dto = req.getSource().contains("-") ? getChannelBySourceId(req.getSource()) : getChannelBySourceName(req.getSource());
// 构建CloudEvent对象
CloudEvent event = CloudEventBuilder.v1().withId(eventId)
.withSource(URI.create(dto.getSourceName())).withType(StringUtils.isEmpty(req.getType()) ? APPLICATION : req.getType())
.withData(MediaType.APPLICATION_JSON_VALUE, PojoCloudEventData.wrap(req.getData(), JSON::toJSONBytes))
.withSubject(req.getSubject()).withTime(OffsetDateTime.now()).build();
// 将CloudEvent对象转换为JSON字符串
String b = new String(new JsonFormat().serialize(event), StandardCharsets.UTF_8);
String body = BODY_PRE + b + BODY_END;
return getResponseDTO(eventId, dto, body);
}
/**
* 发送事件
*
* @param req 发送事件的请求数据
* @return 发送事件的响应数据
* @throws ExecutionException 如果执行异常
*/
private ResultDTO<String> sendList(SendEventDto<List<Object>> req) throws ExecutionException {
// 根据sourceId或sourceName获取通道数据
SourceChannelDto dto = req.getSource().contains("-") ? getChannelBySourceId(req.getSource()) : getChannelBySourceName(req.getSource());
List<String> eventIdList = new ArrayList<>();
List<String> eventList = new ArrayList<>();
for (Object data : req.getData()) {
String eventId = UUID.randomUUID().toString();
eventIdList.add(eventId);
// 构建CloudEvent对象
CloudEvent event = CloudEventBuilder.v1().withId(eventId)
.withSource(URI.create(dto.getSourceName())).withType(StringUtils.isEmpty(req.getType()) ? APPLICATION : req.getType())
.withData(MediaType.APPLICATION_JSON_VALUE, PojoCloudEventData.wrap(data, JSON::toJSONBytes))
.withSubject(req.getSubject()).withTime(OffsetDateTime.now()).build();
// 将CloudEvent对象转换为JSON字符串
eventList.add(new String(new JsonFormat().serialize(event), StandardCharsets.UTF_8));
}
String eId = String.join(COMMA, eventIdList);
String body = BODY_PRE + String.join(COMMA, eventList) + BODY_END;
return getResponseDTO(eId, dto, body);
}
/**
* 发送请求
*
* @param eventId
* @param dto
* @param body
* @return
*/
private ResultDTO getResponseDTO(String eventId, SourceChannelDto dto, String body) {
String url = HTTPS + huaWeiEGConfig.getEndpoint() + V1 +
huaWeiEGConfig.getProjectId() + CHANNELS + dto.getChannelId() + EVENTS;
Map<String, String> headers = null;
try {
// 对请求进行签名
headers = Signer.sign(huaWeiEGConfig.getAk(), huaWeiEGConfig.getSk(), url, body, huaWeiEGConfig.getEndpoint());
} catch (Exception e) {
log.error("sign", e);
}
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setAll(headers);
log.info(HA_WEI_REQUEST, body);
// 发送POST请求
String resp = restTemplate.postForEntity(url, new HttpEntity<>(body.getBytes(StandardCharsets.UTF_8), httpHeaders), String.class).getBody();
log.info(HA_WEI_RESPONSE, resp);
PutEventsResponse response1 = JSON.parseObject(resp, new TypeReference<PutEventsResponse>() {
});
if (response1.getFailedCount() != 0) {
List<PutEventsRespEvents> events = response1.getEvents();
List<String> eventIds = new ArrayList<>();
List<String> errorMsgList = new ArrayList<>();
for (PutEventsRespEvents response : events) {
eventIds.add(response.getEventId());
errorMsgList.add(response.getEventId() + ":" + response.getErrorMsg());
}
ResultDTO res =new ResultDTO();
res.setErrMsg(String.join(COMMA, errorMsgList));
res.setData(String.join(COMMA, eventIds));
res.setReturnCode("500");
return res;
}
return ResultDTO.success(eventId);
}
/**
* 根据事件源ID查询事件通道详情，并返回事件通道信息
*
* @param sourceId 事件源ID
* @return 事件通道信息
* @throws ExecutionException 如果执行查询操作失败，抛出执行异常
*/
@NotNull
private SourceChannelDto getChannelBySourceId(String sourceId) throws ExecutionException {
//根据事件源ID查询详情
ShowDetailOfEventSourceResponse response;
try {
response = huaWeiService.showDetailOfEventSource(sourceId);
} catch (ClientRequestException e) {
throw e;
}
if (response.getHttpStatusCode() != 200) {
log.info("根据事件源ID未查询到事件通道");
}
SourceChannelDto dto = new SourceChannelDto(response.getChannelId(), response.getName());
redisTemplate.opsForHash().put(SOURCE_KEY, sourceId, dto);
return dto;
}
/**
* 根据事件源名称查询事件通道
*
* @param source 事件源名称
* @return 事件通道信息
* @throws ExecutionException 如果执行过程中发生异常，抛出执行异常
*/
@NotNull
private SourceChannelDto getChannelBySourceName(String source) throws ExecutionException {
ListEventSourcesRequest sourcesRequest = new ListEventSourcesRequest();
sourcesRequest.setName(source);
sourcesRequest.setLimit(1);
sourcesRequest.setOffset(0);
//根据事件源名称查询事件通道
ListEventSourcesResponse res = huaWeiService.listEventSources(sourcesRequest);
if (res.getTotal() == 1) {
CustomizeSourceInfo sourceInfo = res.getItems().get(0);
SourceChannelDto dto = new SourceChannelDto(sourceInfo.getChannelId(), sourceInfo.getName());
redisTemplate.opsForHash().put(SOURCE_KEY, source, dto);
return dto;
} else {
log.info("根据事件源名称未查询到事件通道");
}
return null;
}
/**
* 在属性设置之后执行的操作。
*
* @throws Exception 如果在属性设置过程中发生异常
*/
@Override
public void afterPropertiesSet() throws Exception {
// 创建ListEventSourcesRequest对象
ListEventSourcesRequest request = new ListEventSourcesRequest();
// 设置请求的Limit属性为500
request.setLimit(500);
// 设置请求的FuzzyName属性为"source"
request.setFuzzyName("source");
// 调用huaWeiService的listEventSources方法，获取ListEventSourcesResponse对象
ListEventSourcesResponse response = huaWeiService.listEventSources(request);
// 如果response的HttpStatusCode属性不等于200，则抛出BussinessException异常
if (response.getHttpStatusCode() != 200) {
log.info("查询事件源列表异常");
}
// 删除redis中的SOURCE_KEY键值对
redisTemplate.delete(SOURCE_KEY);
// 如果response的Items属性不为空
if (!CollectionUtils.isEmpty(response.getItems())) {
// 遍历response的Items属性
for (CustomizeSourceInfo info : response.getItems()) {
// 创建SourceChannelDto对象
SourceChannelDto dto = new SourceChannelDto(info.getChannelId(), info.getName());
// 将dto存储到redis的SOURCE_KEY键的hash表中，使用info.getId()作为键，dto作为值
redisTemplate.opsForHash().put(SOURCE_KEY, info.getId(), dto);
// 将dto存储到redis的SOURCE_KEY键的hash表中，使用info.getName()作为键，dto作为值
redisTemplate.opsForHash().put(SOURCE_KEY, info.getName(), dto);
}
}
}
}

493
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/HuaWeiEGService.java Normal file
View File

@@ -0,0 +1,493 @@
package com.volvo.ai.analytic.center.service;
import com.alibaba.fastjson.JSON;
import com.huaweicloud.sdk.eg.v1.EgClient;
import com.huaweicloud.sdk.eg.v1.model.*;
import com.volvo.ai.analytic.center.dto.event.ListSubscriptionsRequestDto;
import com.volvo.ai.analytic.center.dto.event.SubscriptionCreateReqDto;
import com.volvo.ai.analytic.center.dto.event.SubscriptionOperateReqDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
@Component
@Slf4j
public class HuaWeiEGService {
private static final String HA_WEI_REQUEST = "hwReq={}";
private static final String HA_WEI_RESPONSE = "hwRes={}";
@Resource
EgClient egClient;
/**
* 发布事件到事件通道 sdk
*
* @param event event
* @param channel channel
*/
public PutEventsResponse sendEvent(CloudEvents event, String channel) {
PutEventsRequest request = new PutEventsRequest();
request.withChannelId(channel);
List<CloudEvents> eventsList = new ArrayList<>();
eventsList.add(event);
PutEventsReq body = new PutEventsReq();
body.setEvents(eventsList);
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
PutEventsResponse response = egClient.putEvents(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建事件订阅
*/
public CreateSubscriptionResponse createSubscription(SubscriptionCreateReqDto reqDto) {
CreateSubscriptionRequest request = new CreateSubscriptionRequest();
SubscriptionCreateReq body = new SubscriptionCreateReq();
BeanUtils.copyProperties(reqDto, body);
List<SubscriptionSource> sources = new ArrayList<>();
sources.add(reqDto.getSources());
body.setSources(sources);
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateSubscriptionResponse response = egClient.createSubscription(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建事件订阅目标
*/
public CreateSubscriptionTargetResponse createSubscriptionTarget(CreateSubscriptionTargetRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateSubscriptionTargetResponse response = egClient.createSubscriptionTarget(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除事件订阅
*/
public void deleteSubscription(String subscriptionId) {
DeleteSubscriptionRequest request = new DeleteSubscriptionRequest();
request.setSubscriptionId(subscriptionId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteSubscriptionResponse response = egClient.deleteSubscription(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
}
/**
* 删除事件订阅目标
*/
public void deleteSubscriptionTarget(String subscriptionId, String targetId) {
DeleteSubscriptionTargetRequest request = new DeleteSubscriptionTargetRequest();
request.setSubscriptionId(subscriptionId);
request.setTargetId(targetId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteSubscriptionTargetResponse response = egClient.deleteSubscriptionTarget(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
}
/**
* 查询事件订阅列表
*/
public ListSubscriptionsResponse listSubscriptions(ListSubscriptionsRequestDto requestDto) {
ListSubscriptionsRequest request = new ListSubscriptionsRequest();
BeanUtils.copyProperties(requestDto, request);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListSubscriptionsResponse response = egClient.listSubscriptions(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 操作事件订阅
*/
public OperateSubscriptionResponse operateSubscription(SubscriptionOperateReqDto req) {
OperateSubscriptionRequest request = new OperateSubscriptionRequest();
SubscriptionOperateReq body = new SubscriptionOperateReq();
List<String> subscriptionIds = new ArrayList<>();
subscriptionIds.add(req.getSubscriptionId());
body.setSubscriptionIds(subscriptionIds);
body.setOperation(req.getStatus() == 1 ? SubscriptionOperateReq.OperationEnum.ENABLE : SubscriptionOperateReq.OperationEnum.DISABLE);
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
OperateSubscriptionResponse response = egClient.operateSubscription(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件订阅详情
*/
public ShowDetailOfSubscriptionResponse showDetailOfSubscription(String subscriptionId) {
ShowDetailOfSubscriptionRequest request = new ShowDetailOfSubscriptionRequest();
request.setSubscriptionId(subscriptionId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfSubscriptionResponse response = egClient.showDetailOfSubscription(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件订阅目标详情
*/
public ShowDetailOfSubscriptionTargetResponse showDetailOfSubscriptionTarget(String subscriptionId, String targetId) {
ShowDetailOfSubscriptionTargetRequest request = new ShowDetailOfSubscriptionTargetRequest();
request.setSubscriptionId(subscriptionId);
request.setTargetId(targetId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfSubscriptionTargetResponse response = egClient.showDetailOfSubscriptionTarget(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新事件订阅
*/
public UpdateSubscriptionResponse updateSubscription(UpdateSubscriptionRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateSubscriptionResponse response = egClient.updateSubscription(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新事件订阅源
*/
public UpdateSubscriptionSourceResponse updateSubscriptionSource(UpdateSubscriptionSourceRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateSubscriptionSourceResponse response = egClient.updateSubscriptionSource(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新事件订阅目标
*/
public UpdateSubscriptionTargetResponse updateSubscriptionTarget(UpdateSubscriptionTargetRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateSubscriptionTargetResponse response = egClient.updateSubscriptionTarget(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建自定义事件通道
*/
public CreateChannelResponse createChannel(ChannelCreateReq body) {
CreateChannelRequest request = new CreateChannelRequest();
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateChannelResponse response = egClient.createChannel(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除自定义事件通道
*/
public DeleteChannelResponse deleteChannel(String channelId) {
DeleteChannelRequest request = new DeleteChannelRequest();
request.withChannelId(channelId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteChannelResponse response = egClient.deleteChannel(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件通道列表
*/
public ListChannelsResponse listChannels(ListChannelsRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListChannelsResponse response = egClient.listChannels(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件通道详情
*/
public ShowDetailOfChannelResponse showDetailOfChannel(String channelId) {
ShowDetailOfChannelRequest request = new ShowDetailOfChannelRequest();
request.withChannelId(channelId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfChannelResponse response = egClient.showDetailOfChannel(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新自定义事件通道
*/
public UpdateChannelResponse updateChannel(UpdateChannelRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateChannelResponse response = egClient.updateChannel(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建自定义事件源
*/
public CreateEventSourceResponse createEventSource(CustomizeSourceCreateReq body) {
CreateEventSourceRequest request = new CreateEventSourceRequest();
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateEventSourceResponse response = egClient.createEventSource(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除自定义事件源
*/
public DeleteEventSourceResponse deleteEventSource(String sourceId) {
DeleteEventSourceRequest request = new DeleteEventSourceRequest();
request.withSourceId(sourceId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteEventSourceResponse response = egClient.deleteEventSource(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件源列表
*/
public ListEventSourcesResponse listEventSources(ListEventSourcesRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListEventSourcesResponse response = egClient.listEventSources(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件源详情
*/
public ShowDetailOfEventSourceResponse showDetailOfEventSource(String sourceId) {
ShowDetailOfEventSourceRequest request = new ShowDetailOfEventSourceRequest();
request.withSourceId(sourceId);
return egClient.showDetailOfEventSource(request);
}
/**
* 更新自定义事件源
*/
public UpdateEventSourceResponse updateEventSource(UpdateEventSourceRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateEventSourceResponse response = egClient.updateEventSource(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建目标连接
*/
public CreateConnectionResponse createConnection(ConnectionCreateReq body) {
CreateConnectionRequest request = new CreateConnectionRequest();
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateConnectionResponse response = egClient.createConnection(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除目标连接
*/
public DeleteConnectionResponse deleteConnection(String connectionId) {
DeleteConnectionRequest request = new DeleteConnectionRequest();
request.withConnectionId(connectionId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteConnectionResponse response = egClient.deleteConnection(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询目标连接列表
*/
public ListConnectionsResponse listConnections(ListConnectionsRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListConnectionsResponse response = egClient.listConnections(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询目标连接详情
*/
public ShowDetailOfConnectionResponse showDetailOfConnection(String connectionId) {
ShowDetailOfConnectionRequest request = new ShowDetailOfConnectionRequest();
request.withConnectionId(connectionId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfConnectionResponse response = egClient.showDetailOfConnection(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新目标连接
*/
public UpdateConnectionResponse updateConnection(UpdateConnectionRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateConnectionResponse response = egClient.updateConnection(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件目标分类
*/
public ListEventTargetResponse listEventTarget(ListEventTargetRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListEventTargetResponse response = egClient.listEventTarget(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建自定义事件模型
*/
public CreateEventSchemaResponse createEventSchema(CustomizeSchemaCreateReq body) {
CreateEventSchemaRequest request = new CreateEventSchemaRequest();
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateEventSchemaResponse response = egClient.createEventSchema(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 创建自定义事件模型
*/
public CreateEventSchemaVersionResponse createEventSchemaVersion(CreateEventSchemaVersionRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
CreateEventSchemaVersionResponse response = egClient.createEventSchemaVersion(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除事件模型
*/
public DeleteEventSchemaResponse deleteEventSchema(String schemaId) {
DeleteEventSchemaRequest request = new DeleteEventSchemaRequest();
request.withSchemaId(schemaId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteEventSchemaResponse response = egClient.deleteEventSchema(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 删除事件模型
*/
public DeleteEventSchemaVersionResponse deleteEventSchemaVersion(String schemaId, Integer version) {
DeleteEventSchemaVersionRequest request = new DeleteEventSchemaVersionRequest();
request.withSchemaId(schemaId);
request.withVersion(version);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DeleteEventSchemaVersionResponse response = egClient.deleteEventSchemaVersion(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 事件模型自动发现
*/
public DiscoverEventSchemaFromDataResponse discoverEventSchemaFromData(DiscoverEventSchemaFromDataReq body) {
DiscoverEventSchemaFromDataRequest request = new DiscoverEventSchemaFromDataRequest();
request.withBody(body);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
DiscoverEventSchemaFromDataResponse response = egClient.discoverEventSchemaFromData(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件模型列表
*/
public ListEventSchemaResponse listEventSchema(ListEventSchemaRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListEventSchemaResponse response = egClient.listEventSchema(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件模型版本列表
*/
public ListEventSchemaVersionsResponse listEventSchemaVersions(ListEventSchemaVersionsRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ListEventSchemaVersionsResponse response = egClient.listEventSchemaVersions(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件模型详情
*/
public ShowDetailOfEventSchemaResponse showDetailOfEventSchema(String schemaId) {
ShowDetailOfEventSchemaRequest request = new ShowDetailOfEventSchemaRequest();
request.withSchemaId(schemaId);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfEventSchemaResponse response = egClient.showDetailOfEventSchema(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 查询事件模型版本详情
*/
public ShowDetailOfEventSchemaVersionResponse showDetailOfEventSchemaVersion(String schemaId, Integer version) {
ShowDetailOfEventSchemaVersionRequest request = new ShowDetailOfEventSchemaVersionRequest();
request.withSchemaId(schemaId);
request.withVersion(version);
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
ShowDetailOfEventSchemaVersionResponse response = egClient.showDetailOfEventSchemaVersion(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
/**
* 更新自定义事件模型
*/
public UpdateEventSchemaResponse updateEventSchema(UpdateEventSchemaRequest request) {
log.info(HA_WEI_REQUEST, JSON.toJSONString(request));
UpdateEventSchemaResponse response = egClient.updateEventSchema(request);
log.info(HA_WEI_RESPONSE, JSON.toJSONString(response));
return response;
}
}

147
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/Signer.java Normal file
View File

@@ -0,0 +1,147 @@
package com.volvo.ai.analytic.center.utils;
import org.apache.commons.codec.digest.DigestUtils;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
public final class Signer {
private static final String AUTHORIZATION = "Authorization";
private static final String SDK_HMAC_SHA256 = "SDK-HMAC-SHA256";
private static final String LINE = "\n";
private static final String ACCESS = " Access=";
private static final String SIGNED_HEADERS = ", SignedHeaders=";
private static final String SIGNATURE = ", Signature=";
private static final String SLASH = "/";
private static final String HMAC_SHA256 = "HmacSHA256";
private static final String POST = "POST";
private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'").withZone(ZoneId.of("UTC"));
private static final String[] HEADERS = new String[]{"Host", "X-Sdk-Date"};
private static final String SIGNED_HEADERS_STR = getSignedHeadersString(HEADERS);
private Signer() {
}
public static Map<String, String> sign(String key, String secret, String url, String body, String host) throws NoSuchAlgorithmException, InvalidKeyException {
String singerDate = DATE_TIME_FORMATTER.format(ZonedDateTime.now());
Map<String, String> headers = new HashMap<>(8);
headers.put(HEADERS[0], host);
headers.put(HEADERS[1], singerDate);
String contentSha256 = toHex(hash(body));
String canonicalRequest = createCanonicalRequest(headers, getPath(url), POST, HEADERS, contentSha256, SIGNED_HEADERS_STR);
String stringToSign = createStringToSign(canonicalRequest, singerDate);
byte[] signingKey = secret.getBytes(StandardCharsets.UTF_8);
byte[] signature = sign(stringToSign.getBytes(StandardCharsets.UTF_8), signingKey, HMAC_SHA256);
headers.put(AUTHORIZATION, buildAuthorizationHeader(SIGNED_HEADERS_STR, signature, key));
return headers;
}
private static String getPath(String url) {
int i = url.indexOf("://");
if (i >= 0) {
url = url.substring(i + 3);
}
i = url.indexOf(47);
return i >= 0 ? url.substring(i) : SLASH;
}
private static String buildAuthorizationHeader(String signedHeadersStr, byte[] signature, String accessKey) {
return new StringBuilder(SDK_HMAC_SHA256).append(ACCESS).append(accessKey).append(SIGNED_HEADERS).append(signedHeadersStr).append(SIGNATURE).append(toHex(signature)).toString();
}
private static byte[] hash(String text) {
MessageDigest md = DigestUtils.getSha256Digest();
md.update(text.getBytes(StandardCharsets.UTF_8));
return md.digest();
}
private static byte[] sign(byte[] data, byte[] key, String algorithm) throws NoSuchAlgorithmException, InvalidKeyException {
Mac mac = Mac.getInstance(algorithm);
mac.init(new SecretKeySpec(key, algorithm));
return mac.doFinal(data);
}
private static String createCanonicalRequest(Map<String, String> requestHeaders, String path, String method,
String[] signedHeaders, String contentSha256, String signedHeadersStr) {
return new StringBuilder(method).append(LINE).append(path)
.append((SLASH)).append(LINE).append(LINE).append(getCanonicalizedHeaderString(requestHeaders, signedHeaders))
.append(LINE).append(signedHeadersStr).append(LINE).append(contentSha256).toString();
}
private static String createStringToSign(String canonicalRequest, String singerDate) {
return new StringBuilder(SDK_HMAC_SHA256).append(LINE).append(singerDate).append(LINE).append(toHex(hash(canonicalRequest))).toString();
}
private static String getCanonicalizedHeaderString(Map<String, String> requestHeaders, String[] signedHeaders) {
StringBuilder buffer = new StringBuilder();
String[] var5 = signedHeaders;
int var6 = signedHeaders.length;
for (int var7 = 0; var7 < var6; ++var7) {
String header = var5[var7];
String key = header.toLowerCase();
String value = requestHeaders.get(header);
buffer.append(key).append(":");
if (value != null) {
buffer.append(value.trim());
}
buffer.append(LINE);
}
return buffer.toString();
}
private static String getSignedHeadersString(String[] signedHeaders) {
StringBuilder buffer = new StringBuilder();
String[] var3 = signedHeaders;
int var4 = signedHeaders.length;
for (int var5 = 0; var5 < var4; ++var5) {
String header = var3[var5];
if (buffer.length() > 0) {
buffer.append(";");
}
buffer.append(header.toLowerCase());
}
return buffer.toString();
}
private static String toHex(byte[] data) {
StringBuilder sb = new StringBuilder(data.length * 2);
byte[] var2 = data;
int var3 = data.length;
for (int var4 = 0; var4 < var3; ++var4) {
byte b = var2[var4];
String hex = Integer.toHexString(b);
if (hex.length() == 1) {
sb.append("0");
} else if (hex.length() == 8) {
hex = hex.substring(6);
}
sb.append(hex);
}
return sb.toString().toLowerCase(Locale.getDefault());
}
}