Issue #495: [ingestion]add last field

smclab · Jun 28, 2023 · 0f53e29 · 0f53e29
1 parent 851bb23
commit 0f53e29
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 14 deletions.
diff --git a/core/app/ingestion/src/main/java/io/openk9/ingestion/dto/IngestionPayload.java b/core/app/ingestion/src/main/java/io/openk9/ingestion/dto/IngestionPayload.java
@@ -43,4 +43,5 @@ public  class IngestionPayload {
 	private ResourcesPayload resources;
 	private Map<String, List<String>> acl;
 	private String scheduleId;
+	private boolean last;
 }
diff --git a/core/app/ingestion/src/main/java/io/openk9/ingestion/web/IngestionEmitter.java b/core/app/ingestion/src/main/java/io/openk9/ingestion/web/IngestionEmitter.java
@@ -17,23 +17,13 @@
 
 package io.openk9.ingestion.web;
 
-import io.openk9.ingestion.dto.BinaryDTO;
-import io.openk9.ingestion.dto.BinaryPayload;
-import io.openk9.ingestion.dto.IngestionDTO;
-import io.openk9.ingestion.dto.IngestionPayload;
-import io.openk9.ingestion.dto.IngestionPayloadWrapper;
-import io.openk9.ingestion.dto.ResourcesDTO;
-import io.openk9.ingestion.dto.ResourcesPayload;
+import io.openk9.ingestion.dto.*;
 import io.openk9.ingestion.grpc.Binary;
 import io.openk9.ingestion.grpc.IngestionRequest;
 import io.openk9.ingestion.grpc.Resources;
 import io.smallrye.reactive.messaging.rabbitmq.OutgoingRabbitMQMetadata;
 import io.vertx.core.json.JsonObject;
-import org.eclipse.microprofile.reactive.messaging.Channel;
-import org.eclipse.microprofile.reactive.messaging.Emitter;
-import org.eclipse.microprofile.reactive.messaging.Message;
-import org.eclipse.microprofile.reactive.messaging.Metadata;
-import org.eclipse.microprofile.reactive.messaging.OnOverflow;
+import org.eclipse.microprofile.reactive.messaging.*;
 import org.jboss.logging.Logger;
 
 import javax.enterprise.context.ApplicationScoped;
@@ -106,7 +96,8 @@ private IngestionPayloadWrapper _of(IngestionRequest dto) {
 					.toArray(new String[0]),
 				_dtoToPayload(dto.getResources()),
 				mappingAcl,
-				dto.getScheduleId()
+				dto.getScheduleId(),
+				dto.getLast()
 			)
 		);
 	}
@@ -126,7 +117,8 @@ private IngestionPayloadWrapper _of(IngestionDTO dto) {
 					.toArray(new String[0]),
 				_dtoToPayload(dto.getResources()),
 				dto.getAcl(),
-				dto.getScheduleId()
+				dto.getScheduleId(),
+				dto.isLast()
 			)
 		);
 	}

diff --git a/core/app/ingestion/src/main/proto/ingestion.proto b/core/app/ingestion/src/main/proto/ingestion.proto
@@ -25,6 +25,7 @@ message IngestionRequest {
   map<string, Acl> acl  = 7;
   string scheduleId = 8;
   string tenantId = 9;
+  bool last = 10;
 }
 
 message Acl {