Use event deserialization logic from events broker
To avoid code repetition events deserialization was moved to
events-broker library
Bug: Issue 14593
Change-Id: I53794eb961131750f79efdd0be6c42a3ca937b33
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 9cbe4ba..0d432f4 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -45,8 +45,8 @@
maven_jar(
name = "events-broker",
- artifact = "com.gerritforge:events-broker:3.4.0",
- sha1 = "031881f18def90f945b21c7aafda3a1ac95e89c8",
+ artifact = "com.gerritforge:events-broker:3.4.0.1",
+ sha1 = "2d406afa8787621442d855e4b458c97bd24f1198",
)
maven_jar(
diff --git a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
index c7be04c..faeb218 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriber.java
@@ -14,22 +14,17 @@
package com.googlesource.gerrit.plugins.pubsub;
-import static java.util.Objects.requireNonNull;
-
+import com.gerritforge.gerrit.eventbroker.EventDeserializer;
import com.gerritforge.gerrit.eventbroker.EventMessage;
-import com.gerritforge.gerrit.eventbroker.EventMessage.Header;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.flogger.FluentLogger;
-import com.google.gerrit.server.events.Event;
-import com.google.gson.Gson;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import com.google.pubsub.v1.PubsubMessage;
import java.io.IOException;
-import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
@@ -42,7 +37,7 @@
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
- private final Gson gson;
+ private final EventDeserializer eventsDeserializer;
private final PubSubSubscriberMetrics subscriberMetrics;
private final String topic;
private final Consumer<EventMessage> messageProcessor;
@@ -52,13 +47,13 @@
@Inject
public PubSubEventSubscriber(
- Gson gson,
+ EventDeserializer eventsDeserializer,
SubscriberProvider subscriberProvider,
PubSubConfiguration config,
PubSubSubscriberMetrics subscriberMetrics,
@Assisted String topic,
@Assisted Consumer<EventMessage> messageProcessor) {
- this.gson = gson;
+ this.eventsDeserializer = eventsDeserializer;
this.subscriberMetrics = subscriberMetrics;
this.topic = topic;
this.messageProcessor = messageProcessor;
@@ -105,7 +100,7 @@
MessageReceiver getMessageReceiver() {
return (PubsubMessage message, AckReplyConsumer consumer) -> {
try {
- EventMessage event = deserialise(message.getData().toStringUtf8());
+ EventMessage event = eventsDeserializer.deserialize(message.getData().toStringUtf8());
messageProcessor.accept(event);
subscriberMetrics.incrementSucceedToConsumeMessage();
} catch (Exception e) {
@@ -118,21 +113,4 @@
}
};
}
-
- private EventMessage deserialise(String json) {
- EventMessage result = gson.fromJson(json, EventMessage.class);
- if (result.getEvent() == null && result.getHeader() == null) {
- Event event = deserialiseEvent(json);
- result = new EventMessage(new Header(UUID.randomUUID(), event.instanceId), event);
- }
- result.validate();
- return result;
- }
-
- private Event deserialiseEvent(String json) {
- Event event = gson.fromJson(json, Event.class);
- requireNonNull(event.type, "Event type cannot be null");
- requireNonNull(event.instanceId, "Event instance id cannot be null");
- return event;
- }
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
index c5a8318..fa85120 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/pubsub/PubSubEventSubscriberTest.java
@@ -20,6 +20,7 @@
import static org.mockito.Mockito.only;
import static org.mockito.Mockito.verify;
+import com.gerritforge.gerrit.eventbroker.EventDeserializer;
import com.gerritforge.gerrit.eventbroker.EventMessage;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
@@ -53,6 +54,7 @@
new EventMessage(
new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID()), new ProjectCreatedEvent());
private Gson gson = new EventGsonProvider().get();
+ private EventDeserializer deserializer = new EventDeserializer(gson);
@Test
public void shouldIncrementFailedToConsumeMessageWhenReceivingFails() {
@@ -131,7 +133,12 @@
private MessageReceiver messageReceiver(Consumer<EventMessage> consumer) {
return new PubSubEventSubscriber(
- gson, subscriberProviderMock, confMock, pubSubSubscriberMetricsMock, TOPIC, consumer)
+ deserializer,
+ subscriberProviderMock,
+ confMock,
+ pubSubSubscriberMetricsMock,
+ TOPIC,
+ consumer)
.getMessageReceiver();
}
}
