You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2025/09/23 04:16:26 UTC
(airavata-custos) branch amie-decoder updated (4e4093189 -> 53f2f81ea)
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a change to branch amie-decoder
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
from 4e4093189 protobuf models for AMIE packets processing events
new 6e56c9a12 model classes for AMIE service
new 334d22f44 AMIE client implementation
new a9da76ff6 polling AMIE endpoint and persists the packet and event in db
new 53f2f81ea AMIE packet processing worker and handler skeleton
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
amie-decoder/pom.xml | 69 ++++++++-
.../main/assembly/amie-decoder-bin-assembly.xml | 1 -
.../apache/custos/amie/AmieDecoderApplication.java | 2 +
.../java/org/apache/custos/amie/AmiePoller.java | 123 ++++++++++++++++
.../org/apache/custos/amie/client/AmieClient.java | 124 ++++++++++++++++
.../apache/custos/amie/config/AmieProperties.java | 121 ++++++++++++++++
.../org/apache/custos/amie/config/AppConfig.java | 31 ++--
.../apache/custos/amie/handler/NoOpHandler.java | 32 ++---
.../apache/custos/amie/handler/PacketHandler.java | 27 ++--
.../apache/custos/amie/handler/PacketRouter.java | 59 ++++++++
.../org/apache/custos/amie/model/PacketEntity.java | 152 ++++++++++++++++++++
.../org/apache/custos/amie/model/PacketStatus.java | 34 ++---
.../custos/amie/model/ProcessingErrorEntity.java | 86 +++++------
.../custos/amie/model/ProcessingEventEntity.java | 158 +++++++++++++++++++++
.../custos/amie/model/ProcessingEventType.java | 9 +-
.../apache/custos/amie/model/ProcessingStatus.java | 27 ++--
.../apache/custos/amie/repo/PacketRepository.java | 18 +--
.../amie/repo/ProcessingErrorRepository.java | 13 +-
.../amie/repo/ProcessingEventRepository.java | 16 ++-
.../org/apache/custos/amie/util/ProtoUtils.java | 76 ++++++++++
.../custos/amie/worker/ProcessingEventWorker.java | 140 ++++++++++++++++++
amie-decoder/src/main/proto/amie_packets.proto | 1 +
amie-decoder/src/main/proto/internal_events.proto | 1 +
amie-decoder/src/main/resources/application.yml | 16 +++
.../resources/distribution/conf/application.yml | 16 +++
pom.xml | 2 +-
26 files changed, 1209 insertions(+), 145 deletions(-)
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/AmiePoller.java
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/client/AmieClient.java
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/config/AmieProperties.java
copy api/src/main/java/org/apache/custos/api/CustosAppController.java => amie-decoder/src/main/java/org/apache/custos/amie/config/AppConfig.java (51%)
copy core/src/main/java/org/apache/custos/core/identity/authzcache/AuthzCache.java => amie-decoder/src/main/java/org/apache/custos/amie/handler/NoOpHandler.java (54%)
copy core/src/main/java/org/apache/custos/core/repo/tenant/TenantRepository.java => amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketHandler.java (52%)
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketRouter.java
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
copy api/src/main/java/org/apache/custos/api/CustosAppController.java => amie-decoder/src/main/java/org/apache/custos/amie/model/PacketStatus.java (59%)
copy core/src/main/java/org/apache/custos/core/model/tenant/TenantStatusUpdateMetadata.java => amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingErrorEntity.java (50%)
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
copy services/src/main/java/org/apache/custos/service/credential/store/Operations.java => amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventType.java (84%)
copy core/src/main/java/org/apache/custos/core/repo/tenant/ContactRepository.java => amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingStatus.java (64%)
copy core/src/main/java/org/apache/custos/core/repo/credential/store/CredentialRepository.java => amie-decoder/src/main/java/org/apache/custos/amie/repo/PacketRepository.java (66%)
copy core/src/main/java/org/apache/custos/core/repo/tenant/ContactRepository.java => amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingErrorRepository.java (74%)
copy core/src/main/java/org/apache/custos/core/repo/credential/store/AgentCredentialRepository.java => amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingEventRepository.java (66%)
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/util/ProtoUtils.java
create mode 100644 amie-decoder/src/main/java/org/apache/custos/amie/worker/ProcessingEventWorker.java
(airavata-custos) 02/04: AMIE client implementation
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch amie-decoder
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 334d22f4432c8efd982624f9316bc07d9d43d7d0
Author: lahiruj <la...@gmail.com>
AuthorDate: Mon Sep 22 23:14:58 2025 -0400
AMIE client implementation
---
.../org/apache/custos/amie/client/AmieClient.java | 124 +++++++++++++++++++++
1 file changed, 124 insertions(+)
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/client/AmieClient.java b/amie-decoder/src/main/java/org/apache/custos/amie/client/AmieClient.java
new file mode 100644
index 000000000..1b787a14d
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/client/AmieClient.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.client;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.custos.amie.config.AmieProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestClientException;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A client for Access CI AMIE REST API
+ */
+@Component
+public class AmieClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AmieClient.class);
+ private static final String PACKETS_PATH = "/packets/{site}/";
+ private static final String HEADER_XA_SITE = "XA-SITE";
+ private static final String HEADER_XA_API_KEY = "XA-API-KEY";
+
+ private final RestTemplate restTemplate;
+ private final AmieProperties properties;
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ public AmieClient(@Qualifier("amieRestTemplate") RestTemplate restTemplate, AmieProperties properties) {
+ this.restTemplate = restTemplate;
+ this.properties = properties;
+ }
+
+ /**
+ * Fetches all in-progress packets from the AMIE API.
+ *
+ * @return A list of packets. Returns an empty list if the call fails or no packets are found.
+ */
+ public List<JsonNode> fetchInProgressPackets() {
+ URI uri = UriComponentsBuilder.fromHttpUrl(properties.getBaseUrl())
+ .path(PACKETS_PATH)
+ .build(properties.getSiteCode());
+
+ HttpHeaders headers = new HttpHeaders();
+ headers.set(HEADER_XA_SITE, properties.getSiteCode());
+ headers.set(HEADER_XA_API_KEY, properties.getApiKey());
+ headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
+
+ HttpEntity<String> entity = new HttpEntity<>(headers);
+
+ try {
+ LOGGER.debug("Fetching in-progress packets from {}", uri);
+ ResponseEntity<String> response = restTemplate.exchange(uri, HttpMethod.GET, entity, String.class);
+
+ if (!response.getStatusCode().is2xxSuccessful() || response.getBody() == null) {
+ LOGGER.warn("Received a non-successful response from AMIE API: {} {}", response.getStatusCode(), response.getBody());
+ return Collections.emptyList();
+ }
+
+ return parsePacketsFromResponse(response.getBody());
+
+ } catch (RestClientException e) {
+ LOGGER.error("Failed to fetch packets from AMIE API at {}", uri, e);
+ return Collections.emptyList();
+ }
+ }
+
+ private List<JsonNode> parsePacketsFromResponse(String responseBody) {
+ List<JsonNode> packets = new ArrayList<>();
+ try {
+ JsonNode root = objectMapper.readTree(responseBody);
+
+ // According to AMIE documentation, the response might be a direct array or an object
+ // containing a "result" key which holds the data.
+ if (root.isArray()) {
+ root.forEach(packets::add);
+ } else if (root.has("result")) {
+ JsonNode resultNode = root.get("result");
+ if (resultNode.isArray()) {
+ resultNode.forEach(packets::add);
+ } else {
+ packets.add(resultNode);
+ }
+ } else {
+ // Handle cases when the response is a single packet object not in an array
+ packets.add(root);
+ }
+ LOGGER.info("Successfully fetched and parsed {} packets from AMIE API.", packets.size());
+ return packets;
+
+ } catch (Exception e) {
+ LOGGER.error("Failed to parse JSON response from AMIE API. Response body: {}", responseBody, e);
+ return Collections.emptyList();
+ }
+ }
+}
(airavata-custos) 04/04: AMIE packet processing worker and handler skeleton
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch amie-decoder
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 53f2f81eafbf77ae6acfe7f42b9794160f5d19ab
Author: lahiruj <la...@gmail.com>
AuthorDate: Tue Sep 23 00:16:02 2025 -0400
AMIE packet processing worker and handler skeleton
---
.../apache/custos/amie/handler/NoOpHandler.java | 45 +++++++
.../apache/custos/amie/handler/PacketHandler.java | 43 +++++++
.../apache/custos/amie/handler/PacketRouter.java | 59 +++++++++
.../custos/amie/worker/ProcessingEventWorker.java | 140 +++++++++++++++++++++
4 files changed, 287 insertions(+)
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/handler/NoOpHandler.java b/amie-decoder/src/main/java/org/apache/custos/amie/handler/NoOpHandler.java
new file mode 100644
index 000000000..c009734c9
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/handler/NoOpHandler.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.handler;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.custos.amie.model.PacketEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+/**
+ * A default PacketHandler that performs no action.
+ */
+@Component
+public class NoOpHandler implements PacketHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(NoOpHandler.class);
+
+ @Override
+ public void handle(JsonNode packetJson, PacketEntity packetEntity) {
+ LOGGER.info("NoOpHandler executed for packet with amie_packet_rec_id [{}] and type [{}]. No action taken.", packetEntity.getAmieId(), packetEntity.getType());
+ // No operations
+ }
+
+ @Override
+ public String supportsType() {
+ return "*";
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketHandler.java b/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketHandler.java
new file mode 100644
index 000000000..9e763b516
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketHandler.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.handler;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.custos.amie.model.PacketEntity;
+
+public interface PacketHandler {
+
+ /**
+ * Executes the business logic for a given AMIE packet
+ *
+ * @param packetJson The raw packet content
+ * @param packetEntity The entity for the packet
+ * @throws Exception if processing fails
+ */
+ void handle(JsonNode packetJson, PacketEntity packetEntity) throws Exception;
+
+ /**
+ * Define which packet type this handler is responsible for.
+ * For example, "request_account_create". "*" indicates a default handler.
+ *
+ * @return The AMIE packet type this handler supports
+ */
+ String supportsType();
+
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketRouter.java b/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketRouter.java
new file mode 100644
index 000000000..50af2c1cd
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/handler/PacketRouter.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.handler;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.custos.amie.model.PacketEntity;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * A service that routes an incoming packet to the appropriate PacketHandler.
+ */
+@Component
+public class PacketRouter {
+
+ private final List<PacketHandler> handlers;
+ private final PacketHandler defaultHandler;
+
+ public PacketRouter(List<PacketHandler> handlers) {
+ this.handlers = handlers;
+ this.defaultHandler = new NoOpHandler();
+ }
+
+ /**
+ * Routes the packet to the correct handler and executes it.
+ *
+ * @param packetJson The raw packet content
+ * @param packetEntity The Packet entity
+ * @throws Exception if the handler logic fails
+ */
+ public void route(JsonNode packetJson, PacketEntity packetEntity) throws Exception {
+ PacketHandler handler = findHandlerFor(packetEntity.getType());
+ handler.handle(packetJson, packetEntity);
+ }
+
+ private PacketHandler findHandlerFor(String packetType) {
+ return handlers.stream()
+ .filter(h -> h.supportsType().equalsIgnoreCase(packetType))
+ .findFirst()
+ .orElse(defaultHandler);
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/worker/ProcessingEventWorker.java b/amie-decoder/src/main/java/org/apache/custos/amie/worker/ProcessingEventWorker.java
new file mode 100644
index 000000000..22edb13bd
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/worker/ProcessingEventWorker.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.worker;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.custos.amie.handler.PacketRouter;
+import org.apache.custos.amie.model.PacketEntity;
+import org.apache.custos.amie.model.PacketStatus;
+import org.apache.custos.amie.model.ProcessingErrorEntity;
+import org.apache.custos.amie.model.ProcessingEventEntity;
+import org.apache.custos.amie.model.ProcessingEventType;
+import org.apache.custos.amie.model.ProcessingStatus;
+import org.apache.custos.amie.repo.PacketRepository;
+import org.apache.custos.amie.repo.ProcessingErrorRepository;
+import org.apache.custos.amie.repo.ProcessingEventRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.time.Instant;
+
+/**
+ * A scheduled worker that fetches for new processing events and executes them.
+ * State of an event (NEW -> RUNNING -> SUCCEEDED/FAILED)
+ */
+@Component
+public class ProcessingEventWorker {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ProcessingEventWorker.class);
+ private static final int MAX_ATTEMPTS = 5;
+
+ private final ProcessingEventRepository eventRepo;
+ private final PacketRepository packetRepo;
+ private final ProcessingErrorRepository errorRepo;
+ private final PacketRouter router;
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ public ProcessingEventWorker(ProcessingEventRepository eventRepo, PacketRepository packetRepo,
+ ProcessingErrorRepository errorRepo, PacketRouter router) {
+ this.eventRepo = eventRepo;
+ this.packetRepo = packetRepo;
+ this.errorRepo = errorRepo;
+ this.router = router;
+ }
+
+ /**
+ * Runs on a fixed delay, checks for NEW events, and processes them one by one.
+ */
+ @Scheduled(fixedDelayString = "${app.amie.scheduler.worker-delay}")
+ @Transactional
+ public void processPendingEvents() {
+ try (var eventStream = eventRepo.findTop50ByStatusOrderByCreatedAtAsc(ProcessingStatus.NEW)) {
+ eventStream.forEach(this::executeEvent);
+ }
+ }
+
+ private void executeEvent(ProcessingEventEntity event) {
+ PacketEntity packet = event.getPacket();
+ LOGGER.info("Processing event [{}] for packet amie_id [{}].", event.getType(), packet.getAmieId());
+
+ event.setStatus(ProcessingStatus.RUNNING);
+ event.setStartedAt(Instant.now());
+ event.setAttempts(event.getAttempts() + 1);
+ eventRepo.save(event);
+
+ try {
+ var packetJson = objectMapper.readTree(packet.getRawJson());
+ router.route(packetJson, packet);
+
+ handleSuccess(event, packet);
+
+ } catch (Exception e) {
+ LOGGER.error("Event processing failed for packet amie_id [{}]. Attempt {} of {}.", packet.getAmieId(), event.getAttempts(), MAX_ATTEMPTS, e);
+ handleFailure(event, packet, e);
+ }
+ }
+
+ private void handleSuccess(ProcessingEventEntity event, PacketEntity packet) {
+ event.setStatus(ProcessingStatus.SUCCEEDED);
+ event.setFinishedAt(Instant.now());
+ eventRepo.save(event);
+
+ if (event.getType() == ProcessingEventType.DECODE_PACKET) {
+ packet.setStatus(PacketStatus.DECODED);
+ packet.setDecodedAt(Instant.now());
+ packetRepo.save(packet);
+ }
+
+ LOGGER.info("Successfully processed event [{}] for packet amie_id [{}].", event.getType(), packet.getAmieId());
+ }
+
+ private void handleFailure(ProcessingEventEntity event, PacketEntity packet, Exception e) {
+ // Check if the event should be retried or marked as failed
+ boolean isRetryable = event.getAttempts() < MAX_ATTEMPTS;
+ ProcessingStatus newStatus = isRetryable ? ProcessingStatus.NEW : ProcessingStatus.FAILED;
+
+ event.setStatus(newStatus);
+ event.setLastError(e.getMessage());
+ event.setFinishedAt(Instant.now());
+ eventRepo.save(event);
+
+ packet.setStatus(PacketStatus.FAILED);
+ packet.setLastError(e.getMessage());
+ packetRepo.save(packet);
+
+ ProcessingErrorEntity error = new ProcessingErrorEntity();
+ error.setPacket(packet);
+ error.setEvent(event);
+ error.setSummary(e.getClass().getSimpleName() + ": " + e.getMessage());
+ error.setDetail(getStackTraceAsString(e));
+ errorRepo.save(error);
+ }
+
+ private String getStackTraceAsString(Exception e) {
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter(sw);
+ e.printStackTrace(pw);
+ return sw.toString();
+ }
+}
(airavata-custos) 03/04: polling AMIE endpoint and persists the packet and event in db
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch amie-decoder
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit a9da76ff635d111cf172174b965575c83d7eba92
Author: lahiruj <la...@gmail.com>
AuthorDate: Mon Sep 22 23:45:46 2025 -0400
polling AMIE endpoint and persists the packet and event in db
---
.../java/org/apache/custos/amie/AmiePoller.java | 123 +++++++++++++++++++++
.../org/apache/custos/amie/model/PacketEntity.java | 4 +-
.../custos/amie/model/ProcessingEventEntity.java | 3 +-
.../org/apache/custos/amie/util/ProtoUtils.java | 76 +++++++++++++
4 files changed, 203 insertions(+), 3 deletions(-)
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmiePoller.java b/amie-decoder/src/main/java/org/apache/custos/amie/AmiePoller.java
new file mode 100644
index 000000000..0efbdb597
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/AmiePoller.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.custos.amie.client.AmieClient;
+import org.apache.custos.amie.model.PacketEntity;
+import org.apache.custos.amie.model.PacketStatus;
+import org.apache.custos.amie.model.ProcessingEventEntity;
+import org.apache.custos.amie.model.ProcessingEventType;
+import org.apache.custos.amie.model.ProcessingStatus;
+import org.apache.custos.amie.repo.PacketRepository;
+import org.apache.custos.amie.repo.ProcessingEventRepository;
+import org.apache.custos.amie.util.ProtoUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * A scheduled service that polls the AMIE API for new packets,
+ * persists them to the database, and later for processing.
+ */
+@Component
+public class AmiePoller {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AmiePoller.class);
+
+ private final AmieClient client;
+ private final PacketRepository packetRepo;
+ private final ProcessingEventRepository eventRepo;
+
+ public AmiePoller(AmieClient client, PacketRepository packetRepo, ProcessingEventRepository eventRepo) {
+ this.client = client;
+ this.packetRepo = packetRepo;
+ this.eventRepo = eventRepo;
+ }
+
+
+ @Scheduled(fixedDelayString = "${app.amie.scheduler.poll-delay}")
+ @Transactional
+ public void pollForPackets() {
+ LOGGER.info("Polling for new AMIE packets...");
+ List<JsonNode> packets = client.fetchInProgressPackets();
+
+ if (packets.isEmpty()) {
+ LOGGER.info("No new packets found.");
+ return;
+ }
+
+ LOGGER.info("Found {} packets to process.", packets.size());
+ for (JsonNode packetNode : packets) {
+ try {
+ processIndividualPacket(packetNode);
+ } catch (Exception e) {
+ // If a malformed packet is found
+ LOGGER.error("An unexpected error occurred while processing a packet. Raw packet: {}", packetNode.toString(), e);
+ }
+ }
+ }
+
+ /**
+ * Processes a single packet from the AMIE API. Only persists new packets.
+ *
+ * @param packetNode The raw packet
+ */
+ private void processIndividualPacket(JsonNode packetNode) {
+ long amiePacketRecId = packetNode.at("/header/packet_rec_id").asLong(-1);
+ String packetType = packetNode.path("type").asText(null);
+
+ if (amiePacketRecId < 0 || packetType == null) {
+ LOGGER.warn("Skipping packet with missing or invalid 'packet_rec_id' or 'type'. Packet: {}", packetNode);
+ return;
+ }
+
+ // Only process if this packetRecId is new
+ packetRepo.findByAmieId(amiePacketRecId).ifPresentOrElse(
+ (existingPacket) -> LOGGER.debug("Packet with amie_packet_rec_id {} already exists. Skip the processing.", amiePacketRecId),
+ () -> {
+ LOGGER.info("Persisting new packet with amie_packet_rec_id {} and type '{}'.", amiePacketRecId, packetType);
+
+ PacketEntity newPacket = new PacketEntity();
+ newPacket.setAmieId(amiePacketRecId);
+ newPacket.setType(packetType);
+ newPacket.setStatus(PacketStatus.NEW);
+ newPacket.setRawJson(packetNode.toString());
+ newPacket.setReceivedAt(Instant.now());
+ packetRepo.save(newPacket);
+
+ ProcessingEventEntity decodeEvent = new ProcessingEventEntity();
+ decodeEvent.setPacket(newPacket);
+ decodeEvent.setType(ProcessingEventType.DECODE_PACKET);
+ decodeEvent.setStatus(ProcessingStatus.NEW);
+
+ byte[] payload = ProtoUtils.createDecodeStartedEvent(decodeEvent.getId(), newPacket.getId(), amiePacketRecId);
+ decodeEvent.setPayload(payload);
+ eventRepo.save(decodeEvent);
+
+ LOGGER.info("Successfully enqueued DECODE_PACKET event for packet {}.", amiePacketRecId);
+ }
+ );
+ }
+}
\ No newline at end of file
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
index 17ede689f..1975a4291 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
@@ -22,19 +22,18 @@ import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.EnumType;
import jakarta.persistence.Enumerated;
-import jakarta.persistence.GeneratedValue;
import jakarta.persistence.Id;
import jakarta.persistence.Lob;
import jakarta.persistence.Table;
import java.time.Instant;
+import java.util.UUID;
@Entity
@Table(name = "packets")
public class PacketEntity {
@Id
- @GeneratedValue(generator = "uuid")
private String id;
@Column(name = "amie_id", nullable = false, unique = true)
@@ -68,6 +67,7 @@ public class PacketEntity {
private String lastError;
public PacketEntity() {
+ this.id = UUID.randomUUID().toString();
}
public String getId() {
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
index 87673d7ee..31ac135ef 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
@@ -31,13 +31,13 @@ import jakarta.persistence.ManyToOne;
import jakarta.persistence.Table;
import java.time.Instant;
+import java.util.UUID;
@Entity
@Table(name = "processing_events")
public class ProcessingEventEntity {
@Id
- @GeneratedValue(generator = "uuid")
private String id;
@ManyToOne(fetch = FetchType.LAZY, optional = false)
@@ -73,6 +73,7 @@ public class ProcessingEventEntity {
private String lastError;
public ProcessingEventEntity() {
+ this.id = UUID.randomUUID().toString();
}
public String getId() {
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/util/ProtoUtils.java b/amie-decoder/src/main/java/org/apache/custos/amie/util/ProtoUtils.java
new file mode 100644
index 000000000..b16f27046
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/util/ProtoUtils.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.util;
+
+import com.google.protobuf.Timestamp;
+import org.apache.custos.amie.internal.events.v1.DecodeStartedPayload;
+import org.apache.custos.amie.internal.events.v1.EventType;
+import org.apache.custos.amie.internal.events.v1.ProcessingEvent;
+import org.apache.custos.amie.internal.events.v1.ProcessingStatus;
+
+import java.time.Instant;
+
+public final class ProtoUtils {
+
+ private ProtoUtils() {
+ }
+
+ /**
+ * Creates the serialized payload for a DECODE_STARTED event.
+ * <p>
+ * This method builds the full {@link ProcessingEvent} envelope with its payload
+ * set to a {@link DecodeStartedPayload}. The resulting byte array is what will be
+ * stored in the {@code payload} column of the {@code processing_events} table.
+ *
+ * @param eventId ID of the event entity
+ * @param packetId ID of the parent packet entity
+ * @param amiePacketRecId The unique record ID from the original AMIE packet header
+ * @return A byte array containing the serialized Protobuf message
+ */
+ public static byte[] createDecodeStartedEvent(String eventId, String packetId, long amiePacketRecId) {
+ DecodeStartedPayload payload = DecodeStartedPayload.newBuilder()
+ .setPacketDbId(packetId)
+ .setAmiePacketRecId(amiePacketRecId)
+ .build();
+
+ ProcessingEvent event = ProcessingEvent.newBuilder()
+ .setId(eventId)
+ .setPacketDbId(packetId)
+ .setAmiePacketRecId(amiePacketRecId)
+ .setType(EventType.DECODE_STARTED)
+ .setStatus(ProcessingStatus.PENDING)
+ .setAttempts(0)
+ .setCreatedAt(instantToTimestamp(Instant.now()))
+ .setDecodeStarted(payload)
+ .build();
+
+ return event.toByteArray();
+ }
+
+ private static Timestamp instantToTimestamp(Instant instant) {
+ if (instant == null) {
+ return Timestamp.getDefaultInstance();
+ }
+ return Timestamp.newBuilder()
+ .setSeconds(instant.getEpochSecond())
+ .setNanos(instant.getNano())
+ .build();
+ }
+}
+
(airavata-custos) 01/04: model classes for AMIE service
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch amie-decoder
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 6e56c9a1218cbda5dc6cb14910d5f2553714fc20
Author: lahiruj <la...@gmail.com>
AuthorDate: Mon Sep 22 23:08:30 2025 -0400
model classes for AMIE service
---
amie-decoder/pom.xml | 69 ++++++++-
.../main/assembly/amie-decoder-bin-assembly.xml | 1 -
.../apache/custos/amie/AmieDecoderApplication.java | 2 +
.../apache/custos/amie/config/AmieProperties.java | 121 ++++++++++++++++
.../org/apache/custos/amie/config/AppConfig.java | 38 +++++
.../org/apache/custos/amie/model/PacketEntity.java | 152 ++++++++++++++++++++
.../PacketStatus.java} | 35 +++--
.../custos/amie/model/ProcessingErrorEntity.java | 107 ++++++++++++++
.../custos/amie/model/ProcessingEventEntity.java | 157 +++++++++++++++++++++
.../ProcessingEventType.java} | 21 +--
.../ProcessingStatus.java} | 35 +++--
.../PacketRepository.java} | 31 ++--
.../ProcessingErrorRepository.java} | 23 ++-
.../ProcessingEventRepository.java} | 28 ++--
amie-decoder/src/main/proto/amie_packets.proto | 1 +
amie-decoder/src/main/proto/internal_events.proto | 1 +
amie-decoder/src/main/resources/application.yml | 16 +++
.../resources/distribution/conf/application.yml | 16 +++
pom.xml | 2 +-
19 files changed, 762 insertions(+), 94 deletions(-)
diff --git a/amie-decoder/pom.xml b/amie-decoder/pom.xml
index 5e8eb2ed6..42446c6d7 100644
--- a/amie-decoder/pom.xml
+++ b/amie-decoder/pom.xml
@@ -1,6 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you 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.
+ -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -26,6 +44,16 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-data-jpa</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-to-slf4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
@@ -35,17 +63,24 @@
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.custos</groupId>
- <artifactId>custos-core</artifactId>
- <version>${project.version}</version>
+ <groupId>org.flywaydb</groupId>
+ <artifactId>flyway-core</artifactId>
</dependency>
<dependency>
- <groupId>org.springframework.vault</groupId>
- <artifactId>spring-vault-core</artifactId>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
</dependency>
</dependencies>
<build>
+ <extensions>
+ <extension>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>1.7.0</version>
+ </extension>
+ </extensions>
+
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -69,6 +104,26 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <version>${protobuf.maven.plugin}</version>
+ <configuration>
+ <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}
+ </protocArtifact>
+ <pluginId>grpc-java</pluginId>
+ <pluginArtifact>io.grpc:protoc-gen-grpc-java:${io.grpc.version}:exe:${os.detected.classifier}
+ </pluginArtifact>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ <goal>compile-custom</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
diff --git a/amie-decoder/src/main/assembly/amie-decoder-bin-assembly.xml b/amie-decoder/src/main/assembly/amie-decoder-bin-assembly.xml
index 43edf6f98..0ff1667d9 100644
--- a/amie-decoder/src/main/assembly/amie-decoder-bin-assembly.xml
+++ b/amie-decoder/src/main/assembly/amie-decoder-bin-assembly.xml
@@ -1,4 +1,3 @@
-
<!--
Licensed to the Apache Software Foundation (ASF) under one
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
index c3d6813ab..6dffe2d0f 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
@@ -22,9 +22,11 @@ package org.apache.custos.amie;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+import org.springframework.scheduling.annotation.EnableScheduling;
@SpringBootApplication
@EnableJpaAuditing
+@EnableScheduling
public class AmieDecoderApplication {
public static void main(String[] args) {
SpringApplication.run(AmieDecoderApplication.class, args);
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/config/AmieProperties.java b/amie-decoder/src/main/java/org/apache/custos/amie/config/AmieProperties.java
new file mode 100644
index 000000000..a4daac8ae
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/config/AmieProperties.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+@ConfigurationProperties(prefix = "app.amie")
+public class AmieProperties {
+
+ /**
+ * The base URL for the AMIE Client API.
+ */
+ private String baseUrl;
+
+ /**
+ * ACCESS CI provided site code (e.g., "NEXUS") for the XA-SITE header
+ */
+ private String siteCode;
+
+ /**
+ * ACCESS CI provided API key for the XA-API-KEY header
+ */
+ private String apiKey;
+
+
+ private final Scheduler scheduler = new Scheduler();
+
+ private final Client client = new Client();
+
+ public static class Scheduler {
+ private Duration pollDelay = Duration.ofSeconds(30);
+ private Duration workerDelay = Duration.ofSeconds(5);
+
+ public Duration getPollDelay() {
+ return pollDelay;
+ }
+
+ public void setPollDelay(Duration pollDelay) {
+ this.pollDelay = pollDelay;
+ }
+
+ public Duration getWorkerDelay() {
+ return workerDelay;
+ }
+
+ public void setWorkerDelay(Duration workerDelay) {
+ this.workerDelay = workerDelay;
+ }
+ }
+
+ public static class Client {
+ private Duration connectTimeout = Duration.ofSeconds(5);
+ private Duration readTimeout = Duration.ofSeconds(20);
+
+ public Duration getConnectTimeout() {
+ return connectTimeout;
+ }
+
+ public void setConnectTimeout(Duration connectTimeout) {
+ this.connectTimeout = connectTimeout;
+ }
+
+ public Duration getReadTimeout() {
+ return readTimeout;
+ }
+
+ public void setReadTimeout(Duration readTimeout) {
+ this.readTimeout = readTimeout;
+ }
+ }
+
+ public String getBaseUrl() {
+ return baseUrl;
+ }
+
+ public void setBaseUrl(String baseUrl) {
+ this.baseUrl = baseUrl;
+ }
+
+ public String getSiteCode() {
+ return siteCode;
+ }
+
+ public void setSiteCode(String siteCode) {
+ this.siteCode = siteCode;
+ }
+
+ public String getApiKey() {
+ return apiKey;
+ }
+
+ public void setApiKey(String apiKey) {
+ this.apiKey = apiKey;
+ }
+
+ public Scheduler getScheduler() {
+ return scheduler;
+ }
+
+ public Client getClient() {
+ return client;
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/config/AppConfig.java b/amie-decoder/src/main/java/org/apache/custos/amie/config/AppConfig.java
new file mode 100644
index 000000000..a56fb6b01
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/config/AppConfig.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.config;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.client.RestTemplate;
+
+@Configuration
+@EnableConfigurationProperties(AmieProperties.class)
+public class AppConfig {
+
+ @Bean("amieRestTemplate")
+ public RestTemplate amieRestTemplate(AmieProperties properties) {
+ return new RestTemplateBuilder()
+ .setConnectTimeout(properties.getClient().getConnectTimeout())
+ .setReadTimeout(properties.getClient().getReadTimeout())
+ .build();
+ }
+}
\ No newline at end of file
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
new file mode 100644
index 000000000..17ede689f
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketEntity.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.model;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.Lob;
+import jakarta.persistence.Table;
+
+import java.time.Instant;
+
+@Entity
+@Table(name = "packets")
+public class PacketEntity {
+
+ @Id
+ @GeneratedValue(generator = "uuid")
+ private String id;
+
+ @Column(name = "amie_id", nullable = false, unique = true)
+ private Long amieId;
+
+ @Column(name = "type", nullable = false, length = 64)
+ private String type;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "status", nullable = false, length = 32)
+ private PacketStatus status = PacketStatus.NEW;
+
+ @Lob
+ @Column(name = "raw_json", columnDefinition = "JSON", nullable = false)
+ private String rawJson;
+
+ @Column(name = "received_at", nullable = false)
+ private Instant receivedAt = Instant.now();
+
+ @Column(name = "decoded_at")
+ private Instant decodedAt;
+
+ @Column(name = "processed_at")
+ private Instant processedAt;
+
+ @Column(name = "retries", nullable = false)
+ private int retries = 0;
+
+ @Lob
+ @Column(name = "last_error")
+ private String lastError;
+
+ public PacketEntity() {
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public Long getAmieId() {
+ return amieId;
+ }
+
+ public void setAmieId(Long amieId) {
+ this.amieId = amieId;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public PacketStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(PacketStatus status) {
+ this.status = status;
+ }
+
+ public String getRawJson() {
+ return rawJson;
+ }
+
+ public void setRawJson(String rawJson) {
+ this.rawJson = rawJson;
+ }
+
+ public Instant getReceivedAt() {
+ return receivedAt;
+ }
+
+ public void setReceivedAt(Instant receivedAt) {
+ this.receivedAt = receivedAt;
+ }
+
+ public Instant getDecodedAt() {
+ return decodedAt;
+ }
+
+ public void setDecodedAt(Instant decodedAt) {
+ this.decodedAt = decodedAt;
+ }
+
+ public Instant getProcessedAt() {
+ return processedAt;
+ }
+
+ public void setProcessedAt(Instant processedAt) {
+ this.processedAt = processedAt;
+ }
+
+ public int getRetries() {
+ return retries;
+ }
+
+ public void setRetries(int retries) {
+ this.retries = retries;
+ }
+
+ public String getLastError() {
+ return lastError;
+ }
+
+ public void setLastError(String lastError) {
+ this.lastError = lastError;
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketStatus.java
similarity index 50%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/model/PacketStatus.java
index c3d6813ab..a35680243 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/PacketStatus.java
@@ -7,26 +7,33 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
+package org.apache.custos.amie.model;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+public enum PacketStatus {
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+ /**
+ * Packet has been received from the AMIE API and persisted but has not yet been processed.
+ */
+ NEW,
+ /**
+ * The packet's raw JSON content has been successfully parsed, and the initial processing event has been created.
+ */
+ DECODED,
+ /**
+ * All required processing events have been completed successfully.
+ */
+ PROCESSED,
+ /**
+ * The processing of this packet failed and will not be automatically retried.
+ */
+ FAILED
}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingErrorEntity.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingErrorEntity.java
new file mode 100644
index 000000000..ae9ef13bb
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingErrorEntity.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.model;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.FetchType;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.Lob;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.Table;
+
+import java.time.Instant;
+
+@Entity
+@Table(name = "processing_errors")
+public class ProcessingErrorEntity {
+
+ @Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private Long id;
+
+ @ManyToOne(fetch = FetchType.LAZY)
+ @JoinColumn(name = "packet_id")
+ private PacketEntity packet;
+
+ @ManyToOne(fetch = FetchType.LAZY)
+ @JoinColumn(name = "event_id")
+ private ProcessingEventEntity event;
+
+ @Column(name = "occurred_at", nullable = false)
+ private Instant occurredAt = Instant.now();
+
+ @Column(name = "summary", nullable = false)
+ private String summary;
+
+ @Lob
+ @Column(name = "detail")
+ private String detail;
+
+ public Long getId() {
+ return id;
+ }
+
+ public void setId(Long id) {
+ this.id = id;
+ }
+
+ public PacketEntity getPacket() {
+ return packet;
+ }
+
+ public void setPacket(PacketEntity packet) {
+ this.packet = packet;
+ }
+
+ public ProcessingEventEntity getEvent() {
+ return event;
+ }
+
+ public void setEvent(ProcessingEventEntity event) {
+ this.event = event;
+ }
+
+ public Instant getOccurredAt() {
+ return occurredAt;
+ }
+
+ public void setOccurredAt(Instant occurredAt) {
+ this.occurredAt = occurredAt;
+ }
+
+ public String getSummary() {
+ return summary;
+ }
+
+ public void setSummary(String summary) {
+ this.summary = summary;
+ }
+
+ public String getDetail() {
+ return detail;
+ }
+
+ public void setDetail(String detail) {
+ this.detail = detail;
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
new file mode 100644
index 000000000..87673d7ee
--- /dev/null
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventEntity.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.custos.amie.model;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.FetchType;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.Lob;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.Table;
+
+import java.time.Instant;
+
+@Entity
+@Table(name = "processing_events")
+public class ProcessingEventEntity {
+
+ @Id
+ @GeneratedValue(generator = "uuid")
+ private String id;
+
+ @ManyToOne(fetch = FetchType.LAZY, optional = false)
+ @JoinColumn(name = "packet_id", nullable = false)
+ private PacketEntity packet;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "type", nullable = false, length = 64)
+ private ProcessingEventType type;
+
+ @Enumerated(EnumType.STRING)
+ @Column(name = "status", nullable = false, length = 32)
+ private ProcessingStatus status = ProcessingStatus.NEW;
+
+ @Column(name = "attempts", nullable = false)
+ private int attempts = 0;
+
+ @Lob
+ @Column(name = "payload", nullable = false)
+ private byte[] payload;
+
+ @Column(name = "created_at", nullable = false)
+ private Instant createdAt = Instant.now();
+
+ @Column(name = "started_at")
+ private Instant startedAt;
+
+ @Column(name = "finished_at")
+ private Instant finishedAt;
+
+ @Lob
+ @Column(name = "last_error")
+ private String lastError;
+
+ public ProcessingEventEntity() {
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public PacketEntity getPacket() {
+ return packet;
+ }
+
+ public void setPacket(PacketEntity packet) {
+ this.packet = packet;
+ }
+
+ public ProcessingEventType getType() {
+ return type;
+ }
+
+ public void setType(ProcessingEventType type) {
+ this.type = type;
+ }
+
+ public ProcessingStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(ProcessingStatus status) {
+ this.status = status;
+ }
+
+ public int getAttempts() {
+ return attempts;
+ }
+
+ public void setAttempts(int attempts) {
+ this.attempts = attempts;
+ }
+
+ public byte[] getPayload() {
+ return payload;
+ }
+
+ public void setPayload(byte[] payload) {
+ this.payload = payload;
+ }
+
+ public Instant getCreatedAt() {
+ return createdAt;
+ }
+
+ public void setCreatedAt(Instant createdAt) {
+ this.createdAt = createdAt;
+ }
+
+ public Instant getStartedAt() {
+ return startedAt;
+ }
+
+ public void setStartedAt(Instant startedAt) {
+ this.startedAt = startedAt;
+ }
+
+ public Instant getFinishedAt() {
+ return finishedAt;
+ }
+
+ public void setFinishedAt(Instant finishedAt) {
+ this.finishedAt = finishedAt;
+ }
+
+ public String getLastError() {
+ return lastError;
+ }
+
+ public void setLastError(String lastError) {
+ this.lastError = lastError;
+ }
+}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventType.java
similarity index 54%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventType.java
index c3d6813ab..e798d5b34 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingEventType.java
@@ -7,26 +7,17 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+package org.apache.custos.amie.model;
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+public enum ProcessingEventType {
+ DECODE_PACKET
}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingStatus.java
similarity index 54%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingStatus.java
index c3d6813ab..5d4cf3865 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/model/ProcessingStatus.java
@@ -7,26 +7,33 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
+package org.apache.custos.amie.model;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+public enum ProcessingStatus {
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+ /**
+ * The event has been created and is waiting to be processed by a worker.
+ */
+ NEW,
+ /**
+ * A worker has picked up the event and is actively processing it.
+ */
+ RUNNING,
+ /**
+ * The event was processed successfully.
+ */
+ SUCCEEDED,
+ /**
+ * The event failed processing and will not be automatically retried.
+ */
+ FAILED
}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/repo/PacketRepository.java
similarity index 50%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/repo/PacketRepository.java
index c3d6813ab..5929f0543 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/repo/PacketRepository.java
@@ -7,26 +7,29 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
+package org.apache.custos.amie.repo;
+
+import org.apache.custos.amie.model.PacketEntity;
+import org.apache.custos.amie.model.PacketStatus;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+@Repository
+public interface PacketRepository extends JpaRepository<PacketEntity, String> {
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+ Optional<PacketEntity> findByAmieId(Long amieId);
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+ Stream<PacketEntity> findTop50ByStatusOrderByReceivedAtAsc(PacketStatus status);
}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingErrorRepository.java
similarity index 55%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingErrorRepository.java
index c3d6813ab..a5e76e870 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingErrorRepository.java
@@ -7,26 +7,21 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
+package org.apache.custos.amie.repo;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+import org.apache.custos.amie.model.ProcessingErrorEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+@Repository
+public interface ProcessingErrorRepository extends JpaRepository<ProcessingErrorEntity, Long> {
}
diff --git a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java b/amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingEventRepository.java
similarity index 51%
copy from amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
copy to amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingEventRepository.java
index c3d6813ab..24c1fd854 100644
--- a/amie-decoder/src/main/java/org/apache/custos/amie/AmieDecoderApplication.java
+++ b/amie-decoder/src/main/java/org/apache/custos/amie/repo/ProcessingEventRepository.java
@@ -7,26 +7,26 @@
* "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
+ * 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.
- *
+ * specific language governing permissions and limitations
+ * under the License.
*/
-package org.apache.custos.amie;
+package org.apache.custos.amie.repo;
+
+import org.apache.custos.amie.model.ProcessingEventEntity;
+import org.apache.custos.amie.model.ProcessingStatus;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.stream.Stream;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+@Repository
+public interface ProcessingEventRepository extends JpaRepository<ProcessingEventEntity, String> {
-@SpringBootApplication
-@EnableJpaAuditing
-public class AmieDecoderApplication {
- public static void main(String[] args) {
- SpringApplication.run(AmieDecoderApplication.class, args);
- }
+ Stream<ProcessingEventEntity> findTop50ByStatusOrderByCreatedAtAsc(ProcessingStatus status);
}
diff --git a/amie-decoder/src/main/proto/amie_packets.proto b/amie-decoder/src/main/proto/amie_packets.proto
index 23c51d3ab..804c39e96 100644
--- a/amie-decoder/src/main/proto/amie_packets.proto
+++ b/amie-decoder/src/main/proto/amie_packets.proto
@@ -1,5 +1,6 @@
syntax = "proto3";
+option java_multiple_files = true;
package org.apache.custos.amie.v1;
import "google/protobuf/struct.proto";
diff --git a/amie-decoder/src/main/proto/internal_events.proto b/amie-decoder/src/main/proto/internal_events.proto
index 8c68ee929..5959ffceb 100644
--- a/amie-decoder/src/main/proto/internal_events.proto
+++ b/amie-decoder/src/main/proto/internal_events.proto
@@ -1,5 +1,6 @@
syntax = "proto3";
+option java_multiple_files = true;
package org.apache.custos.amie.internal.events.v1;
import "google/protobuf/timestamp.proto";
diff --git a/amie-decoder/src/main/resources/application.yml b/amie-decoder/src/main/resources/application.yml
index 163c6c485..3add2af62 100644
--- a/amie-decoder/src/main/resources/application.yml
+++ b/amie-decoder/src/main/resources/application.yml
@@ -28,8 +28,24 @@ spring:
jpa:
hibernate:
ddl-auto: update
+ properties:
+ hibernate.jdbc.time_zone: UTC
show-sql: false
database-platform: org.hibernate.dialect.MySQL8Dialect
+ flyway:
+ locations: classpath:db/migration
+
+app:
+ amie:
+ base-url: https://a3mdev.xsede.org/amie-api-test
+ site-code: ${AMIE_SITE_CODE}
+ api-key: ${AMIE_API_KEY}
+ scheduler:
+ poll-delay: 30s
+ worker-delay: 5s
+ client:
+ connect-timeout: 5s
+ read-timeout: 20s
logging:
pattern:
diff --git a/amie-decoder/src/main/resources/distribution/conf/application.yml b/amie-decoder/src/main/resources/distribution/conf/application.yml
index 163c6c485..3add2af62 100644
--- a/amie-decoder/src/main/resources/distribution/conf/application.yml
+++ b/amie-decoder/src/main/resources/distribution/conf/application.yml
@@ -28,8 +28,24 @@ spring:
jpa:
hibernate:
ddl-auto: update
+ properties:
+ hibernate.jdbc.time_zone: UTC
show-sql: false
database-platform: org.hibernate.dialect.MySQL8Dialect
+ flyway:
+ locations: classpath:db/migration
+
+app:
+ amie:
+ base-url: https://a3mdev.xsede.org/amie-api-test
+ site-code: ${AMIE_SITE_CODE}
+ api-key: ${AMIE_API_KEY}
+ scheduler:
+ poll-delay: 30s
+ worker-delay: 5s
+ client:
+ connect-timeout: 5s
+ read-timeout: 20s
logging:
pattern:
diff --git a/pom.xml b/pom.xml
index 6f6852512..a46cb3790 100644
--- a/pom.xml
+++ b/pom.xml
@@ -191,7 +191,7 @@
<springdoc.version>2.5.0</springdoc.version>
<protobuf.maven.plugin>0.6.1</protobuf.maven.plugin>
- <protobuf.version>3.24.0</protobuf.version>
+ <protobuf.version>4.32.1</protobuf.version>
<protobuf.json.version>0.9.15</protobuf.json.version>
<io.grpc.version>1.58.0</io.grpc.version>
<com.google.protobuf.util>3.21.11</com.google.protobuf.util>