From b4bfd886af0460ae84b2d70233c2a77866930a46 Mon Sep 17 00:00:00 2001 From: Pu Zhibing <393733352@qq.com> Date: 星期一, 02 六月 2025 15:25:21 +0800 Subject: [PATCH] 修改bug和将roacktmq连接方式切换为直连方式 --- ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java | 310 +++++++++++++++++++++++++++------------------------ 1 files changed, 162 insertions(+), 148 deletions(-) diff --git a/ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java b/ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java index 14214bf..6c047ac 100644 --- a/ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java +++ b/ruoyi-service/ruoyi-integration/src/main/java/com/ruoyi/integration/iotda/utils/listener/IotMessageListener.java @@ -17,6 +17,7 @@ import com.ruoyi.integration.iotda.utils.tools.MessageUtil; import com.ruoyi.integration.iotda.utils.tools.StrategyUtil; import com.ruoyi.integration.rocket.model.*; +import com.ruoyi.integration.rocket.produce.ChargingMessageUtil; import com.ruoyi.integration.rocket.produce.EnhanceProduce; import io.swagger.annotations.ApiOperation; import lombok.extern.slf4j.Slf4j; @@ -58,6 +59,9 @@ @Resource private ChargingPileClient chargingPileClient; + @Resource + private ChargingMessageUtil chargingMessageUtil; + @@ -70,19 +74,26 @@ @PostMapping(value = "/message") public AjaxResult<String> message(@RequestBody JSONObject jsonObject) throws IOException { log.info("接收到消息中转:{}",jsonObject); - JSONObject content = jsonObject.getJSONObject("notify_data").getJSONObject("body").getJSONObject("content"); + JSONObject body = jsonObject.getJSONObject("notify_data").getJSONObject("body"); + JSONObject content = null; + try { + content = body.getJSONObject("content"); + }catch (Exception e){ + String content1 = body.getString("content"); + String substring = content1.substring(content1.length() - 1); + if(!"}".equals(substring)){ + content1 = content1.substring(0, content1.length() - 1) + "}"; + } + System.out.println("content1:"+content1); + content = JSON.parseObject(content1); + } JSONObject header = jsonObject.getJSONObject("notify_data").getJSONObject("header"); String deviceId = header.getString("device_id"); // 设备id String nodeId = header.getString("node_id"); // 设备编号 String productId = header.getString("product_id"); // 产品id String service_id = content.getString("service_id"); log.info("服务id:{}",service_id); - // 写入文件 -// FileWriter writer = new FileWriter("D:/sample.txt",true); -// writer.write("接收到消息中转:"+jsonObject+"\n"); -// writer.write("服务id:"+service_id+"\n"); -// writer.close(); - SendResult sendResult; + SendResult sendResult = null; ChargingMessage chargingMessage = new ChargingMessage(); chargingMessage.setServiceId(service_id); // 设备消息下发 @@ -90,9 +101,11 @@ switch (service_id){ case SendTagConstant.ONLINE: OnlineMessage onlineMessage = JSON.parseObject(content.toJSONString(),OnlineMessage.class); - sendResult = enhanceProduce.onlineMessage(onlineMessage); -// chargingMessage.setOnlineMessage(onlineMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.onlineMessage(onlineMessage); + //直连 + chargingMessage.setOnlineMessage(onlineMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 // 业务处理 登录认证应答 OnlineReply onlineReply = new OnlineReply(); @@ -105,23 +118,18 @@ timingSettingReplyOnline.setCharging_pile_code(onlineMessage.getCharging_pile_code()); timingSettingReplyOnline.setCurrent_time(CP56Time2aConverter.convertToCP56Time2a(new Date())); iotMessageProduce.sendMessage(timingSettingReplyOnline.getCharging_pile_code(), ServiceIdMenu.TIMING_SETTING_REPLY.getKey(),messageUtil.timingSettingReply(timingSettingReplyOnline)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.PING: PingMessage pingMessage = JSON.parseObject(content.toJSONString(),PingMessage.class); //存储缓存中,5分钟有效 redisTemplate.opsForValue().set("ping:" + pingMessage.getCharging_pile_code() + pingMessage.getCharging_gun_code(), pingMessage, 5, TimeUnit.MINUTES); - sendResult = null; -// sendResult = enhanceProduce.pingMessage(pingMessage); -// chargingMessage.setPingMessage(pingMessage); -// chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 Pong pong = new Pong(); pong.setCharging_pile_code(pingMessage.getCharging_pile_code()); pong.setCharging_gun_code(pingMessage.getCharging_gun_code()); pong.setCharging_gun_status(0); iotMessageProduce.sendMessage(pong.getCharging_pile_code(), ServiceIdMenu.PONG.getKey(), messageUtil.pong(pong)); - + //检查修改设备状态 UpdateChargingPileStatusVo vo1 = new UpdateChargingPileStatusVo(); vo1.setGun_code(pingMessage.getCharging_gun_code()); vo1.setPile_code(pingMessage.getCharging_pile_code()); @@ -135,31 +143,30 @@ gunStatusMessage.setFullNumber(gunCode); chargingMessage1.setGunStatusMessage(gunStatusMessage); sendResult = enhanceProduce.gunStatusMessage(chargingMessage1); - - -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.END_CHARGE: EndChargeMessage endChargeMessage = JSON.parseObject(content.toJSONString(),EndChargeMessage.class); - sendResult = enhanceProduce.endChargeMessage(endChargeMessage); -// chargingMessage.setEndChargeMessage(endChargeMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.endChargeMessage(endChargeMessage); + //直连 + chargingMessage.setEndChargeMessage(endChargeMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.ERROR_MESSAGE: ErrorMessageMessage errorMessageMessage = JSON.parseObject(content.toJSONString(),ErrorMessageMessage.class); - sendResult = enhanceProduce.errorMessageMessage(errorMessageMessage); -// chargingMessage.setErrorMessageMessage(errorMessageMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.errorMessageMessage(errorMessageMessage); + //直连 + chargingMessage.setErrorMessageMessage(errorMessageMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.BILLING_MODE_VERIFY: BillingModeVerifyMessage billingModeVerifyMessage = JSON.parseObject(content.toJSONString(),BillingModeVerifyMessage.class); - sendResult = enhanceProduce.billingModeVerifyMessage(billingModeVerifyMessage); -// chargingMessage.setBillingModeVerifyMessage(billingModeVerifyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.billingModeVerifyMessage(billingModeVerifyMessage); + //直连 + chargingMessage.setBillingModeVerifyMessage(billingModeVerifyMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 BillingModeVerifyReply billingModeVerifyReply = new BillingModeVerifyReply(); if(billingModeVerifyMessage.getBilling_model_code().equals("0")){ @@ -183,13 +190,14 @@ } } iotMessageProduce.sendMessage(billingModeVerifyReply.getCharging_pile_code(), ServiceIdMenu.BILLING_MODE_VERIFY_REPLY.getKey(),messageUtil.billingModeVerifyReply(billingModeVerifyReply)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.ACQUISITION_BILLING_MODE: AcquisitionBillingModeMessage acquisitionBillingModeMessage = JSON.parseObject(content.toJSONString(),AcquisitionBillingModeMessage.class); - sendResult = enhanceProduce.acquisitionBillingModeMessage(acquisitionBillingModeMessage); -// chargingMessage.setAcquisitionBillingModeMessage(acquisitionBillingModeMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.acquisitionBillingModeMessage(acquisitionBillingModeMessage); + //直连 + chargingMessage.setAcquisitionBillingModeMessage(acquisitionBillingModeMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 计费模型请求应答 1=尖阶段,2=峰阶段,3=平阶段,4=谷阶段 List<TAccountingStrategyDetail> accountingStrategyDetails = accountingStrategyDetailClient.getDetailListByCode(acquisitionBillingModeMessage.getCharging_pile_code()).getData(); Map<Integer, TAccountingStrategyDetail> strategyPrice = StrategyUtil.getStrategyPrice(accountingStrategyDetails); @@ -203,68 +211,71 @@ acquisitionBillingModeReply.setCharging_pile_code(acquisitionBillingModeMessage.getCharging_pile_code()); acquisitionBillingModeReply.setLoss_ratio(BigDecimal.ZERO); iotMessageProduce.sendMessage(acquisitionBillingModeReply.getCharging_pile_code(), ServiceIdMenu.ACQUISITION_BILLING_MODE_REPLY.getKey(),messageUtil.acquisitionBillingModeReply(acquisitionBillingModeReply)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.UPLOAD_REAL_TIME_MONITORING_DATA: log.info("充电实时数据上传"); UploadRealTimeMonitoringDataMessage uploadRealTimeMonitoringDataMessage = JSON.parseObject(content.toJSONString(),UploadRealTimeMonitoringDataMessage.class); - sendResult = enhanceProduce.uploadRealTimeMonitoringDataMessage(uploadRealTimeMonitoringDataMessage); -// chargingMessage.setUploadRealTimeMonitoringDataMessage(uploadRealTimeMonitoringDataMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.uploadRealTimeMonitoringDataMessage(uploadRealTimeMonitoringDataMessage); + //直连 + chargingMessage.setUploadRealTimeMonitoringDataMessage(uploadRealTimeMonitoringDataMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.CHARGING_HANDSHAKE: ChargingHandshakeMessage chargingHandshakeMessage = JSON.parseObject(content.toJSONString(),ChargingHandshakeMessage.class); - sendResult = enhanceProduce.chargingHandshakeMessage(chargingHandshakeMessage); -// chargingMessage.setChargingHandshakeMessage(chargingHandshakeMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.chargingHandshakeMessage(chargingHandshakeMessage); + //直连 + chargingMessage.setChargingHandshakeMessage(chargingHandshakeMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.PARAMETER_SETTING: ParameterSettingMessage parameterSettingMessage = JSON.parseObject(content.toJSONString(),ParameterSettingMessage.class); - sendResult = enhanceProduce.parameterSettingMessage(parameterSettingMessage); -// chargingMessage.setParameterSettingMessage(parameterSettingMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.parameterSettingMessage(parameterSettingMessage); + //直连 + chargingMessage.setParameterSettingMessage(parameterSettingMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.BMS_ABORT: BmsAbortMessage bmsAbortMessage = JSON.parseObject(content.toJSONString(),BmsAbortMessage.class); - sendResult = enhanceProduce.bmsAbortMessage(bmsAbortMessage); -// chargingMessage.setBmsAbortMessage(bmsAbortMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.bmsAbortMessage(bmsAbortMessage); + //直连 + chargingMessage.setBmsAbortMessage(bmsAbortMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.MOTOR_ABORT: MotorAbortMessage motorAbortMessage = JSON.parseObject(content.toJSONString(),MotorAbortMessage.class); - sendResult = enhanceProduce.motorAbortMessage(motorAbortMessage); -// chargingMessage.setMotorAbortMessage(motorAbortMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.motorAbortMessage(motorAbortMessage); + //直连 + chargingMessage.setMotorAbortMessage(motorAbortMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.BMS_DEMAND_AND_CHARGER_EXPORTATION: BmsDemandAndChargerExportationMessage bmsDemandAndChargerExportationMessage = JSON.parseObject(content.toJSONString(),BmsDemandAndChargerExportationMessage.class); - sendResult = enhanceProduce.bmsDemandAndChargerExportationMessage(bmsDemandAndChargerExportationMessage); -// chargingMessage.setBmsDemandAndChargerExportationMessage(bmsDemandAndChargerExportationMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.bmsDemandAndChargerExportationMessage(bmsDemandAndChargerExportationMessage); + //直连 + chargingMessage.setBmsDemandAndChargerExportationMessage(bmsDemandAndChargerExportationMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.BMS_INFORMATION: BmsInformationMessage bmsInformationMessage = JSON.parseObject(content.toJSONString(),BmsInformationMessage.class); - sendResult = enhanceProduce.bmsInformationMessage(bmsInformationMessage); -// chargingMessage.setBmsInformationMessage(bmsInformationMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.bmsInformationMessage(bmsInformationMessage); + //直连 + chargingMessage.setBmsInformationMessage(bmsInformationMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.CHARGING_PILE_STARTS_CHARGING: ChargingPileStartsChargingMessage chargingPileStartsChargingMessage = JSON.parseObject(content.toJSONString(),ChargingPileStartsChargingMessage.class); - sendResult = enhanceProduce.chargingPileStartsChargingMessage(chargingPileStartsChargingMessage); -// chargingMessage.setChargingPileStartsChargingMessage(chargingPileStartsChargingMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.chargingPileStartsChargingMessage(chargingPileStartsChargingMessage); + //直连 + chargingMessage.setChargingPileStartsChargingMessage(chargingPileStartsChargingMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 PlatformConfirmationCharging platformConfirmationCharging = new PlatformConfirmationCharging(); platformConfirmationCharging.setCharging_pile_code(chargingPileStartsChargingMessage.getCharging_pile_code()); @@ -274,136 +285,137 @@ platformConfirmationCharging.setAuthentication(1); // TODO 若是失败,给出失败原因 iotMessageProduce.sendMessage(platformConfirmationCharging.getCharging_pile_code(), ServiceIdMenu.PLATFORM_CONFIRMATION_CHARGING.getKey(),messageUtil.platformConfirmationCharging(platformConfirmationCharging)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.PLATFORM_START_CHARGING_REPLY: PlatformStartChargingReplyMessage platformStartChargingReplyMessage = JSON.parseObject(content.toJSONString(),PlatformStartChargingReplyMessage.class); - sendResult = enhanceProduce.platformStartChargingReplyMessage(platformStartChargingReplyMessage); -// chargingMessage.setPlatformStartChargingReplyMessage(platformStartChargingReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.platformStartChargingReplyMessage(platformStartChargingReplyMessage); + //直连 + chargingMessage.setPlatformStartChargingReplyMessage(platformStartChargingReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.PLATFORM_STOP_CHARGING_REPLY: PlatformStopChargingReplyMessage platformStopChargingReplyMessage = JSON.parseObject(content.toJSONString(),PlatformStopChargingReplyMessage.class); - sendResult = enhanceProduce.platformStopChargingReplyMessage(platformStopChargingReplyMessage); -// chargingMessage.setPlatformStopChargingReplyMessage(platformStopChargingReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.platformStopChargingReplyMessage(platformStopChargingReplyMessage); + //直连 + chargingMessage.setPlatformStopChargingReplyMessage(platformStopChargingReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.TRANSACTION_RECORD: TransactionRecordMessage transactionRecordMessage = JSON.parseObject(content.toJSONString(),TransactionRecordMessage.class); transactionRecordMessage.setResult(content.toJSONString()); - sendResult = enhanceProduce.transactionRecordMessage(transactionRecordMessage); -// chargingMessage.setTransactionRecordMessage(transactionRecordMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.transactionRecordMessage(transactionRecordMessage); + //直连 + chargingMessage.setTransactionRecordMessage(transactionRecordMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 ConfirmTransactionRecord confirmTransactionRecord = new ConfirmTransactionRecord(); confirmTransactionRecord.setTransaction_serial_number(transactionRecordMessage.getTransaction_serial_number()); confirmTransactionRecord.setConfirm_result(0); iotMessageProduce.sendMessage(transactionRecordMessage.getCharging_pile_code(), ServiceIdMenu.CONFIRM_TRANSACTION_RECORD.getKey(),messageUtil.confirmTransactionRecord(confirmTransactionRecord)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.UPDATE_BALANCE_REPLY: UpdateBalanceReplyMessage updateBalanceReplyMessage = JSON.parseObject(content.toJSONString(),UpdateBalanceReplyMessage.class); - sendResult = enhanceProduce.updateBalanceReplyMessage(updateBalanceReplyMessage); -// chargingMessage.setUpdateBalanceReplyMessage(updateBalanceReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.updateBalanceReplyMessage(updateBalanceReplyMessage); + //直连 + chargingMessage.setUpdateBalanceReplyMessage(updateBalanceReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.SYNCHRONIZE_OFFLINE_CARD_REPLY: SynchronizeOfflineCardReplyMessage synchronizeOfflineCardReplyMessage = JSON.parseObject(content.toJSONString(),SynchronizeOfflineCardReplyMessage.class); - sendResult = enhanceProduce.synchronizeOfflineCardReplyMessage(synchronizeOfflineCardReplyMessage); -// chargingMessage.setSynchronizeOfflineCardReplyMessage(synchronizeOfflineCardReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.synchronizeOfflineCardReplyMessage(synchronizeOfflineCardReplyMessage); + //直连 + chargingMessage.setSynchronizeOfflineCardReplyMessage(synchronizeOfflineCardReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.CLEAR_OFFLINE_CARD_REPLY: ClearOfflineCardReplyMessage clearOfflineCardReplyMessage = JSON.parseObject(content.toJSONString(),ClearOfflineCardReplyMessage.class); - sendResult = enhanceProduce.clearOfflineCardReplyMessage(clearOfflineCardReplyMessage); -// chargingMessage.setClearOfflineCardReplyMessage(clearOfflineCardReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.clearOfflineCardReplyMessage(clearOfflineCardReplyMessage); + //直连 + chargingMessage.setClearOfflineCardReplyMessage(clearOfflineCardReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.WORKING_PARAMETER_SETTING_REPLY: WorkingParameterSettingReplyMessage workingParameterSettingReplyMessage = JSON.parseObject(content.toJSONString(),WorkingParameterSettingReplyMessage.class); - sendResult = enhanceProduce.workingParameterSettingReplyMessage(workingParameterSettingReplyMessage); -// chargingMessage.setWorkingParameterSettingReplyMessage(workingParameterSettingReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.workingParameterSettingReplyMessage(workingParameterSettingReplyMessage); + //直连 + chargingMessage.setWorkingParameterSettingReplyMessage(workingParameterSettingReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.TIMING_SETTING: TimingSettingMessage timingSettingMessage = JSON.parseObject(content.toJSONString(),TimingSettingMessage.class); - sendResult = enhanceProduce.timingSettingMessage(timingSettingMessage); -// chargingMessage.setTimingSettingMessage(timingSettingMessage); -// chargingMessageUtil.handleMessage(chargingMessage); + //走rocketmq +// sendResult = enhanceProduce.timingSettingMessage(timingSettingMessage); + //直连 + chargingMessage.setTimingSettingMessage(timingSettingMessage); + chargingMessageUtil.handleMessage(chargingMessage); // 响应硬件 对时设置应答 TimingSettingReply timingSettingReply = new TimingSettingReply(); timingSettingReply.setCharging_pile_code(timingSettingMessage.getCharging_pile_code()); timingSettingReply.setCurrent_time(CP56Time2aConverter.convertToCP56Time2a(new Date())); iotMessageProduce.sendMessage(timingSettingReply.getCharging_pile_code(), ServiceIdMenu.TIMING_SETTING_REPLY.getKey(),messageUtil.timingSettingReply(timingSettingReply)); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); break; case SendTagConstant.SETUP_BILLING_MODEL_REPLY: SetupBillingModelReplyMessage setupBillingModelReplyMessage = JSON.parseObject(content.toJSONString(),SetupBillingModelReplyMessage.class); - sendResult = enhanceProduce.setupBillingModelReplyMessage(setupBillingModelReplyMessage); -// chargingMessage.setSetupBillingModelReplyMessage(setupBillingModelReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.setupBillingModelReplyMessage(setupBillingModelReplyMessage); + //直连 + chargingMessage.setSetupBillingModelReplyMessage(setupBillingModelReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.GROUND_LOCK_REAL_TIME_DATA: GroundLockRealTimeDataMessage groundLockRealTimeDataMessage = JSON.parseObject(content.toJSONString(),GroundLockRealTimeDataMessage.class); - sendResult = enhanceProduce.groundLockRealTimeDataMessage(groundLockRealTimeDataMessage); -// chargingMessage.setGroundLockRealTimeDataMessage(groundLockRealTimeDataMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.groundLockRealTimeDataMessage(groundLockRealTimeDataMessage); + //直连 + chargingMessage.setGroundLockRealTimeDataMessage(groundLockRealTimeDataMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.CHARGING_PILE_RETURNS_GROUND_LOCK_DATA: ChargingPileReturnsGroundLockDataMessage chargingPileReturnsGroundLockDataMessage = JSON.parseObject(content.toJSONString(),ChargingPileReturnsGroundLockDataMessage.class); - sendResult = enhanceProduce.chargingPileReturnsGroundLockDataMessage(chargingPileReturnsGroundLockDataMessage); -// chargingMessage.setChargingPileReturnsGroundLockDataMessage(chargingPileReturnsGroundLockDataMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.chargingPileReturnsGroundLockDataMessage(chargingPileReturnsGroundLockDataMessage); + //直连 + chargingMessage.setChargingPileReturnsGroundLockDataMessage(chargingPileReturnsGroundLockDataMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.PLATFORM_RESTART_REPLY: PlatformRestartReplyMessage platformRestartReplyMessage = JSON.parseObject(content.toJSONString(),PlatformRestartReplyMessage.class); - sendResult = enhanceProduce.platformRestartReplyMessage(platformRestartReplyMessage); -// chargingMessage.setPlatformRestartReplyMessage(platformRestartReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.platformRestartReplyMessage(platformRestartReplyMessage); + //直连 + chargingMessage.setPlatformRestartReplyMessage(platformRestartReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.QR_CODE_DELIVERY_REPLY: QrCodeDeliveryReplyMessage qrCodeDeliveryReplyMessage = JSON.parseObject(content.toJSONString(),QrCodeDeliveryReplyMessage.class); - sendResult = enhanceProduce.qrCodeDeliveryReplyMessage(qrCodeDeliveryReplyMessage); -// chargingMessage.setQrCodeDeliveryReplyMessage(qrCodeDeliveryReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.qrCodeDeliveryReplyMessage(qrCodeDeliveryReplyMessage); + //直连 + chargingMessage.setQrCodeDeliveryReplyMessage(qrCodeDeliveryReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; case SendTagConstant.SECURITY_DETECTION: SecurityDetectionMessage securityDetectionMessage = JSON.parseObject(content.toJSONString(),SecurityDetectionMessage.class); - sendResult = enhanceProduce.securityDetectionMessage(securityDetectionMessage); -// chargingMessage.setSecurityDetectionMessage(securityDetectionMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.securityDetectionMessage(securityDetectionMessage); + //直连 + chargingMessage.setSecurityDetectionMessage(securityDetectionMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; default: PlatformRemoteUpdateReplyMessage platformRemoteUpdateReplyMessage = JSON.parseObject(content.toJSONString(),PlatformRemoteUpdateReplyMessage.class); - sendResult = enhanceProduce.platformRemoteUpdateReplyMessage(platformRemoteUpdateReplyMessage); -// chargingMessage.setPlatformRemoteUpdateReplyMessage(platformRemoteUpdateReplyMessage); -// chargingMessageUtil.handleMessage(chargingMessage); -// sendResult = enhanceProduce.chargingMessage(chargingMessage); - // 响应硬件 + //走rocketmq +// sendResult = enhanceProduce.platformRemoteUpdateReplyMessage(platformRemoteUpdateReplyMessage); + //直连 + chargingMessage.setPlatformRemoteUpdateReplyMessage(platformRemoteUpdateReplyMessage); + chargingMessageUtil.handleMessage(chargingMessage); break; } log.info("rocketmq消息下发结果:{}",sendResult); @@ -553,5 +565,7 @@ iotMessageProduce.sendMessage(deviceId, ServiceIdMenu.ACQUISITION_BILLING_MODE_REPLY.getKey(),messageUtil.acquisitionBillingModeReply(acquisitionBillingModeReply)); return R.ok(); } - + + + } \ No newline at end of file -- Gitblit v1.7.1