topics, DataSource ds,EventBus eb, EventBusMessage
logger.atInfo().log("Consumer client started successfully");
-
} catch (Exception e) {
logger.atError()
.setCause(e)
@@ -92,7 +116,8 @@ public void close() {
for (AutoCloseable c : closeables) {
try {
- if(c != null) c.close();
+ if (c != null)
+ c.close();
} catch (Exception e) {
logger.atWarn()
.setCause(e)
diff --git a/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusCatchupService.java b/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusCatchupService.java
index aa06861..79a7220 100644
--- a/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusCatchupService.java
+++ b/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusCatchupService.java
@@ -34,7 +34,6 @@
* {@code catchup.getLatestMessageId} - Get the latest message ID for a
* topic
*
- *
*
*
* Example usage:
@@ -53,7 +52,10 @@
public class EventBusCatchupService implements AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(EventBusCatchupService.class);
+ /** EventBus address prefix for fetch events requests. */
public static final String FETCH_EVENTS_ADDRESS = "catchup.fetch_events.";
+
+ /** EventBus address prefix for get latest message ID requests. */
public static final String GET_LATEST_MESSAGE_ID_ADDRESS = "catchup.get_latest.";
private final CatchupServerInterface catchupServer;
@@ -68,11 +70,13 @@ public class EventBusCatchupService implements AutoCloseable {
*
* @param catchupServer The underlying catchup server implementation
* @param eventBus The Vert.x EventBus to use for messaging
+ * @param topics The set of topics to handle catchup requests for
+ * @param executor The async executor for handling requests
*/
public EventBusCatchupService(CatchupServerInterface catchupServer,
- EventBus eventBus,
- Set topics,
- AsyncExecutor executor) {
+ EventBus eventBus,
+ Set topics,
+ AsyncExecutor executor) {
this.catchupServer = catchupServer;
this.eventBus = eventBus;
this.topics = topics;
@@ -85,7 +89,7 @@ public EventBusCatchupService(CatchupServerInterface catchupServer,
* requests.
*/
public void start() {
- if(fetchEventsConsumers == null) {
+ if (fetchEventsConsumers == null) {
logger.atInfo().log("Starting EventBusCatchupService");
@@ -168,7 +172,7 @@ private void handleFetchEvents(Message message) {
executor.submit(() -> {
- try{
+ try {
List events = catchupServer.fetchEvents(fromId, toId, limit, topic);
// Serialize events to JSON and reply
@@ -180,7 +184,7 @@ private void handleFetchEvents(Message message) {
.addArgument(topic)
.log("Successfully fetched {} events for topic {}", events.size(), topic);
- } catch (Exception e){
+ } catch (Exception e) {
logger.atError()
.setCause(e)
.log("Error handling fetchEvents request");
diff --git a/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusMessageBroker.java b/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusMessageBroker.java
index f96e068..ba6c2ed 100644
--- a/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusMessageBroker.java
+++ b/vertx/src/main/java/com/p14n/postevent/vertx/adapter/EventBusMessageBroker.java
@@ -31,7 +31,6 @@
* Events are then published to the Vert.x EventBus for real-time
* distribution
*
- *
*
*
* Subscribers receive events from the EventBus, providing low-latency
@@ -187,8 +186,8 @@ public void subscribeToEventBus(String topic, MessageSubscriber subscribe
});
// Store consumer for potential cleanup
- consumers.compute(topic, (k,l) -> {
- if(l == null){
+ consumers.compute(topic, (k, l) -> {
+ if (l == null) {
l = new ArrayList<>();
}
l.add(consumer);
@@ -209,7 +208,7 @@ public void subscribeToEventBus(String topic, MessageSubscriber subscribe
public void unsubscribe(String topic) {
List> consumerList = consumers.remove(topic);
if (consumerList != null) {
- for(var consumer: consumerList){
+ for (var consumer : consumerList) {
consumer.unregister();
}
logger.atInfo()
@@ -226,7 +225,7 @@ public void close() {
logger.atInfo().log("Closing EventBusMessageBroker");
// Unregister all consumers
- consumers.values().forEach( l -> {
+ consumers.values().forEach(l -> {
l.forEach(MessageConsumer::unregister);
});
consumers.clear();
diff --git a/vertx/src/main/java/com/p14n/postevent/vertx/codec/EventCodec.java b/vertx/src/main/java/com/p14n/postevent/vertx/codec/EventCodec.java
index 10271d9..c7e57fa 100644
--- a/vertx/src/main/java/com/p14n/postevent/vertx/codec/EventCodec.java
+++ b/vertx/src/main/java/com/p14n/postevent/vertx/codec/EventCodec.java
@@ -14,7 +14,8 @@
*
*
* The codec uses JSON serialization for simplicity and debugging ease.
- * Events are encoded as JSON strings with a length prefix for efficient parsing.
+ * Events are encoded as JSON strings with a length prefix for efficient
+ * parsing.
*
*
*
@@ -24,6 +25,14 @@
*/
public class EventCodec implements MessageCodec {
+ /**
+ * Creates a new EventCodec for serializing Event objects on the Vert.x
+ * EventBus.
+ */
+ public EventCodec() {
+ // Default constructor
+ }
+
/**
* Encodes an Event object to the wire format.
* The event is serialized to JSON and prefixed with its length.
@@ -35,7 +44,7 @@ public class EventCodec implements MessageCodec {
public void encodeToWire(Buffer buffer, Event event) {
String json = Json.encode(event);
byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
-
+
// Write length prefix followed by JSON bytes
buffer.appendInt(jsonBytes.length);
buffer.appendBytes(jsonBytes);
@@ -53,11 +62,11 @@ public void encodeToWire(Buffer buffer, Event event) {
public Event decodeFromWire(int pos, Buffer buffer) {
// Read length prefix
int length = buffer.getInt(pos);
-
+
// Read JSON bytes and convert to string
byte[] jsonBytes = buffer.getBytes(pos + 4, pos + 4 + length);
- String json = new String(jsonBytes,StandardCharsets.UTF_8);
-
+ String json = new String(jsonBytes, StandardCharsets.UTF_8);
+
// Deserialize from JSON
return Json.decodeValue(json, Event.class);
}