From 5ba13f10fe36192d7cdd0d1c82b6d6440a34bcaf Mon Sep 17 00:00:00 2001
From: bgehrels <benjamin.gehrels.extern@zalando.de>
Date: Tue, 23 May 2017 16:47:53 +0200
Subject: [PATCH] #23 fixed scheduling

---
 .../EventTransmissionScheduler.java           | 26 +++++++++++++++++++
 .../NakadiProducerAutoConfiguration.java      |  2 ++
 2 files changed, 28 insertions(+)
 create mode 100644 nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/EventTransmissionScheduler.java

diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/EventTransmissionScheduler.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/EventTransmissionScheduler.java
new file mode 100644
index 00000000..ea5746fc
--- /dev/null
+++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/EventTransmissionScheduler.java
@@ -0,0 +1,26 @@
+package org.zalando.nakadiproducer;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.zalando.nakadiproducer.transmission.impl.EventTransmitter;
+
+@Component
+public class EventTransmissionScheduler {
+    private final EventTransmitter eventTransmitter;
+    private final boolean scheduledTransmissionEnabled;
+
+    @Autowired
+    public EventTransmissionScheduler(EventTransmitter eventTransmitter, @Value("${nakadi-producer.scheduled-transmission-enabled:true}") boolean scheduledTransmissionEnabled) {
+        this.eventTransmitter = eventTransmitter;
+        this.scheduledTransmissionEnabled = scheduledTransmissionEnabled;
+    }
+
+    @Scheduled(fixedDelayString = "${nakadi-producer.transmission-polling-delay:1000}")
+    protected void sendEventsIfSchedulingEnabled() {
+        if (scheduledTransmissionEnabled) {
+            eventTransmitter.sendEvents();
+        }
+    }
+}
diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java
index 43617ccc..73271b6b 100644
--- a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java
+++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java
@@ -29,6 +29,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.zalando.fahrschein.NakadiClient;
 import org.zalando.nakadiproducer.eventlog.EventLogWriter;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
@@ -167,6 +168,7 @@ public EventTransmissionService eventTransmissionService(EventLogRepository even
         return new EventTransmissionService(eventLogRepository, nakadiPublishingClient, objectMapper);
     }
 
+
     @PostConstruct
     public void migrateFlyway() {
         Flyway flyway = new Flyway();