juplo.de Git - demos/spring/data-jdbc/blob - src/main/java/de/trion/kafka/outbox/OutboxService.java

   1 package de.trion.kafka.outbox;
   2
   3 import org.slf4j.Logger;
   4 import org.slf4j.LoggerFactory;
   5 import org.springframework.http.ResponseEntity;
   6 import org.springframework.stereotype.Component;
   7 import org.springframework.web.context.request.async.DeferredResult;
   8
   9 import java.util.HashMap;
  10 import java.util.Map;
  11 import java.util.Random;
  12
  13 @Component
  14 public class OutboxService {
  15
  16     private static final Logger LOG = LoggerFactory.getLogger(OutboxService.class);
  17
  18
  19     private final Map<String, String> state = new HashMap<>();
  20     private final Map<String, DeferredResult> requests = new HashMap<>();
  21
  22     private long counter = 1;
  23
  24
  25     public OutboxService() {}
  26
  27
  28     public String bearbeiteVorgang(String vorgangId, String vbId, String data) {
  29         if (vorgangId == null)
  30             throw new IllegalArgumentException("vorgangId must not be null!");
  31
  32         // Fehler beim Sichern simulieren
  33         Random r = new Random();
  34         int i = r.nextInt(10);
  35         if (i == 0)
  36             throw new RuntimeException("FEHLER!!!!!!");
  37
  38         String result = vorgangId + "|vbId=" + vbId + "|" + counter++ + ", rand=" + i + ": " + data;
  39
  40         if (state.containsKey(vorgangId))
  41             LOG.info("Bearbeite Vorgang {}: alt={}, neu={}", vorgangId, state.get(vorgangId), data);
  42         else
  43             LOG.info("Bearbeite Vorgang {}: neu={}", vorgangId, data);
  44
  45         process(vorgangId, result);
  46         return result;
  47     }
  48
  49     public synchronized void process(String vorgangId, DeferredResult result) {
  50         String data = state.get(vorgangId);
  51         if (data != null) {
  52             result.setResult(ResponseEntity.ok(data));
  53         }
  54         else {
  55             requests.put(vorgangId, result);
  56         }
  57     }
  58
  59     private synchronized void process(String vorgangId, String result) {
  60         state.put(vorgangId, result);
  61         DeferredResult request = requests.get(vorgangId);
  62         if (request != null)
  63           request.setResult(ResponseEntity.ok(result));
  64     }
  65 }