[MODAUD-174] - Implements kafka handler and creating table

.gitignore

@@ -18,3 +18,4 @@ nbproject/
 .settings/
 .classpath
 /bin/
+/src/main/resources/postgres-conf.json

mod-audit-server/src/main/java/org/folio/rest/impl/InitAPIs.java

@@ -17,6 +17,7 @@
 import org.folio.verticle.SpringVerticleFactory;
 import org.folio.verticle.acquisition.OrderEventConsumersVerticle;
 import org.folio.verticle.acquisition.OrderLineEventConsumersVerticle;
+import org.folio.verticle.acquisition.PieceEventConsumersVerticle;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.support.AbstractApplicationContext;
 
@@ -30,6 +31,8 @@ public class InitAPIs implements InitAPI {
   private int acqOrderConsumerInstancesNumber;
   @Value("${acq.order-lines.kafka.consumer.instancesNumber:1}")
   private int acqOrderLineConsumerInstancesNumber;
+  @Value("${acq.pieces.kafka.consumer.instancesNumber:1}")
+  private int acqPieceConsumerInstancesNumber;
 
   @Override
   public void init(Vertx vertx, Context context, Handler<AsyncResult<Boolean>> handler) {
@@ -60,6 +63,7 @@ private Future<?> deployConsumersVerticles(Vertx vertx) {
 
     Promise<String> orderEventsConsumer = Promise.promise();
     Promise<String> orderLineEventsConsumer = Promise.promise();
+    Promise<String> pieceEventsConsumer = Promise.promise();
 
     vertx.deployVerticle(getVerticleName(verticleFactory, OrderEventConsumersVerticle.class),
       new DeploymentOptions()
@@ -71,10 +75,16 @@ private Future<?> deployConsumersVerticles(Vertx vertx) {
         .setWorker(true)
         .setInstances(acqOrderLineConsumerInstancesNumber), orderLineEventsConsumer);
 
+    vertx.deployVerticle(getVerticleName(verticleFactory, PieceEventConsumersVerticle.class),
+      new DeploymentOptions()
+        .setWorker(true)
+        .setInstances(acqPieceConsumerInstancesNumber), pieceEventsConsumer);
+
     LOGGER.info("deployConsumersVerticles:: All consumer verticles were successfully deployed");
     return GenericCompositeFuture.all(Arrays.asList(
       orderEventsConsumer.future(),
-      orderLineEventsConsumer.future()));
+      orderLineEventsConsumer.future(),
+      pieceEventsConsumer.future()));
   }
 
   private <T> String getVerticleName(VerticleFactory verticleFactory, Class<T> clazz) {

mod-audit-server/src/main/java/org/folio/verticle/acquisition/PieceEventConsumersVerticle.java

@@ -3,6 +3,7 @@
 import java.util.List;
 
 import org.folio.kafka.AsyncRecordHandler;
+import org.folio.kafka.KafkaConfig;
 import org.folio.util.AcquisitionEventType;
 import org.folio.verticle.AbstractConsumersVerticle;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -11,20 +12,18 @@
 @Component
 public class PieceEventConsumersVerticle extends AbstractConsumersVerticle {
 
-  private final AsyncRecordHandler<String, String> orderLineEventsHandler;
-
   @Autowired
-  public PieceEventConsumersVerticle(AsyncRecordHandler<String, String> orderLineEventsHandler) {
-    this.orderLineEventsHandler = orderLineEventsHandler;
-  }
+  private KafkaConfig kafkaConfig;
+  @Autowired
+  private AsyncRecordHandler<String, String> pieceEventsHandler;
 
   @Override
   public List<String> getEvents() {
-    return List.of(AcquisitionEventType.ACQ_PIECE_CHANGED.getTopicName());
+    return List.of(AcquisitionEventType.ACQ_ORDER_CHANGED.getTopicName());
   }
 
   @Override
   public AsyncRecordHandler<String, String> getHandler() {
-    return orderLineEventsHandler;
+    return pieceEventsHandler;
   }
 }

mod-audit-server/src/main/resources/templates/db_scripts/schema.json

@@ -74,6 +74,11 @@
       "run": "after",
       "snippetPath": "acquisition/create_acquisition_order_line_log_table.sql",
       "fromModuleVersion": "mod-audit-2.7.0"
+    },
+    {
+      "run": "after",
+      "snippetPath": "acquisition/create_acquisition_piece_log_table.sql",
+      "fromModuleVersion": "mod-audit-2.8.1"
     }
   ]
 }