From 826c661819d044c8f1cf950795ee0f33cc7675c6 Mon Sep 17 00:00:00 2001 From: Anton Tananaev Date: Sun, 13 Nov 2022 16:44:15 -0800 Subject: Implement Kafka forwarding --- src/main/java/org/traccar/MainModule.java | 21 +++++--- src/main/java/org/traccar/config/Keys.java | 23 +++++--- .../org/traccar/forward/EventForwarderKafka.java | 61 ++++++++++++++++++++++ .../traccar/forward/PositionForwarderKafka.java | 55 +++++++++++++++++++ 4 files changed, 147 insertions(+), 13 deletions(-) create mode 100644 src/main/java/org/traccar/forward/EventForwarderKafka.java create mode 100644 src/main/java/org/traccar/forward/PositionForwarderKafka.java (limited to 'src/main/java') 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; diff --git a/src/main/java/org/traccar/config/Keys.java b/src/main/java/org/traccar/config/Keys.java index b60cd82a0..2224192d9 100644 --- a/src/main/java/org/traccar/config/Keys.java +++ b/src/main/java/org/traccar/config/Keys.java @@ -698,6 +698,14 @@ public final class Keys { List.of(KeyType.CONFIG), "max-age=3600,public"); + /** + * Position forwarding format. Available options are "url", "json" and "kafka". Default is "url". + */ + public static final ConfigKey FORWARD_TYPE = new StringConfigKey( + "forward.type", + List.of(KeyType.CONFIG), + "url"); + /** * URL to forward positions. Data is passed through URL parameters. For example, {uniqueId} for device identifier, * {latitude} and {longitude} for coordinates. @@ -713,13 +721,6 @@ public final class Keys { "forward.header", List.of(KeyType.CONFIG)); - /** - * Boolean value to enable forwarding in JSON format. - */ - public static final ConfigKey FORWARD_JSON = new BooleanConfigKey( - "forward.json", - List.of(KeyType.CONFIG)); - /** * Position forwarding retrying enable. When enabled, additional attempts are made to deliver positions. If initial * delivery fails, because of an unreachable server or an HTTP response different from '2xx', the software waits @@ -758,6 +759,14 @@ public final class Keys { List.of(KeyType.CONFIG), 100); + /** + * Events forwarding format. Available options are "json" and "kafka". Default is "json". + */ + public static final ConfigKey EVENT_FORWARD_TYPE = new StringConfigKey( + "event.forward.type", + List.of(KeyType.CONFIG), + "json"); + /** * Events forwarding URL. */ diff --git a/src/main/java/org/traccar/forward/EventForwarderKafka.java b/src/main/java/org/traccar/forward/EventForwarderKafka.java new file mode 100644 index 000000000..71e06ddd1 --- /dev/null +++ b/src/main/java/org/traccar/forward/EventForwarderKafka.java @@ -0,0 +1,61 @@ +/* + * Copyright 2022 Anton Tananaev (anton@traccar.org) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.forward; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.traccar.config.Config; +import org.traccar.config.Keys; + +import java.util.Properties; + +public class EventForwarderKafka implements EventForwarder { + + private final Producer producer; + private final ObjectMapper objectMapper; + + public EventForwarderKafka(Config config, ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + Properties properties = new Properties(); + properties.put("bootstrap.servers", config.getString(Keys.EVENT_FORWARD_URL)); + properties.put("acks", "all"); + properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + producer = new KafkaProducer<>(properties); + } + + @SuppressWarnings("deprecation") + @Override + protected void finalize() { + producer.close(); + } + + @Override + public void forward(EventData eventData, ResultHandler resultHandler) { + try { + String key = Long.toString(eventData.getDevice().getId()); + String value = objectMapper.writeValueAsString(eventData); + producer.send(new ProducerRecord<>("events", key, value)); + resultHandler.onResult(true, null); + } catch (JsonProcessingException e) { + resultHandler.onResult(false, e); + } + } + +} diff --git a/src/main/java/org/traccar/forward/PositionForwarderKafka.java b/src/main/java/org/traccar/forward/PositionForwarderKafka.java new file mode 100644 index 000000000..3921539ac --- /dev/null +++ b/src/main/java/org/traccar/forward/PositionForwarderKafka.java @@ -0,0 +1,55 @@ +/* + * Copyright 2022 Anton Tananaev (anton@traccar.org) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.traccar.forward; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.traccar.config.Config; +import org.traccar.config.Keys; + +import java.util.Properties; + +public class PositionForwarderKafka implements PositionForwarder { + + private final Producer producer; + private final ObjectMapper objectMapper; + + public PositionForwarderKafka(Config config, ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + Properties properties = new Properties(); + properties.put("bootstrap.servers", config.getString(Keys.EVENT_FORWARD_URL)); + properties.put("acks", "all"); + properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); + producer = new KafkaProducer<>(properties); + } + + @Override + public void forward(PositionData positionData, ResultHandler resultHandler) { + try { + String key = Long.toString(positionData.getDevice().getId()); + String value = objectMapper.writeValueAsString(positionData); + producer.send(new ProducerRecord<>("positions", key, value)); + resultHandler.onResult(true, null); + } catch (JsonProcessingException e) { + resultHandler.onResult(false, e); + } + } + +} -- cgit v1.2.3