1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
| IntegrationFlow
.from(ChannelName.INBOUND_OSD)
.filter(Message.class,source->{
try {
TopicOsdRequest response = Common.getObjectMapper().readValue((byte[]) source.getPayload(),new TypeReference<TopicOsdRequest>() {});
log.info("{}",response.getData());
}catch (Exception e){
log.error(e.getMessage());
}
return true;
})
.transform(Message.class, source -> {
try {
TopicOsdRequest response = Common.getObjectMapper().readValue((byte[]) source.getPayload(),
new TypeReference<TopicOsdRequest>() {
});
String topic = String.valueOf(source.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
return response.setFrom(topic.substring((THING_MODEL_PRE + PRODUCT).length(),
topic.indexOf(OSD_SUF)));
} catch (IOException e) {
throw new CloudSDKException(e);
}
}, null)
.<TopicOsdRequest>handle((response, headers) -> {
String gateway_sn=response.getGateway();
GatewayManager gateway=droneCommon.registerDevice(gateway_sn);
if(gateway==null){
gateway=gateway= SDKManager.getDeviceSDK(gateway_sn);
}
OsdDeviceTypeEnum typeEnum = OsdDeviceTypeEnum.find(gateway.getType(),
response.getFrom().equals(response.getGateway()));
Map<String, Object> data = (Map<String, Object>) response.getData();
if (!typeEnum.isGateway()) {
List payloadData = (List) data.getOrDefault(PayloadModelConst.PAYLOAD_KEY, new ArrayList<>());
PayloadModelConst.getAllIndexWithPosition().stream().filter(data::containsKey)
.map(data::get).forEach(payloadData::add);
data.put(PayloadModelConst.PAYLOAD_KEY, payloadData);
}
return response.setData(Common.getObjectMapper().convertValue(data, typeEnum.getClassType()));
})
.<TopicOsdRequest, OsdDeviceTypeEnum>route(response -> OsdDeviceTypeEnum.find(response.getData().getClass()),
mapping -> Arrays.stream(OsdDeviceTypeEnum.values()).forEach(key -> mapping.channelMapping(key, key.getChannelName())))
.get();
|
