Zigmars Moldengauers

@Component
public class OrderRoute extends RouteBuilder {

    @Override
    public void configure() {
        from("kafka:orders")
            .unmarshal().json(Order.class)
            .process("validateOrder")
            .choice()
                .when(simple("${body.type} == 'PREMIUM'"))
                    .to("direct:premiumFlow")
                .otherwise()
                    .to("direct:standardFlow")
            .end();
    }
}

@Service
@Slf4j
public class IntegrationService {

    @Autowired
    private KafkaTemplate<String, Event> kafka;

    @Transactional
    public Response processEvent(Event event) {
        log.info("Processing: {}", event.getId());
        var result = transform(event);
        kafka.send("events.processed", result);
        return Response.ok(result);
    }
}

@KafkaListener(
    topics = "integration.events",
    groupId = "processor-group"
)
public void consume(
    ConsumerRecord<String, String> record
) {
    var event = mapper.readValue(
        record.value(), IntegrationEvent.class
    );
    pipeline.execute(event);
    log.info("Processed offset: {}",
        record.offset());
}

@RestController
@RequestMapping("/api/v2")
public class ApiController {

    @GetMapping("/integrations/{id}")
    public ResponseEntity<Integration>
        getIntegration(
            @PathVariable String id
    ) {
        return service
            .findById(id)
            .map(ResponseEntity::ok)
            .orElse(ResponseEntity
                .notFound().build());
    }
}

version: '3.8'
services:
  integration-api:
    build: ./api
    ports:
      - "8080:8080"
    environment:
      KAFKA_BROKERS: kafka:9092
      DB_URL: jdbc:oracle:thin:@db:1521
    depends_on:
      - kafka
      - oracle-db
  kafka:
    image: confluentinc/cp-kafka
    ports:
      - "9092:9092"

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Bean
    public Step processStep(
        StepBuilderFactory steps
    ) {
        return steps.get("process")
            .<Input, Output>chunk(100)
            .reader(dbReader())
            .processor(transformer())
            .writer(kafkaWriter())
            .faultTolerant()
            .retryLimit(3)
            .build();
    }
}

<xsl:stylesheet version="2.0">
  <xsl:template match="/">
    <Response>
      <xsl:for-each
          select="//Order">
        <Item>
          <Id>
            <xsl:value-of
              select="@id"/>
          </Id>
          <Status>mapped</Status>
        </Item>
      </xsl:for-each>
    </Response>
  </xsl:template>
</xsl:stylesheet>

@Component
public class QueueListener {

    @JmsListener(
        destination = "PAYMENT.QUEUE",
        containerFactory = "ibmMQ"
    )
    public void onMessage(
        Message msg
    ) {
        var payment = parse(
            msg.getBody(String.class)
        );
        enricher.enrich(payment);
        routeToTarget(payment);
    }
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: integration-svc
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: api
          image: registry/int:latest
          resources:
            limits:
              memory: "512Mi"
              cpu: "500m"
          readinessProbe:
            httpGet:
              path: /health

from transformers import pipeline

class IntegrationAI:
    def __init__(self):
        self.clf = pipeline(
            "text-classification",
            model="int-router"
        )

    def route(self, msg):
        result = self.clf(
            msg.payload()
        )
        target = resolve(
            result["label"]
        )
        return self.forward(
            msg, target
        )