content: package com.food.order.system.messaging.publisher.kafka;
  
  
  import com.food.order.system.kafka.order.avro.model.RestaurantApprovalRequestAvroModel;
  import com.food.order.system.kafka.producer.KafkaMessageHelper;
  import com.food.order.system.kafka.producer.service.KafkaProducer;
  import com.food.order.system.messaging.mapper.OrderMessagingDataMapper;
  import com.food.order.system.outbox.OutboxStatus;
  import com.food.order.system.config.OrderServiceConfigData;
  import com.food.order.system.outbox.model.approval.OrderApprovalEventPayload;
  import com.food.order.system.outbox.model.approval.OrderApprovalOutboxMessage;
  import com.food.order.system.ports.output.message.publisher.restaurantapproval.RestaurantApprovalRequestMessagePublisher;
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.stereotype.Component;
  
  import java.util.function.BiConsumer;
  
  @Slf4j
  @Component
  @RequiredArgsConstructor
  public class OrderApprovalEventKafkaPublisher implements RestaurantApprovalRequestMessagePublisher {
  
      private final OrderMessagingDataMapper orderMessagingDataMapper;
      private final KafkaProducer<String, RestaurantApprovalRequestAvroModel> kafkaProducer;
      private final OrderServiceConfigData orderServiceConfigData;
      private final KafkaMessageHelper kafkaMessageHelper;
  
  
  
      @Override
      public void publish(OrderApprovalOutboxMessage orderApprovalOutboxMessage,
                          BiConsumer<OrderApprovalOutboxMessage, OutboxStatus> outboxCallback) {
          OrderApprovalEventPayload orderApprovalEventPayload =
                  kafkaMessageHelper.getOrderEventPayload(orderApprovalOutboxMessage.getPayload(),
                          OrderApprovalEventPayload.class);
  
          String sagaId = orderApprovalOutboxMessage.getSagaId().toString();
  
          log.info("Received OrderApprovalOutboxMessage for order id: {} and saga id: {}",
                  orderApprovalEventPayload.getOrderId(),
                  sagaId);
  
          try {
              RestaurantApprovalRequestAvroModel restaurantApprovalRequestAvroModel =
                      orderMessagingDataMapper
                              .orderApprovalEventToRestaurantApprovalRequestAvroModel(sagaId,
                                      orderApprovalEventPayload);
  
              kafkaProducer.send(orderServiceConfigData.getRestaurantApprovalRequestTopicName(),
                      sagaId,
                      restaurantApprovalRequestAvroModel,
                      kafkaMessageHelper.getKafkaCallback(orderServiceConfigData.getRestaurantApprovalRequestTopicName(),
                              restaurantApprovalRequestAvroModel,
                              orderApprovalOutboxMessage,
                              outboxCallback,
                              orderApprovalEventPayload.getOrderId(),
                              "RestaurantApprovalRequestAvroModel"));
  
              log.info("OrderApprovalEventPayload sent to kafka for order id: {} and saga id: {}",
                      restaurantApprovalRequestAvroModel.getOrderId(), sagaId);
          } catch (Exception e) {
              log.error("Error while sending OrderApprovalEventPayload to kafka for order id: {} and saga id: {}," +
                      " error: {}", orderApprovalEventPayload.getOrderId(), sagaId, e.getMessage());
          }
  
  
      }
  }
  