<commit_before>package io.flowing.retail.kafka.order.messages;

import java.io.IOException;
import java.util.HashMap;

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.domain.Order;
import io.flowing.retail.kafka.order.domain.OrderFlowContext;
import io.flowing.retail.kafka.order.flow.payload.GoodsFetchedEventPayload;
import io.flowing.retail.kafka.order.flow.payload.GoodsShippedEventPayload;
import io.flowing.retail.kafka.order.flow.payload.PaymentReceivedEventPayload;
import io.flowing.retail.kafka.order.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.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.workflowClient().newPublishMessageCommand() //<commit_msg>changed transactional behavior or order entity with Zeebe example to avoid too quick handler calls (as get() of the future should not be in the @transactional scope)<commit_after>package io.flowing.retail.kafka.order.messages;

import java.io.IOException;
import java.util.HashMap;

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.domain.Order;
import io.flowing.retail.kafka.order.domain.OrderFlowContext;
import io.flowing.retail.kafka.order.flow.payload.GoodsFetchedEventPayload;
import io.flowing.retail.kafka.order.flow.payload.GoodsShippedEventPayload;
import io.flowing.retail.kafka.order.flow.payload.PaymentReceivedEventPayload;
import io.flowing.retail.kafka.order.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'")
  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
    // (if we want to do this "transactional" this could be a step in the workflow)
    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.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.workflowClient().newPublishMessageCommand() //
{'fraction_non_alphanumeric': 0.11243898333613869, 'fraction_numerical': 0.0, 'mean_word_length': 5.375536480686695, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, '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': '1378927', 'n_tokens_mistral': 1682, 'n_tokens_neox': 1639, 'n_words': 318}