Skip to content

Commit b19f857

Browse files
lmmhbtlmmhbt
committed
replace paho lib with hive lib
1 parent 0f62e5e commit b19f857

2 files changed

Lines changed: 99 additions & 92 deletions

File tree

‎application/pom.xml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -348,9 +348,9 @@
348348
<version>2.0</version>
349349
</dependency>
350350
<dependency>
351-
<groupId>org.eclipse.paho</groupId>
352-
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
353-
<version>1.2.5</version>
351+
<groupId>com.hivemq</groupId>
352+
<artifactId>hivemq-mqtt-client</artifactId>
353+
<version>1.3.7</version>
354354
</dependency>
355355
<dependency>
356356
<groupId>io.github.ci-cmg</groupId>

‎application/src/main/java/org/opentripplanner/updater/trip/gtfs/updater/mqtt/MqttGtfsRealtimeUpdater.java‎

Lines changed: 96 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -5,17 +5,20 @@
55

66
import com.google.protobuf.InvalidProtocolBufferException;
77
import com.google.transit.realtime.GtfsRealtime;
8+
import com.hivemq.client.mqtt.datatypes.MqttQos;
9+
import com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedContext;
10+
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
11+
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
12+
import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5SimpleAuth;
13+
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
814
import java.net.URI;
15+
import java.net.URISyntaxException;
16+
import java.nio.charset.StandardCharsets;
917
import java.util.ArrayList;
1018
import java.util.List;
19+
import java.util.Optional;
20+
import java.util.UUID;
1121
import java.util.function.Consumer;
12-
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
13-
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
14-
import org.eclipse.paho.client.mqttv3.MqttClient;
15-
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
16-
import org.eclipse.paho.client.mqttv3.MqttException;
17-
import org.eclipse.paho.client.mqttv3.MqttMessage;
18-
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
1922
import org.opentripplanner.updater.spi.GraphUpdater;
2023
import org.opentripplanner.updater.spi.UpdateResult;
2124
import org.opentripplanner.updater.spi.WriteToGraphCallback;
@@ -30,7 +33,7 @@
3033
import org.slf4j.LoggerFactory;
3134

3235
/**
33-
* This class starts an Paho MQTT client which opens a connection to a GTFS-RT data source. A
36+
* This class starts a hive MQTT client which opens a connection to a GTFS-RT data source. A
3437
* callback is registered which handles incoming GTFS-RT messages as they stream in by placing a
3538
* GTFS-RT decoder Runnable task in the single-threaded executor for handling.
3639
* <p>
@@ -56,16 +59,14 @@ public class MqttGtfsRealtimeUpdater implements GraphUpdater {
5659
private final int qos;
5760
private final ForwardsDelayPropagationType forwardsDelayPropagationType;
5861
private final BackwardsDelayPropagationType backwardsDelayPropagationType;
59-
private final String clientId = "OpenTripPlanner-" + MqttClient.generateClientId();
6062
private final String configRef;
61-
private final MemoryPersistence persistence = new MemoryPersistence();
6263
private final GtfsRealTimeTripUpdateAdapter adapter;
6364
private final Consumer<UpdateResult> recordMetrics;
6465
private WriteToGraphCallback saveResultOnGraph;
6566

6667
private final boolean fuzzyTripMatching;
6768

68-
private MqttClient client;
69+
private Mqtt5AsyncClient client;
6970

7071
public MqttGtfsRealtimeUpdater(
7172
MqttGtfsRealtimeUpdaterParameters parameters,
@@ -92,106 +93,112 @@ public void setup(WriteToGraphCallback writeToGraphCallback) {
9293

9394
@Override
9495
public void run() throws Exception {
95-
client = new MqttClient(url, clientId, persistence);
96-
MqttConnectOptions connOpts = new MqttConnectOptions();
97-
connOpts.setCleanSession(true);
98-
connOpts.setAutomaticReconnect(true);
96+
client = connectAndSubscribeToClient();
97+
}
98+
99+
private Mqtt5AsyncClient connectAndSubscribeToClient() throws URISyntaxException {
100+
Mqtt5SimpleAuth auth;
101+
99102
URI parsedUrl = new URI(url);
100103
if (parsedUrl.getUserInfo() != null) {
101104
String[] userinfo = parsedUrl.getUserInfo().split(":");
102-
connOpts.setUserName(userinfo[0]);
103-
connOpts.setPassword(userinfo[1].toCharArray());
105+
auth = Mqtt5SimpleAuth.builder()
106+
.username(userinfo[0])
107+
.password(userinfo[1].getBytes(StandardCharsets.UTF_8))
108+
.build();
109+
} else {
110+
auth = null;
104111
}
105-
client.setCallback(new Callback());
106112

107-
LOG.debug("Connecting to broker: {}", url);
108-
client.connect(connOpts);
113+
Mqtt5AsyncClient asyncClient = Mqtt5Client.builder()
114+
.identifier("OpenTripPlanner-" + UUID.randomUUID())
115+
.serverHost(parsedUrl.getHost())
116+
.serverPort(parsedUrl.getPort())
117+
.simpleAuth(auth)
118+
.automaticReconnectWithDefaultConfig()
119+
.addConnectedListener(ctx -> onConnect())
120+
.addDisconnectedListener(this::onDisconnect)
121+
.buildAsync();
122+
123+
asyncClient.connectWith().keepAlive(30).cleanStart(false).send().join();
124+
125+
asyncClient
126+
.subscribeWith()
127+
.topicFilter(topic)
128+
.qos(Optional.ofNullable(MqttQos.fromCode(qos)).orElse(MqttQos.AT_MOST_ONCE))
129+
.callback(this::onMessage)
130+
.send()
131+
.join();
132+
133+
return asyncClient;
134+
}
135+
136+
private void onDisconnect(MqttClientDisconnectedContext ctx) {
137+
LOG.info("Disconnected client from MQTT broker: {}", url, ctx.getCause());
138+
}
139+
140+
private void onConnect() {
141+
LOG.info("Connected client to MQTT broker: {} with qos: {}", url, qos);
109142
}
110143

111144
@Override
112145
public void teardown() {
113-
try {
114-
client.disconnect();
115-
} catch (MqttException e) {
116-
LOG.error("Error disconnecting", e);
117-
}
146+
client.disconnect();
118147
}
119148

120149
@Override
121150
public String getConfigRef() {
122151
return configRef;
123152
}
124153

125-
private class Callback implements MqttCallbackExtended {
126-
127-
@Override
128-
public void connectComplete(boolean reconnect, String serverURI) {
129-
try {
130-
LOG.debug("Connected");
131-
client.subscribe(topic, qos);
132-
} catch (MqttException e) {
133-
LOG.warn("Could not subscribe to: {}", topic);
154+
private void onMessage(Mqtt5Publish message) {
155+
List<GtfsRealtime.TripUpdate> updates = null;
156+
UpdateIncrementality updateIncrementality = FULL_DATASET;
157+
try {
158+
// Decode message
159+
GtfsRealtime.FeedMessage feedMessage = GtfsRealtime.FeedMessage.parseFrom(
160+
message.getPayloadAsBytes()
161+
);
162+
List<GtfsRealtime.FeedEntity> feedEntityList = feedMessage.getEntityList();
163+
164+
// Change fullDataset value if this is an incremental update
165+
if (
166+
feedMessage.hasHeader() &&
167+
feedMessage.getHeader().hasIncrementality() &&
168+
feedMessage
169+
.getHeader()
170+
.getIncrementality()
171+
.equals(GtfsRealtime.FeedHeader.Incrementality.DIFFERENTIAL)
172+
) {
173+
updateIncrementality = DIFFERENTIAL;
134174
}
135-
}
136175

137-
@Override
138-
public void connectionLost(Throwable cause) {
139-
LOG.debug("Disconnected");
140-
}
141-
142-
@Override
143-
public void messageArrived(String topic, MqttMessage message) {
144-
List<GtfsRealtime.TripUpdate> updates = null;
145-
UpdateIncrementality updateIncrementality = FULL_DATASET;
146-
try {
147-
// Decode message
148-
GtfsRealtime.FeedMessage feedMessage = GtfsRealtime.FeedMessage.parseFrom(
149-
message.getPayload()
150-
);
151-
List<GtfsRealtime.FeedEntity> feedEntityList = feedMessage.getEntityList();
152-
153-
// Change fullDataset value if this is an incremental update
154-
if (
155-
feedMessage.hasHeader() &&
156-
feedMessage.getHeader().hasIncrementality() &&
157-
feedMessage
158-
.getHeader()
159-
.getIncrementality()
160-
.equals(GtfsRealtime.FeedHeader.Incrementality.DIFFERENTIAL)
161-
) {
162-
updateIncrementality = DIFFERENTIAL;
176+
// Create List of TripUpdates
177+
updates = new ArrayList<>(feedEntityList.size());
178+
for (GtfsRealtime.FeedEntity feedEntity : feedEntityList) {
179+
if (feedEntity.hasTripUpdate()) {
180+
updates.add(feedEntity.getTripUpdate());
163181
}
164-
165-
// Create List of TripUpdates
166-
updates = new ArrayList<>(feedEntityList.size());
167-
for (GtfsRealtime.FeedEntity feedEntity : feedEntityList) {
168-
if (feedEntity.hasTripUpdate()) {
169-
updates.add(feedEntity.getTripUpdate());
170-
}
171-
}
172-
} catch (InvalidProtocolBufferException e) {
173-
LOG.error("Could not decode gtfs-rt message:", e);
174-
}
175-
176-
if (updates != null) {
177-
// Handle trip updates via graph writer runnable
178-
saveResultOnGraph.execute(
179-
new TripUpdateGraphWriterRunnable(
180-
adapter,
181-
fuzzyTripMatching,
182-
forwardsDelayPropagationType,
183-
backwardsDelayPropagationType,
184-
updateIncrementality,
185-
updates,
186-
feedId,
187-
recordMetrics
188-
)
189-
);
190182
}
183+
} catch (InvalidProtocolBufferException e) {
184+
LOG.error("Could not decode gtfs-rt message:", e);
191185
}
192186

193-
@Override
194-
public void deliveryComplete(IMqttDeliveryToken token) {}
187+
if (updates != null) {
188+
// Handle trip updates via graph writer runnable
189+
saveResultOnGraph.execute(
190+
new TripUpdateGraphWriterRunnable(
191+
adapter,
192+
fuzzyTripMatching,
193+
forwardsDelayPropagationType,
194+
backwardsDelayPropagationType,
195+
updateIncrementality,
196+
updates,
197+
feedId,
198+
recordMetrics
199+
)
200+
);
201+
}
195202
}
196203

197204
@Override

0 commit comments

Comments
 (0)