class DemandService {

	public void createMarriageRegnDemand(MarriageRegnRequest marriageRegnRequest) {
		log.info(" DemandService  createDemand" + marriageRegnRequest);
		MarriageRegn marriageRegn = marriageRegnRequest.getMarriageRegn();
		RequestInfo requestInfo = marriageRegnRequest.getRequestInfo();
		List<Demand> demands = demandServiceRepository.prepareDemand(marriageRegnRequest);
		log.info(" DemandService  createDemand demands" + demands.get(0).getConsumerCode());

		DemandResponse demandResponse = demandServiceRepository.createDemand(demands, requestInfo);
		log.info(" DemandService  createDemand" + demandResponse);
		marriageRegn.setDemands(demandResponse.getDemands());
		List<String> demandIds=new ArrayList<>();
		for (Demand demand : demandResponse.getDemands()) {
			String demandId=demand.getId();
			demandIds.add(demandId);
			}
		log.info("demandIds"+demandIds);
		marriageRegn.setDemandIds(demandIds);
        marriageRegnRequest.setMarriageRegn(marriageRegn);

		log.info(" DemandService  marriageRegnRequest.getRequestInfo()" + marriageRegnRequest);
		/*
		 * ResponseEntity<?> billResponse = demandServiceRepository
		 * .generatebillForMarriageRegnDemand(demandResponse.getDemands(),
		 * requestInfo); log.info(" DemandService  createDemand" +
		 * billResponse); log.info(" DemandService  createDemand" +
		 * marriageRegnRequest); log.info(" DemandService  createDemand" +
		 * propertiesManager.getCreateMarriageRegnTopicName()); BillResponse
		 * billResponseBody=null; billResponseBody= (BillResponse)
		 * billResponse.getBody();
		 * 
		 * log.info(" DemandService  createReciept billResponseBody" +
		 * billResponseBody);
		 * log.info(" DemandService  createReciept  requestInfo" + requestInfo)
		 */;
		/*
		 * ResponseEntity<?> receiptResponse
		 * =demandServiceRepository.generaterecieptForMarriagerRegnPayment(
		 * billResponseBody.getBill(), requestInfo);
		 */
		 String kafkaTopic = null;
		 if (marriageRegn.getSource().equals(Source.DATA_ENTRY)) {
				kafkaTopic = propertiesManager.getCreateMarriageRegnTopicName();
				kafkaTemplate.send(kafkaTopic, marriageRegnRequest);
			} else if (marriageRegn.getSource().equals(Source.SYSTEM)) {
				kafkaTopic = propertiesManager.getCreateWorkflowTopicName();
				marriageRegn.setStatus(ApplicationStatus.WORKFLOW);
				log.info(" DemandService  marriageRegnRequest before calling workflow" + marriageRegnRequest);
				kafkaTemplate.send(kafkaTopic, marriageRegnRequest);
			}
	
	}

}