diff options
| -rw-r--r-- | build.gradle | 1 | ||||
| -rw-r--r-- | src/main/java/org/traccar/MainModule.java | 13 | ||||
| -rw-r--r-- | src/main/java/org/traccar/forward/EventForwarderMqtt.java | 103 |
3 files changed, 113 insertions, 4 deletions
diff --git a/build.gradle b/build.gradle index a48a07be5..91ba2cd5c 100644 --- a/build.gradle +++ b/build.gradle @@ -84,6 +84,7 @@ dependencies { implementation "javax.activation:activation:1.1.1" implementation "com.amazonaws:aws-java-sdk-sns:1.12.349" implementation "org.apache.kafka:kafka-clients:3.3.1" + implementation 'com.hivemq:hivemq-mqtt-client:1.3.0' implementation("com.google.firebase:firebase-admin:9.1.1") { exclude group: 'com.google.cloud', module: 'google-cloud-firestore' exclude group: 'com.google.cloud', module: 'google-cloud-storage' diff --git a/src/main/java/org/traccar/MainModule.java b/src/main/java/org/traccar/MainModule.java index a8f895d3b..55b26010f 100644 --- a/src/main/java/org/traccar/MainModule.java +++ b/src/main/java/org/traccar/MainModule.java @@ -38,6 +38,7 @@ import org.traccar.database.StatisticsManager; import org.traccar.forward.EventForwarder; import org.traccar.forward.EventForwarderJson; import org.traccar.forward.EventForwarderKafka; +import org.traccar.forward.EventForwarderMqtt; import org.traccar.forward.PositionForwarder; import org.traccar.forward.PositionForwarderJson; import org.traccar.forward.PositionForwarderKafka; @@ -322,10 +323,14 @@ public class MainModule extends AbstractModule { @Provides public static EventForwarder provideEventForwarder(Config config, Client client, ObjectMapper objectMapper) { if (config.hasKey(Keys.EVENT_FORWARD_URL)) { - if (config.getString(Keys.EVENT_FORWARD_TYPE).equals("kafka")) { - return new EventForwarderKafka(config, objectMapper); - } else { - return new EventForwarderJson(config, client); + String forwardType = config.getString(Keys.EVENT_FORWARD_TYPE); + switch (forwardType) { + case "kafka": + return new EventForwarderKafka(config, objectMapper); + case "mqtt": + return new EventForwarderMqtt(config, objectMapper); + default: + return new EventForwarderJson(config, client); } } return null; diff --git a/src/main/java/org/traccar/forward/EventForwarderMqtt.java b/src/main/java/org/traccar/forward/EventForwarderMqtt.java new file mode 100644 index 000000000..dc95cb4e2 --- /dev/null +++ b/src/main/java/org/traccar/forward/EventForwarderMqtt.java @@ -0,0 +1,103 @@ +/* + * 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 com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.Mqtt5Client; +import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5SimpleAuth; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.traccar.config.Config; +import org.traccar.config.Keys; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.UUID; + +public class EventForwarderMqtt implements EventForwarder { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventForwarderMqtt.class); + private final Mqtt5AsyncClient client; + private final ObjectMapper objectMapper; + + private final String topic; + + public EventForwarderMqtt(Config config, ObjectMapper objectMapper) { + URI url; + try { + url = new URI(config.getString(Keys.EVENT_FORWARD_URL)); + } catch (URISyntaxException e) { + throw new RuntimeException(e); + } + + String userInfo = url.getUserInfo(); + Mqtt5SimpleAuth simpleAuth = null; + if (userInfo != null) { + int delimiter = userInfo.indexOf(':'); + if (delimiter == -1) { + throw new IllegalArgumentException("Wrong credentials. Should be in format \"username:password\""); + } else { + simpleAuth = Mqtt5SimpleAuth.builder() + .username(userInfo.substring(0, delimiter++)) + .password(userInfo.substring(delimiter).getBytes()) + .build(); + } + } + + String host = url.getHost(); + int port = url.getPort(); + client = Mqtt5Client.builder() + .identifier("traccar-" + UUID.randomUUID().toString()) + .serverHost(host) + .serverPort(port) + .simpleAuth(simpleAuth) + .automaticReconnectWithDefaultConfig() + .buildAsync(); + + client.connectWith() + .send() + .whenComplete((message, e) -> { + if (e != null) { + throw new RuntimeException(e); + } + }); + + this.objectMapper = objectMapper; + topic = config.getString(Keys.EVENT_FORWARD_TOPIC); + } + + @Override + public void forward(EventData eventData, ResultHandler resultHandler) { + byte[] payload; + try { + payload = objectMapper.writeValueAsString(eventData).getBytes(); + } catch (JsonProcessingException e) { + resultHandler.onResult(false, e); + return; + } + + client.publishWith() + .topic(topic) + .qos(MqttQos.AT_LEAST_ONCE) + .payload(payload) + .send() + .whenComplete((message, e) -> resultHandler.onResult(e == null, e)); + } + +} |