Implement Kafka forwarding

1 files changed, 15 insertions, 6 deletions

diff --git a/src/main/java/org/traccar/MainModule.java b/src/main/java/org/traccar/MainModule.java
index 9d450fef7..6e59527bc 100644
--- a/src/main/java/org/traccar/MainModule.java
+++ b/src/main/java/org/traccar/MainModule.java
@@ -37,8 +37,10 @@ import org.traccar.database.LdapProvider;
 import org.traccar.database.StatisticsManager;
 import org.traccar.forward.EventForwarder;
 import org.traccar.forward.EventForwarderJson;
+import org.traccar.forward.EventForwarderKafka;
 import org.traccar.forward.PositionForwarder;
 import org.traccar.forward.PositionForwarderJson;
+import org.traccar.forward.PositionForwarderKafka;
 import org.traccar.forward.PositionForwarderUrl;
 import org.traccar.geocoder.AddressFormat;
 import org.traccar.geocoder.BanGeocoder;
@@ -314,9 +316,13 @@ public class MainModule extends AbstractModule {
 
     @Singleton
     @Provides
-    public static EventForwarder provideEventForwarder(Config config, Client client) {
+    public static EventForwarder provideEventForwarder(Config config, Client client, ObjectMapper objectMapper) {
         if (config.hasKey(Keys.EVENT_FORWARD_URL)) {
-            return new EventForwarderJson(config, client);
+            if (config.getString(Keys.EVENT_FORWARD_TYPE).equals("kafka")) {
+                return new EventForwarderKafka(config, objectMapper);
+            } else {
+                return new EventForwarderJson(config, client);
+            }
         }
         return null;
     }
@@ -325,10 +331,13 @@ public class MainModule extends AbstractModule {
     @Provides
     public static PositionForwarder providePositionForwarder(Config config, Client client, ObjectMapper objectMapper) {
         if (config.hasKey(Keys.FORWARD_URL)) {
-            if (config.getBoolean(Keys.FORWARD_JSON)) {
-                return new PositionForwarderJson(config, client, objectMapper);
-            } else {
-                return new PositionForwarderUrl(config, client, objectMapper);
+            switch (config.getString(Keys.FORWARD_TYPE)) {
+                case "json":
+                    return new PositionForwarderJson(config, client, objectMapper);
+                case "kafka":
+                    return new PositionForwarderKafka(config, objectMapper);
+                default:
+                    return new PositionForwarderUrl(config, client, objectMapper);
             }
         }
         return null;