package com.highradius.g4.intelliparse.listener.kafka.consumer;

import com.highradius.g4.core.common.G4Log;
import com.highradius.g4.core.common.G4LogManager;
import com.highradius.g4.intelliparse.dto.kafka.ExtractionRequestDto;
import com.highradius.g4.intelliparse.service.IdpExtractionService;

import lombok.RequiredArgsConstructor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
public class DocumentExtractionConsumer {

  private static final G4Log LOGGER = G4LogManager.getLog(DocumentExtractionConsumer.class);

  private final IdpExtractionService idpExtractionService;

  @KafkaListener(topicPattern = "${spring.kafka.consumer.topic.document-processing}",
      groupId = "${spring.kafka.consumer.group-id}",
      containerFactory = "idpKafkaListenerContainerFactory")
  public void consumeDocumentExtraction(ExtractionRequestDto message, Acknowledgment acknowledgment) {
    acknowledgment.acknowledge();
    LOGGER.info("Message received in DocumentExtractionConsumer: " + message);
    idpExtractionService.extractionServiceVersion1(message);
  }



}