CIRCSTORE-527 Merge ecs-tlr-feature branch into master

ramls/request-queue-reordering.json

@@ -0,0 +1,35 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "title": "Requests batch update",
+  "description": "List of ids reordered requests",
+  "type": "object",
+  "properties": {
+    "instanceId": {
+      "description": "Instance ID of reordered requests",
+      "type": "string",
+      "$ref": "raml-util/schemas/uuid.schema"
+    },
+    "itemId": {
+      "description": "Item ID of reordered requests",
+      "type": "string",
+      "$ref": "raml-util/schemas/uuid.schema"
+    },
+    "requestLevel": {
+      "description": "Level of the request - Item or Title",
+      "type": "string",
+      "enum": ["Item", "Title"]
+    },
+    "requestIds": {
+      "description": "Array of requests ids",
+      "type": "array",
+      "items": {
+        "type": "string",
+        "$ref": "raml-util/schemas/uuid.schema"
+      }
+    }
+  },
+  "additionalProperties": false,
+  "required": [
+    "requestLevel"
+  ]
+}

ramls/request-storage-batch.raml

@@ -11,6 +11,7 @@ documentation:
 types:
   requests-batch: !include requests-batch.json
   errors: !include raml-util/schemas/errors.schema
+  request-queue-reordering: !include request-queue-reordering.json
 
 traits:
   validate: !include raml-util/traits/validation.raml

ramls/request.json

@@ -18,6 +18,11 @@
       "type": "string",
       "enum": ["Hold", "Recall", "Page"]
     },
+    "ecsRequestPhase": {
+      "description": "Stage in ECS request process, absence of this field means this is a single-tenant request",
+      "type": "string",
+      "enum": ["Primary", "Secondary"]
+    },
     "requestDate": {
       "description": "Date the request was made",
       "type": "string",

src/main/java/org/folio/rest/impl/CirculationSettingsAPI.java

@@ -47,28 +47,28 @@ public void getCirculationSettingsStorageCirculationSettingsByCirculationSetting
     String circulationSettingsId, Map<String, String> okapiHeaders,
     Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {
 
-      new CirculationSettingsService(vertxContext, okapiHeaders)
-        .findById(circulationSettingsId)
-        .onComplete(asyncResultHandler);
+    new CirculationSettingsService(vertxContext, okapiHeaders)
+      .findById(circulationSettingsId)
+      .onComplete(asyncResultHandler);
   }
 
   @Override
   public void putCirculationSettingsStorageCirculationSettingsByCirculationSettingsId(
     String circulationSettingsId, CirculationSetting entity, Map<String, String> okapiHeaders,
     Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {
 
-      new CirculationSettingsService(vertxContext, okapiHeaders)
-        .update(circulationSettingsId, entity)
-        .onComplete(asyncResultHandler);
+    new CirculationSettingsService(vertxContext, okapiHeaders)
+      .update(circulationSettingsId, entity)
+      .onComplete(asyncResultHandler);
   }
 
   @Override
   public void deleteCirculationSettingsStorageCirculationSettingsByCirculationSettingsId(
     String circulationSettingsId, Map<String, String> okapiHeaders,
     Handler<AsyncResult<Response>> asyncResultHandler, Context vertxContext) {
 
-      new CirculationSettingsService(vertxContext, okapiHeaders)
-        .delete(circulationSettingsId)
-        .onComplete(asyncResultHandler);
+    new CirculationSettingsService(vertxContext, okapiHeaders)
+      .delete(circulationSettingsId)
+      .onComplete(asyncResultHandler);
   }
 }

src/main/java/org/folio/rest/impl/RequestsBatchAPI.java

@@ -6,17 +6,14 @@
 import static org.folio.rest.jaxrs.resource.RequestStorageBatch.PostRequestStorageBatchRequestsResponse.respond201;
 import static org.folio.rest.jaxrs.resource.RequestStorageBatch.PostRequestStorageBatchRequestsResponse.respond422WithApplicationJson;
 import static org.folio.rest.jaxrs.resource.RequestStorageBatch.PostRequestStorageBatchRequestsResponse.respond500WithTextPlain;
-import static org.folio.rest.tools.utils.TenantTool.tenantId;
 
 import java.util.Map;
 
 import javax.ws.rs.core.Response;
 import org.folio.rest.annotations.Validate;
 import org.folio.rest.jaxrs.model.RequestsBatch;
 import org.folio.rest.jaxrs.resource.RequestStorageBatch;
-import org.folio.rest.persist.PgUtil;
 import org.folio.rest.tools.utils.MetadataUtil;
-import org.folio.service.BatchResourceService;
 import org.folio.service.request.RequestBatchResourceService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,7 +23,7 @@
 import io.vertx.core.Handler;
 
 public class RequestsBatchAPI implements RequestStorageBatch {
-  private static final Logger LOG = LoggerFactory.getLogger(RequestsBatchAPI.class);
+  private static final Logger log = LoggerFactory.getLogger(RequestsBatchAPI.class);
 
   @Validate
   @Override
@@ -38,36 +35,30 @@ public void postRequestStorageBatchRequests(
       MetadataUtil.populateMetadata(entity.getRequests(), okapiHeaders);
     } catch (Throwable e) {
       String msg = "Cannot populate metadata of request list elements: " + e.getMessage();
-      LOG.error(msg, e);
+      log.error(msg, e);
       asyncResultHandler.handle(succeededFuture(respond500WithTextPlain(msg)));
       return;
     }
 
-    BatchResourceService batchUpdateService = new BatchResourceService(
-      PgUtil.postgresClient(context, okapiHeaders)
-    );
-
-    RequestBatchResourceService requestBatchUpdateService =
-      new RequestBatchResourceService(tenantId(okapiHeaders), batchUpdateService);
-
-    requestBatchUpdateService.executeRequestBatchUpdate(entity.getRequests(),
-      updateResult -> {
+    log.info("postRequestStorageBatchRequests:: requests: {}", entity.getRequests());
+    new RequestBatchResourceService(context, okapiHeaders)
+      .executeRequestBatchUpdate(entity.getRequests(), updateResult -> {
         // Successfully updated
         if (updateResult.succeeded()) {
-          LOG.debug("Batch update executed successfully");
+          log.debug("Batch update executed successfully");
           asyncResultHandler.handle(succeededFuture(respond201()));
           return;
         }
 
         // Update failed due to can not have more then one request in the same position
         if (hasSamePositionConstraintViolated(updateResult.cause())) {
-          LOG.warn("Same position constraint violated", updateResult.cause());
+          log.warn("Same position constraint violated", updateResult.cause());
           asyncResultHandler.handle(succeededFuture(
             respond422WithApplicationJson(samePositionInQueueError(null, null))
           ));
         } else {
           // Other failure occurred
-          LOG.warn("Unhandled error occurred during update", updateResult.cause());
+          log.warn("Unhandled error occurred during update", updateResult.cause());
           asyncResultHandler.handle(succeededFuture(
             respond500WithTextPlain(updateResult.cause().getMessage())
           ));

src/main/java/org/folio/service/BatchResourceService.java

@@ -23,7 +23,7 @@
 import io.vertx.sqlclient.RowSet;
 
 public class BatchResourceService {
-  private static final Logger LOG = LoggerFactory.getLogger(BatchResourceService.class);
+  private static final Logger log = LoggerFactory.getLogger(BatchResourceService.class);
   private static final String WHERE_CLAUSE = "WHERE id = '%s'";
 
   private final PostgresClient postgresClient;
@@ -38,14 +38,17 @@ public BatchResourceService(PostgresClient postgresClient) {
    * @param batchFactories  - Factory to create a batch update chunk.
    * @param onFinishHandler - Callback.
    */
-  public void executeBatchUpdate(
+  public Future<Void> executeBatchUpdate(
     List<Function<SQLConnection, Future<RowSet<Row>>>> batchFactories,
     Handler<AsyncResult<Void>> onFinishHandler) {
 
+    Promise<Void> promise = Promise.promise();
+
     postgresClient.startTx(connectionResult -> {
       if (connectionResult.failed()) {
-        LOG.warn("Can not start transaction", connectionResult.cause());
+        log.warn("Cannot start transaction", connectionResult.cause());
         onFinishHandler.handle(failedFuture(connectionResult.cause()));
+        promise.fail(connectionResult.cause());
         return;
       }
 
@@ -60,21 +63,32 @@ public void executeBatchUpdate(
       // Handle overall update result and decide on whether to commit or rollback transaction
       lastUpdate.onComplete(updateResult -> {
         if (updateResult.failed()) {
-          LOG.warn("Batch update rejected", updateResult.cause());
+          log.warn("Batch update rejected", updateResult.cause());
 
           // Rollback transaction and keep original cause.
-          postgresClient.rollbackTx(connectionResult,
-            rollback -> onFinishHandler.handle(failedFuture(updateResult.cause()))
-          );
+          postgresClient.rollbackTx(connectionResult, rollback -> {
+            onFinishHandler.handle(failedFuture(updateResult.cause()));
+            promise.fail(updateResult.cause());
+          });
         } else {
-          LOG.debug("Update successful, committing transaction");
-
-          postgresClient.endTx(connectionResult, onFinishHandler);
+          log.debug("Update successful, committing transaction");
+
+          postgresClient.endTx(connectionResult, commitResult -> {
+            if (commitResult.succeeded()) {
+              onFinishHandler.handle(succeededFuture());
+              promise.complete();
+            } else {
+              log.warn("Failed to commit transaction", commitResult.cause());
+              onFinishHandler.handle(failedFuture(commitResult.cause()));
+              promise.fail(commitResult.cause());
+            }
+          });
         }
       });
     });
-  }
 
+    return promise.future();
+  }
   /**
    * Creates update single entity batch function.
    *
@@ -92,7 +106,7 @@ public <T> Function<SQLConnection, Future<RowSet<Row>>> updateSingleEntityBatchF
       final Promise<RowSet<Row>> promise = promise();
       final Future<SQLConnection> connectionResult = succeededFuture(connection);
 
-      LOG.debug("Updating entity {} with id {}", entity, id);
+      log.debug("Updating entity {} with id {}", entity, id);
 
       postgresClient.update(connectionResult, tableName, entity, "jsonb",
         String.format(WHERE_CLAUSE, id), false, promise);
@@ -113,7 +127,7 @@ public Function<SQLConnection, Future<RowSet<Row>>> queryWithParamsBatchFactory(
     String query, Collection<?> params) {
 
     return connection -> {
-      LOG.debug("Executing SQL [{}], got [{}] parameters", query, params.size());
+      log.debug("Executing SQL [{}], got [{}] parameters", query, params.size());
 
       final Promise<RowSet<Row>> promise = promise();
       final Future<SQLConnection> connectionResult = succeededFuture(connection);

src/main/java/org/folio/service/CirculationSettingsService.java

@@ -60,17 +60,21 @@ public Future<Response> findById(String circulationSettingsId) {
       GetCirculationSettingsStorageCirculationSettingsByCirculationSettingsIdResponse.class);
   }
 
-  public Future<Response> update(String circulationSettingsId, CirculationSetting circulationSetting) {
-    return PgUtil.put(CIRCULATION_SETTINGS_TABLE, circulationSetting, circulationSettingsId, okapiHeaders, vertxContext,
+  public Future<Response> update(String circulationSettingsId,
+    CirculationSetting circulationSetting) {
+
+    return PgUtil.put(CIRCULATION_SETTINGS_TABLE, circulationSetting, circulationSettingsId,
+        okapiHeaders, vertxContext,
         PutCirculationSettingsStorageCirculationSettingsByCirculationSettingsIdResponse.class)
       .compose(eventPublisher.publishUpdated(circulationSetting));
   }
 
   public Future<Response> delete(String circulationSettingsId) {
-    return repository.getById(circulationSettingsId).compose (
-      circulationSetting -> PgUtil.deleteById(CIRCULATION_SETTINGS_TABLE, circulationSettingsId, okapiHeaders, vertxContext,
-        DeleteCirculationSettingsStorageCirculationSettingsByCirculationSettingsIdResponse.class)
-      .compose(eventPublisher.publishRemoved(circulationSetting))
+    return repository.getById(circulationSettingsId)
+      .compose(circulationSetting -> PgUtil.deleteById(CIRCULATION_SETTINGS_TABLE,
+          circulationSettingsId, okapiHeaders, vertxContext,
+          DeleteCirculationSettingsStorageCirculationSettingsByCirculationSettingsIdResponse.class)
+        .compose(eventPublisher.publishRemoved(circulationSetting))
     );
   }
 
@@ -109,4 +113,4 @@ private Future<List<CirculationSetting>> getSettingsByName(String settingsName)
 
     return repository.get(filter);
   }
-}
+}

src/main/java/org/folio/service/event/EntityChangedEventPublisherFactory.java

@@ -5,6 +5,7 @@
 import static org.folio.support.kafka.topic.CirculationStorageKafkaTopic.CIRCULATION_SETTINGS;
 import static org.folio.support.kafka.topic.CirculationStorageKafkaTopic.LOAN;
 import static org.folio.support.kafka.topic.CirculationStorageKafkaTopic.REQUEST;
+import static org.folio.support.kafka.topic.CirculationStorageKafkaTopic.REQUEST_QUEUE_REORDERING;
 import static org.folio.support.kafka.topic.CirculationStorageKafkaTopic.RULES;
 
 import java.util.Map;
@@ -19,6 +20,7 @@
 import org.folio.rest.jaxrs.model.CirculationSetting;
 import org.folio.rest.jaxrs.model.Loan;
 import org.folio.rest.jaxrs.model.Request;
+import org.folio.rest.jaxrs.model.RequestQueueReordering;
 
 import io.vertx.core.Context;
 import lombok.extern.log4j.Log4j2;
@@ -53,6 +55,15 @@ public static EntityChangedEventPublisher<String, Request> requestEventPublisher
         new RequestRepository(vertxContext, okapiHeaders));
   }
 
+  public static EntityChangedEventPublisher<String, RequestQueueReordering>
+  requestBatchEventPublisher(Context vertxContext, Map<String, String> okapiHeaders) {
+
+    return new EntityChangedEventPublisher<>(okapiHeaders, RequestQueueReordering::getInstanceId,
+      NULL_ID, new EntityChangedEventFactory<>(), new DomainEventPublisher<>(vertxContext,
+      REQUEST_QUEUE_REORDERING.fullTopicName(tenantId(okapiHeaders)),
+      FailureHandler.noOperation()), null);
+  }
+
   public static EntityChangedEventPublisher<String, CheckIn> checkInEventPublisher(
       Context vertxContext, Map<String, String> okapiHeaders) {