From 0f53e29c323ce90f6a93e86359f8dcdcef1387f1 Mon Sep 17 00:00:00 2001
From: Cristian Bianco <cristian.bianco@smc.it>
Date: Wed, 28 Jun 2023 16:21:02 +0200
Subject: [PATCH] Issue #495: [ingestion]add last field

---
 .../ingestion/dto/IngestionPayload.java       |  1 +
 .../ingestion/web/IngestionEmitter.java       | 20 ++++++-------------
 .../ingestion/src/main/proto/ingestion.proto  |  1 +
 3 files changed, 8 insertions(+), 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
index 9d7f64d7d..fc2e1265e 100644
--- 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;
 }
\ No newline at end of file
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
index 2c0a633c7..79b51d2cb 100644
--- 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
index 3ac4fc6df..f44f0e9c7 100644
--- 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 {