@Test public void testProcessDnsTim() throws XmlUtilsException { new Expectations() { { XmlUtils.toJSONObject(anyString); result = mockConsumed; mockConsumed.getJSONObject(anyString); result = mockConsumed; mockConsumed.getInt(anyString); result = J2735DSRCmsgID.TravelerInformation.getMsgID(); mockConsumed.getString(anyString); result = "dnMsg"; mockConsumerRecord.key(); result = "testKey"; mockBsmMessageProducer.send(null, anyString, (OdeBsmData) any); times = 0; mockTimMessageProducer.send(null, anyString, anyString); times = 2; } }; testAsn1DecodedDataRouter.setRecord(mockConsumerRecord); testAsn1DecodedDataRouter.process("testConsumedData"); }
@Override public Object process(String consumedData) { try { JSONObject consumed = XmlUtils.toJSONObject(consumedData).getJSONObject(OdeAsn1Data.class.getSimpleName()); int messageId = consumed.getJSONObject(AppContext.PAYLOAD_STRING).getJSONObject(AppContext.DATA_STRING) .getJSONObject("MessageFrame").getInt("messageId"); RecordType recordType = RecordType .valueOf(consumed.getJSONObject(AppContext.METADATA_STRING).getString("recordType")); if (messageId == J2735DSRCmsgID.BasicSafetyMessage.getMsgID()) { OdeBsmData odeBsmData = OdeBsmDataCreatorHelper.createOdeBsmData(consumedData); if (recordType == RecordType.bsmLogDuringEvent) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmDuringEventPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.rxMsg) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmRxPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.bsmTx) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmTxPojo(), getRecord().key(), odeBsmData); } bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData); } else if (messageId == J2735DSRCmsgID.TravelerInformation.getMsgID()) { String odeTimData = TimTransmogrifier.createOdeTimData(consumed).toString(); if (recordType == RecordType.dnMsg) { timProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeTimData); } else if (recordType == RecordType.rxMsg) { timProducer.send(odeProperties.getKafkaTopicOdeTimRxJson(), getRecord().key(), odeTimData); } timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData); } } catch (Exception e) { logger.error("Failed to route received data: " + consumedData, e); } return null; }
Asn1DecodedDataRouter extends AbstractSubscriberProcessor<String, String> { @Override public Object process(String consumedData) { try { JSONObject consumed = XmlUtils.toJSONObject(consumedData).getJSONObject(OdeAsn1Data.class.getSimpleName()); int messageId = consumed.getJSONObject(AppContext.PAYLOAD_STRING).getJSONObject(AppContext.DATA_STRING) .getJSONObject("MessageFrame").getInt("messageId"); RecordType recordType = RecordType .valueOf(consumed.getJSONObject(AppContext.METADATA_STRING).getString("recordType")); if (messageId == J2735DSRCmsgID.BasicSafetyMessage.getMsgID()) { OdeBsmData odeBsmData = OdeBsmDataCreatorHelper.createOdeBsmData(consumedData); if (recordType == RecordType.bsmLogDuringEvent) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmDuringEventPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.rxMsg) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmRxPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.bsmTx) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmTxPojo(), getRecord().key(), odeBsmData); } bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData); } else if (messageId == J2735DSRCmsgID.TravelerInformation.getMsgID()) { String odeTimData = TimTransmogrifier.createOdeTimData(consumed).toString(); if (recordType == RecordType.dnMsg) { timProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeTimData); } else if (recordType == RecordType.rxMsg) { timProducer.send(odeProperties.getKafkaTopicOdeTimRxJson(), getRecord().key(), odeTimData); } timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData); } } catch (Exception e) { logger.error("Failed to route received data: " + consumedData, e); } return null; } }
Asn1DecodedDataRouter extends AbstractSubscriberProcessor<String, String> { @Override public Object process(String consumedData) { try { JSONObject consumed = XmlUtils.toJSONObject(consumedData).getJSONObject(OdeAsn1Data.class.getSimpleName()); int messageId = consumed.getJSONObject(AppContext.PAYLOAD_STRING).getJSONObject(AppContext.DATA_STRING) .getJSONObject("MessageFrame").getInt("messageId"); RecordType recordType = RecordType .valueOf(consumed.getJSONObject(AppContext.METADATA_STRING).getString("recordType")); if (messageId == J2735DSRCmsgID.BasicSafetyMessage.getMsgID()) { OdeBsmData odeBsmData = OdeBsmDataCreatorHelper.createOdeBsmData(consumedData); if (recordType == RecordType.bsmLogDuringEvent) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmDuringEventPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.rxMsg) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmRxPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.bsmTx) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmTxPojo(), getRecord().key(), odeBsmData); } bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData); } else if (messageId == J2735DSRCmsgID.TravelerInformation.getMsgID()) { String odeTimData = TimTransmogrifier.createOdeTimData(consumed).toString(); if (recordType == RecordType.dnMsg) { timProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeTimData); } else if (recordType == RecordType.rxMsg) { timProducer.send(odeProperties.getKafkaTopicOdeTimRxJson(), getRecord().key(), odeTimData); } timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData); } } catch (Exception e) { logger.error("Failed to route received data: " + consumedData, e); } return null; } Asn1DecodedDataRouter(OdeProperties odeProps); }
Asn1DecodedDataRouter extends AbstractSubscriberProcessor<String, String> { @Override public Object process(String consumedData) { try { JSONObject consumed = XmlUtils.toJSONObject(consumedData).getJSONObject(OdeAsn1Data.class.getSimpleName()); int messageId = consumed.getJSONObject(AppContext.PAYLOAD_STRING).getJSONObject(AppContext.DATA_STRING) .getJSONObject("MessageFrame").getInt("messageId"); RecordType recordType = RecordType .valueOf(consumed.getJSONObject(AppContext.METADATA_STRING).getString("recordType")); if (messageId == J2735DSRCmsgID.BasicSafetyMessage.getMsgID()) { OdeBsmData odeBsmData = OdeBsmDataCreatorHelper.createOdeBsmData(consumedData); if (recordType == RecordType.bsmLogDuringEvent) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmDuringEventPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.rxMsg) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmRxPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.bsmTx) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmTxPojo(), getRecord().key(), odeBsmData); } bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData); } else if (messageId == J2735DSRCmsgID.TravelerInformation.getMsgID()) { String odeTimData = TimTransmogrifier.createOdeTimData(consumed).toString(); if (recordType == RecordType.dnMsg) { timProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeTimData); } else if (recordType == RecordType.rxMsg) { timProducer.send(odeProperties.getKafkaTopicOdeTimRxJson(), getRecord().key(), odeTimData); } timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData); } } catch (Exception e) { logger.error("Failed to route received data: " + consumedData, e); } return null; } Asn1DecodedDataRouter(OdeProperties odeProps); @Override Object process(String consumedData); }
Asn1DecodedDataRouter extends AbstractSubscriberProcessor<String, String> { @Override public Object process(String consumedData) { try { JSONObject consumed = XmlUtils.toJSONObject(consumedData).getJSONObject(OdeAsn1Data.class.getSimpleName()); int messageId = consumed.getJSONObject(AppContext.PAYLOAD_STRING).getJSONObject(AppContext.DATA_STRING) .getJSONObject("MessageFrame").getInt("messageId"); RecordType recordType = RecordType .valueOf(consumed.getJSONObject(AppContext.METADATA_STRING).getString("recordType")); if (messageId == J2735DSRCmsgID.BasicSafetyMessage.getMsgID()) { OdeBsmData odeBsmData = OdeBsmDataCreatorHelper.createOdeBsmData(consumedData); if (recordType == RecordType.bsmLogDuringEvent) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmDuringEventPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.rxMsg) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmRxPojo(), getRecord().key(), odeBsmData); } else if (recordType == RecordType.bsmTx) { bsmProducer.send(odeProperties.getKafkaTopicOdeBsmTxPojo(), getRecord().key(), odeBsmData); } bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData); } else if (messageId == J2735DSRCmsgID.TravelerInformation.getMsgID()) { String odeTimData = TimTransmogrifier.createOdeTimData(consumed).toString(); if (recordType == RecordType.dnMsg) { timProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeTimData); } else if (recordType == RecordType.rxMsg) { timProducer.send(odeProperties.getKafkaTopicOdeTimRxJson(), getRecord().key(), odeTimData); } timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData); } } catch (Exception e) { logger.error("Failed to route received data: " + consumedData, e); } return null; } Asn1DecodedDataRouter(OdeProperties odeProps); @Override Object process(String consumedData);  }