adding warehouse message

gbif/pipelines#1078

Author: fmendezh

src/main/java/org/gbif/common/messaging/DefaultMessageRegistry.java

@@ -104,6 +104,7 @@ public class DefaultMessageRegistry implements MessageRegistry {
     messageToExchangeMappingInternal.put(CamtrapDpDownloadFinishedMessage.class, "crawler");
     messageToExchangeMappingInternal.put(DownloadLauncherMessage.class, "occurrence");
     messageToExchangeMappingInternal.put(DownloadCancelMessage.class, "occurrence");
+    messageToExchangeMappingInternal.put(DataWarehouseMessage.class, "occurrence");
     MESSAGE_TO_EXCHANGE_MAPPING = Collections.unmodifiableMap(messageToExchangeMappingInternal);
 
     Map<Class<? extends Message>, String> messageToRoutingKeyMapping = new HashMap<>();
@@ -186,6 +187,7 @@ public class DefaultMessageRegistry implements MessageRegistry {
     messageToRoutingKeyMapping.put(
         DownloadLauncherMessage.class, DownloadLauncherMessage.ROUTING_KEY);
     messageToRoutingKeyMapping.put(DownloadCancelMessage.class, DownloadCancelMessage.ROUTING_KEY);
+    messageToRoutingKeyMapping.put(DataWarehouseMessage.class, DataWarehouseMessage.ROUTING_KEY);
     MESSAGE_TO_ROUTING_KEY_MAPPING = Collections.unmodifiableMap(messageToRoutingKeyMapping);
   }
 

src/main/java/org/gbif/common/messaging/api/messages/DataWarehouseMessage.java

@@ -0,0 +1,40 @@
+package org.gbif.common.messaging.api.messages;
+
+import org.gbif.api.model.pipelines.PipelinesWorkflow;
+import org.gbif.api.model.pipelines.StepType;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.SneakyThrows;
+
+public class DataWarehouseMessage extends PipelinesHdfsViewMessage {
+
+  public DataWarehouseMessage() {}
+
+  @JsonCreator
+  public DataWarehouseMessage(
+      @JsonProperty("datasetUuid") UUID datasetUuid,
+      @JsonProperty("attempt") int attempt,
+      @JsonProperty("pipelineSteps") Set<String> pipelineSteps,
+      @JsonProperty("runner") String runner,
+      @JsonProperty("executionId") Long executionId) {
+    super(datasetUuid, attempt, pipelineSteps, runner, executionId);
+  }
+
+  @SneakyThrows
+  public static void main(String[] args) {
+    DataWarehouseMessage message = new DataWarehouseMessage(UUID.fromString("a92f4b3c-ae5c-45af-8dac-4a6a88d35ddd"), 25, Collections.singleton("DATA_WAREHOUSE"), "DISTRIBUTED" ,814400L);
+    ObjectMapper mapper = new ObjectMapper();
+    System.out.println(mapper.writeValueAsString(message));
+    PipelinesWorkflow.Graph<StepType> workflow = PipelinesWorkflow.getOccurrenceWorkflow();
+    List<PipelinesWorkflow.Graph<StepType>.Edge> edges = workflow.getNodeEdges(StepType.HDFS_VIEW);
+    System.out.println(edges);
+  }
+}