/*******************************************************************************
 * Copyright 2018 572682
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License.  You may obtain a copy
 * of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations under
 * the License.
 ******************************************************************************/
package us.dot.its.jpo.ode.services.asn1;

import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import us.dot.its.jpo.ode.OdeProperties;
import us.dot.its.jpo.ode.coder.OdeBsmDataCreatorHelper;
import us.dot.its.jpo.ode.coder.OdeMapDataCreatorHelper;
import us.dot.its.jpo.ode.coder.OdeSpatDataCreatorHelper;
import us.dot.its.jpo.ode.coder.OdeSsmDataCreatorHelper;
import us.dot.its.jpo.ode.coder.OdeSrmDataCreatorHelper;
import us.dot.its.jpo.ode.context.AppContext;
import us.dot.its.jpo.ode.model.OdeAsn1Data;
import us.dot.its.jpo.ode.model.OdeBsmData;
import us.dot.its.jpo.ode.model.OdeLogMetadata.RecordType;
import us.dot.its.jpo.ode.plugin.j2735.J2735DSRCmsgID;
import us.dot.its.jpo.ode.traveler.TimTransmogrifier;
import us.dot.its.jpo.ode.util.XmlUtils;
import us.dot.its.jpo.ode.wrapper.AbstractSubscriberProcessor;
import us.dot.its.jpo.ode.wrapper.MessageProducer;
import us.dot.its.jpo.ode.wrapper.serdes.OdeBsmSerializer;

public class Asn1DecodedDataRouter extends AbstractSubscriberProcessor<String, String> {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private OdeProperties odeProperties;
	private MessageProducer<String, OdeBsmData> bsmProducer;
	private MessageProducer<String, String> timProducer;
	private MessageProducer<String, String> spatProducer;
	private MessageProducer<String, String> mapProducer;
	private MessageProducer<String, String> ssmProducer;
	private MessageProducer<String, String> srmProducer;

	public Asn1DecodedDataRouter(OdeProperties odeProps) {
		super();
		this.odeProperties = odeProps;
		this.bsmProducer = new MessageProducer<>(odeProps.getKafkaBrokers(), odeProps.getKafkaProducerType(), null,
				OdeBsmSerializer.class.getName(), odeProperties.getKafkaTopicsDisabledSet());
		this.timProducer = MessageProducer.defaultStringMessageProducer(odeProps.getKafkaBrokers(),
				odeProps.getKafkaProducerType(), odeProperties.getKafkaTopicsDisabledSet());
		this.spatProducer = MessageProducer.defaultStringMessageProducer(odeProps.getKafkaBrokers(),
				odeProps.getKafkaProducerType(), odeProperties.getKafkaTopicsDisabledSet());
		this.mapProducer = MessageProducer.defaultStringMessageProducer(odeProps.getKafkaBrokers(),
				odeProps.getKafkaProducerType(), odeProperties.getKafkaTopicsDisabledSet());
		this.ssmProducer = MessageProducer.defaultStringMessageProducer(odeProps.getKafkaBrokers(),
				odeProps.getKafkaProducerType(), odeProperties.getKafkaTopicsDisabledSet());
		this.srmProducer = MessageProducer.defaultStringMessageProducer(odeProps.getKafkaBrokers(),
				odeProps.getKafkaProducerType(), odeProperties.getKafkaTopicsDisabledSet());
	}

	@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()) {
				// ODE-518/ODE-604 Demultiplex the messages to appropriate topics based on the
				// "recordType"
				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);
				}
				// Send all BSMs also to OdeBsmPojo
				bsmProducer.send(odeProperties.getKafkaTopicOdeBsmPojo(), getRecord().key(), odeBsmData);
				logger.debug("Submitted to BSM Pojo topic");
			} 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);
				}
				// Send all TIMs also to OdeTimJson
				timProducer.send(odeProperties.getKafkaTopicOdeTimJson(), getRecord().key(), odeTimData);
				logger.debug("Submitted to TIM Pojo topic");
			} else if (messageId == J2735DSRCmsgID.SPATMessage.getMsgID()) {
				String odeSpatData = OdeSpatDataCreatorHelper.createOdeSpatData(consumedData).toString();
				if (recordType == RecordType.dnMsg) {
					spatProducer.send(odeProperties.getKafkaTopicOdeDNMsgJson(), getRecord().key(), odeSpatData);
				} else if (recordType == RecordType.rxMsg) {
					spatProducer.send(odeProperties.getKafkaTopicOdeSpatRxJson(), getRecord().key(), odeSpatData);
				} else if (recordType == RecordType.spatTx) {
					spatProducer.send(odeProperties.getKafkaTopicOdeSpatTxPojo(), getRecord().key(), odeSpatData);
				}
				// Send all SPATs also to OdeSpatJson
				spatProducer.send(odeProperties.getKafkaTopicOdeSpatJson(), getRecord().key(), odeSpatData);
				logger.debug("Submitted to SPAT Pojo topic");
			} else if (messageId == J2735DSRCmsgID.MAPMessage.getMsgID()) {
				String odeMapData = OdeMapDataCreatorHelper.createOdeMapData(consumedData).toString();
				if (recordType == RecordType.mapTx) {
					mapProducer.send(odeProperties.getKafkaTopicOdeMapTxPojo(), getRecord().key(), odeMapData);
				}
				// Send all Map also to OdeMapJson
				mapProducer.send(odeProperties.getKafkaTopicOdeMapJson(), getRecord().key(), odeMapData);
				logger.debug("Submitted to MAP Pojo topic");
			} else if (messageId == J2735DSRCmsgID.SSMMessage.getMsgID()) {
				String odeSsmData = OdeSsmDataCreatorHelper.createOdeSsmData(consumedData).toString();
				if (recordType == RecordType.ssmTx) {
					ssmProducer.send(odeProperties.getKafkaTopicOdeSsmPojo(), getRecord().key(), odeSsmData);
				}
				// Send all SSMs also to OdeSsmJson
				ssmProducer.send(odeProperties.getKafkaTopicOdeSsmJson(), getRecord().key(), odeSsmData);
				logger.debug("Submitted to SSM Pojo topic");
			} else if (messageId == J2735DSRCmsgID.SRMMessage.getMsgID()) {
				String odeSrmData = OdeSrmDataCreatorHelper.createOdeSrmData(consumedData).toString();
				if (recordType == RecordType.srmTx) {
					srmProducer.send(odeProperties.getKafkaTopicOdeSrmTxPojo(), getRecord().key(), odeSrmData);
				}
				// Send all SRMs also to OdeSrmJson
				srmProducer.send(odeProperties.getKafkaTopicOdeSrmJson(), getRecord().key(), odeSrmData);
				logger.debug("Submitted to SRM Pojo topic");
			}
		} catch (Exception e) {
			logger.error("Failed to route received data: " + consumedData, e);
		}
		return null;
	}
}
