juplo.de Git - demos/kafka/demos-kafka-payment-system-transfer/blob - src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java

   1 package de.juplo.kafka.payment.transfer;
   2
   3
   4 import com.fasterxml.jackson.databind.ObjectMapper;
   5 import de.juplo.kafka.payment.transfer.adapter.*;
   6 import de.juplo.kafka.payment.transfer.domain.Transfer;
   7 import de.juplo.kafka.payment.transfer.persistence.InMemoryTransferRepository;
   8 import de.juplo.kafka.payment.transfer.ports.TransferRepository;
   9 import de.juplo.kafka.payment.transfer.ports.TransferService;
  10 import lombok.extern.slf4j.Slf4j;
  11 import org.apache.kafka.clients.admin.AdminClient;
  12 import org.apache.kafka.clients.admin.AdminClientConfig;
  13 import org.apache.kafka.clients.consumer.ConsumerConfig;
  14 import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
  15 import org.apache.kafka.clients.consumer.KafkaConsumer;
  16 import org.apache.kafka.clients.producer.KafkaProducer;
  17 import org.apache.kafka.clients.producer.ProducerConfig;
  18 import org.apache.kafka.common.serialization.StringDeserializer;
  19 import org.apache.kafka.common.serialization.StringSerializer;
  20 import org.springframework.boot.SpringApplication;
  21 import org.springframework.boot.autoconfigure.SpringBootApplication;
  22 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  23 import org.springframework.context.annotation.Bean;
  24 import org.springframework.util.Assert;
  25
  26 import java.util.Optional;
  27 import java.util.Properties;
  28
  29
  30 @SpringBootApplication
  31 @EnableConfigurationProperties(TransferServiceProperties.class)
  32 @Slf4j
  33 public class TransferServiceApplication
  34 {
  35   @Bean(destroyMethod = "close")
  36   AdminClient adminClient(TransferServiceProperties properties)
  37   {
  38     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  39
  40     Properties props = new Properties();
  41     props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  42
  43     return AdminClient.create(props);
  44   }
  45
  46   @Bean(destroyMethod = "close")
  47   KafkaProducer<String, String> producer(TransferServiceProperties properties)
  48   {
  49     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  50     Assert.hasText(properties.getTopic(), "juplo.transfer.topic must be set");
  51     Assert.notNull(properties.getNumPartitions(), "juplo.transfer.num-partitions must be set");
  52
  53     Properties props = new Properties();
  54     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  55     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  56     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  57     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, TransferPartitioner.class);
  58     props.put(TransferPartitioner.TOPIC, properties.getTopic());
  59     props.put(TransferPartitioner.NUM_PARTITIONS, properties.getNumPartitions());
  60
  61     return new KafkaProducer<>(props);
  62   }
  63
  64   @Bean
  65   KafkaConsumer<String, String> consumer(TransferServiceProperties properties)
  66   {
  67     Assert.hasText(properties.getBootstrapServers(), "juplo.transfer.bootstrap-servers must be set");
  68     Assert.hasText(properties.getGroupId(), "juplo.transfer.group-id must be set");
  69     Assert.hasText(properties.getGroupInstanceId(), "juplo.transfer.group-instance-id must be set");
  70
  71     Properties props = new Properties();
  72     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
  73     props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getGroupId());
  74     props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, properties.getGroupInstanceId());
  75     props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getCanonicalName());
  76     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  77     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  78     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  79     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  80
  81     return new KafkaConsumer<>(props);
  82   }
  83
  84   @Bean(destroyMethod = "shutdown")
  85   TransferConsumer transferConsumer(
  86       TransferServiceProperties properties,
  87       KafkaConsumer<String, String> consumer,
  88       AdminClient adminClient,
  89       TransferRepository repository,
  90       ObjectMapper mapper,
  91       TransferService productionTransferService,
  92       TransferService restoreTransferService)
  93   {
  94     return
  95         new TransferConsumer(
  96             properties.getTopic(),
  97             properties.getNumPartitions(),
  98             properties.getInstanceIdUriMapping(),
  99             consumer,
 100             adminClient,
 101             repository,
 102             mapper,
 103             new TransferConsumer.ConsumerUseCases() {
 104               @Override
 105               public void create(Long id, Long payer, Long payee, Integer amount)
 106               {
 107                 productionTransferService.create(id, payer, payee, amount);
 108               }
 109
 110               @Override
 111               public Optional<Transfer> get(Long id)
 112               {
 113                 return productionTransferService.get(id);
 114               }
 115
 116               @Override
 117               public void handleStateChange(Long id, Transfer.State state)
 118               {
 119                 productionTransferService.handleStateChange(id, state);
 120               }
 121             },
 122             new TransferConsumer.ConsumerUseCases() {
 123               @Override
 124               public void create(Long id, Long payer, Long payee, Integer amount)
 125               {
 126                 restoreTransferService.create(id, payer, payee, amount);
 127               }
 128
 129               @Override
 130               public Optional<Transfer> get(Long id)
 131               {
 132                 return restoreTransferService.get(id);
 133               }
 134
 135               @Override
 136               public void handleStateChange(Long id, Transfer.State state)
 137               {
 138                 restoreTransferService.handleStateChange(id, state);
 139               }
 140             });
 141   }
 142
 143   @Bean
 144   KafkaMessagingService kafkaMessagingService(
 145       KafkaProducer<String, String> producer,
 146       ObjectMapper mapper,
 147       TransferServiceProperties properties)
 148   {
 149     return new KafkaMessagingService(producer, mapper, properties.getTopic());
 150   }
 151
 152   @Bean
 153   InMemoryTransferRepository inMemoryTransferRepository(
 154       TransferServiceProperties properties,
 155       ObjectMapper mapper)
 156   {
 157     return new InMemoryTransferRepository(properties.getNumPartitions(), mapper);
 158   }
 159
 160   @Bean
 161   TransferService productionTransferService(
 162       TransferRepository repository,
 163       KafkaMessagingService kafkaMessagingService)
 164   {
 165     return new TransferService(repository, kafkaMessagingService);
 166   }
 167
 168   @Bean
 169   TransferService restoreTransferService(
 170       TransferRepository repository,
 171       NoOpMessageService noOpMessageService)
 172   {
 173     return new TransferService(repository, noOpMessageService);
 174   }
 175
 176   @Bean
 177   TransferController transferController(
 178       TransferService productionTransferService,
 179       KafkaMessagingService kafkaMessagingService,
 180       TransferConsumer transferConsumer)
 181   {
 182     return new TransferController(productionTransferService, kafkaMessagingService, transferConsumer);
 183   }
 184
 185
 186   public static void main(String[] args)
 187   {
 188     SpringApplication.run(TransferServiceApplication.class, args);
 189   }
 190 }