<commit_before>package io.flowing.retail.kafka.order.port.message;

import java.io.IOException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import io.flowing.retail.kafka.order.adapter.payload.GoodsFetchedEventPayload;
import io.flowing.retail.kafka.order.adapter.payload.GoodsShippedEventPayload;
import io.flowing.retail.kafka.order.adapter.payload.PaymentReceivedEventPayload;
import io.flowing.retail.kafka.order.domain.Order;
import io.flowing.retail.kafka.order.domain.OrderFlowContext;
import io.flowing.retail.kafka.order.port.persistence.OrderRepository;
import io.zeebe.gateway.ZeebeClient;
import io.zeebe.gateway.api.events.MessageEvent;

@Component
@EnableBinding(Sink.class)
public class MessageListener {

	@Autowired
	private OrderRepository repository;

	@Autowired
	private ZeebeClient zeebe;

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='OrderPlacedEvent'")
  @Transactional
  public void orderPlacedReceived(String messageJson) throws JsonParseException, JsonMappingException, IOException {
    // read data
    Message<Order> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<Order>>() {});
    Order order = message.getPayload();
    
    // persist domain entity
    repository.save(order);

    // prepare data for workflow
    OrderFlowContext context = new OrderFlowContext();
    context.setOrderId(order.getId());
    context.setTraceId(message.getTraceId());

    // and kick of a new flow instance
    System.out.println("New order placed, start flow. " + context.asJson());
    zeebe.topicClient().workflowClient().newCreateInstanceCommand() //
        .bpmnProcessId("order-kafka") //
        .latestVersion() // 
        .payload(context.asJson()) //
        .send().join();
  }

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='PaymentReceivedEvent'")
  @Transactional
  public void paymentReceived(String messageJson) throws Exception {
    Message<PaymentReceivedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<PaymentReceivedEventPayload>>() {});

    PaymentReceivedEventPayload event = message.getPayload(); // TODO: Read something from it? 

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
      .messageName("PaymentReceivedEvent")
      .correlationKey(message.getCorrelationId())
      .send().join();
    
    System.out.println("Correlated " + messageEvent );
  }

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='GoodsFetchedEvent'")
  @Transactional
  public void goodsFetchedReceived(String messageJson) throws Exception {
    Message<GoodsFetchedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<GoodsFetchedEventPayload>>() {});

    String pickId = message.getPayload().getPickId();     

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
        .messageName("PaymentReceivedEvent")
        .correlationKey(message.getCorrelationId())
        .payload("{\"pickId\":\"" + pickId + "\"}") //
        .send().join();

    System.out.println("Correlated " + messageEvent );
  }


  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='GoodsShippedEvent'")
  @Transactional
  public void goodsShippedReceived(String messageJson) throws Exception {
    Message<GoodsShippedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<GoodsShippedEventPayload>>() {});

    String shipmentId = message.getPayload().getShipmentId();     

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
        .messageName("PaymentReceivedEvent")
        .correlationKey(message.getCorrelationId())
        .payload("{\"shipmentId\":\"" + shipmentId + "\"}") //
        .send().join();

    System.out.println("Correlated " + messageEvent );
  }
}<commit_msg>fixed message type name for correlation<commit_after>package io.flowing.retail.kafka.order.port.message;

import java.io.IOException;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import io.flowing.retail.kafka.order.adapter.payload.GoodsFetchedEventPayload;
import io.flowing.retail.kafka.order.adapter.payload.GoodsShippedEventPayload;
import io.flowing.retail.kafka.order.adapter.payload.PaymentReceivedEventPayload;
import io.flowing.retail.kafka.order.domain.Order;
import io.flowing.retail.kafka.order.domain.OrderFlowContext;
import io.flowing.retail.kafka.order.port.persistence.OrderRepository;
import io.zeebe.gateway.ZeebeClient;
import io.zeebe.gateway.api.events.MessageEvent;

@Component
@EnableBinding(Sink.class)
public class MessageListener {

	@Autowired
	private OrderRepository repository;

	@Autowired
	private ZeebeClient zeebe;

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='OrderPlacedEvent'")
  @Transactional
  public void orderPlacedReceived(String messageJson) throws JsonParseException, JsonMappingException, IOException {
    // read data
    Message<Order> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<Order>>() {});
    Order order = message.getPayload();
    
    // persist domain entity
    repository.save(order);

    // prepare data for workflow
    OrderFlowContext context = new OrderFlowContext();
    context.setOrderId(order.getId());
    context.setTraceId(message.getTraceId());

    // and kick of a new flow instance
    System.out.println("New order placed, start flow. " + context.asJson());
    zeebe.topicClient().workflowClient().newCreateInstanceCommand() //
        .bpmnProcessId("order-kafka") //
        .latestVersion() // 
        .payload(context.asJson()) //
        .send().join();
  }

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='PaymentReceivedEvent'")
  @Transactional
  public void paymentReceived(String messageJson) throws Exception {
    Message<PaymentReceivedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<PaymentReceivedEventPayload>>() {});

    PaymentReceivedEventPayload event = message.getPayload(); // TODO: Read something from it? 

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
      .messageName(message.getMessageType())
      .correlationKey(message.getCorrelationId())
      .send().join();
    
    System.out.println("Correlated " + messageEvent );
  }

  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='GoodsFetchedEvent'")
  @Transactional
  public void goodsFetchedReceived(String messageJson) throws Exception {
    Message<GoodsFetchedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<GoodsFetchedEventPayload>>() {});

    String pickId = message.getPayload().getPickId();     

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
        .messageName(message.getMessageType())
        .correlationKey(message.getCorrelationId())
        .payload("{\"pickId\":\"" + pickId + "\"}") //
        .send().join();

    System.out.println("Correlated " + messageEvent );
  }


  @StreamListener(target = Sink.INPUT, condition = "(headers['messageType']?:'')=='GoodsShippedEvent'")
  @Transactional
  public void goodsShippedReceived(String messageJson) throws Exception {
    Message<GoodsShippedEventPayload> message = new ObjectMapper().readValue(messageJson, new TypeReference<Message<GoodsShippedEventPayload>>() {});

    String shipmentId = message.getPayload().getShipmentId();     

    MessageEvent messageEvent = zeebe.topicClient().workflowClient().newPublishMessageCommand() //
        .messageName(message.getMessageType())
        .correlationKey(message.getCorrelationId())
        .payload("{\"shipmentId\":\"" + shipmentId + "\"}") //
        .send().join();

    System.out.println("Correlated " + messageEvent );
  }
}
{'fraction_non_alphanumeric': 0.12917667613015943, 'fraction_numerical': 0.0, 'mean_word_length': 5.005901639344263, 'pattern_counts': {'":': 4, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3442166', 'n_tokens_mistral': 2595, 'n_tokens_neox': 2522, 'n_words': 409}