CIRCSTORE-540: Extended ItemUpdateProcessorForRequest to sync item lo…

…cation and SP updates in Request records (#503)

* CIRCSTORE-540: ItemUpdateProcessorForRequest update item's location and SP

* CIRCSTORE-540: added log for debugging

* CIRCSTORE-540: added log for debugging

* CIRCSTORE-540: set additionalProperties true inside location.json

* CIRCSTORE-540: fix exception issue

* CIRCSTORE-540: fix exception issue

* CIRCSTORE-540: 1) added async mechanism in EventProcessor class to execute and support non-blocking api calls. 2) Added test case.

* CIRCSTORE-540: removed unnecessary log and code

* CIRCSTORE-540: using location and SP constants

* CIRCSTORE-540: fix NLP due to logging of null object

* CIRCSTORE-540: fixed sonar issues

* CIRCSTORE-540: added interface dependency for location and service-point api

descriptors/ModuleDescriptor-template.json

@@ -29,6 +29,14 @@
     {
       "id": "configuration",
       "version": "2.0"
+    },
+    {
+      "id": "locations",
+      "version": "3.1"
+    },
+    {
+      "id": "service-points",
+      "version": "3.4"
     }
   ],
   "provides": [

ramls/locations/location.json

@@ -0,0 +1,67 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "description": "A (shelf) location, the forth-level location unit below institution, campus, and library.",
+  "javaType": "org.folio.rest.jaxrs.model.Location",
+  "type": "object",
+  "properties": {
+    "id": {
+      "description": "id of this (shelf) location record as UUID.",
+      "type": "string"
+    },
+    "name": {
+      "description": "Name of the (shelf) location",
+      "type": "string"
+    },
+    "code": {
+      "description": "Code of the (shelf) location, usually an abbreviation of the name.",
+      "type": "string"
+    },
+    "description": {
+      "description": "Description of the (shelf) location.",
+      "type": "string"
+    },
+    "discoveryDisplayName": {
+      "description": "Name of the (shelf) location to be shown in the discovery.",
+      "type": "string"
+    },
+    "isActive": {
+      "description": "Whether this (shelf) location is active. Inactive (shelf) locations can no longer been used.",
+      "type": "boolean"
+    },
+    "institutionId": {
+      "description": "The UUID of the institution, the first-level location unit, this (shelf) location belongs to.",
+      "type": "string"
+    },
+    "campusId": {
+      "description": "The UUID of the campus, the second-level location unit, this (shelf) location belongs to.",
+      "type": "string"
+    },
+    "libraryId": {
+      "description": "The UUID of the library, the third-level location unit, this (shelf) location belongs to.",
+      "type": "string"
+    },
+    "primaryServicePoint": {
+      "description": "The UUID of the primary service point of this (shelf) location.",
+      "format": "uuid",
+      "type": "string"
+    },
+    "servicePointIds": {
+      "description": "All service points that this (shelf) location has.",
+      "type": "array",
+      "items": {
+        "description": "The UUID of a service point that belongs to this (shelf) location.",
+        "type": "string",
+        "format": "uuid",
+        "not": {
+          "type": "null"
+        }
+      }
+    },
+    "metadata": {
+      "type": "object",
+      "$ref": "../raml-util/schemas/metadata.schema",
+      "readonly": true
+    }
+  },
+  "additionalProperties": true
+}

ramls/locations/locations.json

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "description": "List of (shelf) locations.",
+  "type": "object",
+  "properties": {
+    "locations": {
+      "id": "locations",
+      "description": "List of (shelf) locations.",
+      "type": "array",
+      "items": {
+        "type": "object",
+        "$ref": "location.json"
+      }
+    },
+    "totalRecords": {
+      "description": "Estimated or exact total number of records",
+      "type": "integer"
+    }
+  },
+  "required": [
+    "locations",
+    "totalRecords"
+  ]
+}

ramls/request-storage.raml

@@ -13,6 +13,8 @@ types:
   requests: !include requests.json
   errors: !include raml-util/schemas/errors.schema
   parameters: !include raml-util/schemas/parameters.schema
+  location: !include locations/location.json
+  locations: !include locations/locations.json
 
 traits:
   pageable: !include raml-util/traits/pageable.raml

src/main/java/org/folio/rest/client/InventoryStorageClient.java

@@ -3,6 +3,7 @@
 import java.util.Collection;
 import java.util.Map;
 
+import org.folio.rest.jaxrs.model.Location;
 import org.folio.rest.jaxrs.model.Servicepoint;
 
 import io.vertx.core.Future;
@@ -12,6 +13,9 @@ public class InventoryStorageClient extends OkapiClient {
   private static final String SERVICE_POINTS_URL = "/service-points";
   private static final String SERVICE_POINTS_COLLECTION_NAME = "servicepoints";
 
+  private static final String LOCATION_URL = "/locations";
+  private static final String LOCATION_COLLECTION_NAME = "locations";
+
   public InventoryStorageClient(Vertx vertx, Map<String, String> okapiHeaders) {
     super(vertx, okapiHeaders);
   }
@@ -20,4 +24,8 @@ public Future<Collection<Servicepoint>> getServicePoints(Collection<String> ids)
     return get(SERVICE_POINTS_URL, ids, SERVICE_POINTS_COLLECTION_NAME, Servicepoint.class);
   }
 
+  public Future<Collection<Location>> getLocations(Collection<String> ids) {
+    return get(LOCATION_URL, ids, LOCATION_COLLECTION_NAME, Location.class);
+  }
+
 }

src/main/java/org/folio/service/event/handler/ItemUpdateEventHandler.java

@@ -5,6 +5,7 @@
 import org.apache.commons.collections4.map.CaseInsensitiveMap;
 import org.folio.kafka.AsyncRecordHandler;
 import org.folio.persist.RequestRepository;
+import org.folio.rest.client.InventoryStorageClient;
 import org.folio.service.event.handler.processor.ItemUpdateProcessorForRequest;
 
 import io.vertx.core.Context;
@@ -14,7 +15,6 @@
 
 public class ItemUpdateEventHandler implements AsyncRecordHandler<String, String> {
   private final Context context;
-
   public ItemUpdateEventHandler(Context context) {
     this.context = context;
   }
@@ -24,9 +24,8 @@ public Future<String> handle(KafkaConsumerRecord<String, String> kafkaConsumerRe
     JsonObject payload = new JsonObject(kafkaConsumerRecord.value());
     CaseInsensitiveMap<String, String> headers =
       new CaseInsensitiveMap<>(kafkaHeadersToMap(kafkaConsumerRecord.headers()));
-
     ItemUpdateProcessorForRequest itemUpdateProcessorForRequest =
-      new ItemUpdateProcessorForRequest(new RequestRepository(context, headers));
+      new ItemUpdateProcessorForRequest(new RequestRepository(context, headers), new InventoryStorageClient(context.owner(), headers));
 
     return itemUpdateProcessorForRequest.run(kafkaConsumerRecord.key(), payload);
   }

src/main/java/org/folio/service/event/handler/processor/EventProcessor.java

@@ -49,20 +49,24 @@ private Future<List<T>> processEvent(JsonObject payload) {
       return succeededFuture();
     }
 
-    List<Change<T>> relevantChanges = collectRelevantChanges(payload);
+    Future<List<Change<T>>> relevantChangesFuture =
+      collectRelevantChanges(payload);
 
-    if (relevantChanges.isEmpty()) {
-      log.info("processEvent:: no relevant changes detected");
-      return succeededFuture();
-    }
+    return relevantChangesFuture.compose(relevantChanges -> {
+
+      if (relevantChanges.isEmpty()) {
+        log.info("processEvent:: no relevant changes detected");
+        return succeededFuture();
+      }
 
-    log.info("processEvent:: {} relevant changes detected, applying", relevantChanges::size);
-    return applyChanges(relevantChanges, payload);
+      log.info("processEvent:: {} relevant changes detected, applying", relevantChanges::size);
+      return applyChanges(relevantChanges, payload);
+    });
   }
 
   protected abstract boolean validatePayload(JsonObject payload);
 
-  protected abstract List<Change<T>> collectRelevantChanges(JsonObject payload);
+  protected abstract Future<List<Change<T>>> collectRelevantChanges(JsonObject payload);
 
   private Future<List<T>> applyChanges(List<Change<T>> changes, JsonObject payload) {
     log.debug("applyChanges:: payload: {}", payload);

src/main/java/org/folio/service/event/handler/processor/ItemUpdateProcessorForRequest.java

@@ -1,16 +1,28 @@
 package org.folio.service.event.handler.processor;
 
+import static io.vertx.core.Future.succeededFuture;
 import static org.apache.commons.lang3.ObjectUtils.notEqual;
 import static org.folio.service.event.InventoryEventType.INVENTORY_ITEM_UPDATED;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
+import io.vertx.core.Future;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.folio.persist.RequestRepository;
+import org.folio.rest.client.InventoryStorageClient;
 import org.folio.rest.jaxrs.model.CallNumberComponents;
+import org.folio.rest.jaxrs.model.Item;
+import org.folio.rest.jaxrs.model.Location;
 import org.folio.rest.jaxrs.model.Request;
+import org.folio.rest.jaxrs.model.Servicepoint;
 import org.folio.rest.persist.Criteria.Criteria;
 import org.folio.rest.persist.Criteria.Criterion;
 
@@ -19,17 +31,26 @@
 public class ItemUpdateProcessorForRequest extends UpdateEventProcessor<Request> {
 
   private static final Logger log = LogManager.getLogger(ItemUpdateProcessorForRequest.class);
+  private final InventoryStorageClient inventoryStorageClient;
+
   private static final String EFFECTIVE_SHELVING_ORDER_KEY = "effectiveShelvingOrder";
   private static final String EFFECTIVE_CALL_NUMBER_COMPONENTS_KEY = "effectiveCallNumberComponents";
   private static final String CALL_NUMBER_KEY = "callNumber";
   private static final String CALL_NUMBER_PREFIX_KEY = "prefix";
   private static final String CALL_NUMBER_SUFFIX_KEY = "suffix";
 
-  public ItemUpdateProcessorForRequest(RequestRepository repository) {
+  public static final String ITEM_EFFECTIVE_LOCATION_ID = "itemEffectiveLocationId";
+  public static final String ITEM_EFFECTIVE_LOCATION_NAME = "itemEffectiveLocationName";
+  public static final String RETRIEVAL_SERVICE_POINT_ID = "retrievalServicePointId";
+  public static final String RETRIEVAL_SERVICE_POINT_NAME = "retrievalServicePointName";
+
+  public ItemUpdateProcessorForRequest(RequestRepository repository, InventoryStorageClient inventoryStorageClient) {
     super(INVENTORY_ITEM_UPDATED, repository);
+    this.inventoryStorageClient = inventoryStorageClient;
   }
 
-  protected List<Change<Request>> collectRelevantChanges(JsonObject payload) {
+  @Override
+  protected Future<List<Change<Request>>> collectRelevantChanges(JsonObject payload) {
     JsonObject oldObject = payload.getJsonObject("old");
     JsonObject newObject = payload.getJsonObject("new");
 
@@ -50,9 +71,69 @@ protected List<Change<Request>> collectRelevantChanges(JsonObject payload) {
       changes.add(new Change<>(request -> request.getSearchIndex().setShelvingOrder(newShelvingOrder)));
     }
 
-    return changes;
+    Future<Map<String, String>> fetchLocationAndServicePoint = updateItemAndServicePoint(newObject);
+    return fetchLocationAndServicePoint
+            .compose(locationAndSpData -> addLocationAndServicePointChanges(locationAndSpData, changes))
+            .compose(r -> Future.succeededFuture(changes))
+            .recover(throwable -> Future.succeededFuture(changes));
+  }
+
+  private static Future<List<Change<Request>>> addLocationAndServicePointChanges(Map<String, String> locationAndSpData, List<Change<Request>> changes) {
+    log.info("ItemUpdateProcessorForRequest :: locationAndSpData: {}", locationAndSpData);
+    changes.add(new Change<>(request -> {
+      if (request.getItem() == null) {
+        request.setItem(new Item());
+      }
+      request.getItem().setItemEffectiveLocationId(locationAndSpData.get(ITEM_EFFECTIVE_LOCATION_ID));
+      request.getItem().setItemEffectiveLocationName(locationAndSpData.get(ITEM_EFFECTIVE_LOCATION_NAME));
+      request.getItem().setRetrievalServicePointId(locationAndSpData.get(RETRIEVAL_SERVICE_POINT_ID));
+      request.getItem().setRetrievalServicePointName(locationAndSpData.get(RETRIEVAL_SERVICE_POINT_NAME));
+    }));
+    return Future.succeededFuture(changes);
+  }
+
+  private Future<Map<String, String>> updateItemAndServicePoint(JsonObject newObject) {
+    String effectiveLocationId = newObject.getString("effectiveLocationId");
+    Map<String, String> locationAndSpData = new HashMap<>();
+    locationAndSpData.put(ITEM_EFFECTIVE_LOCATION_ID, effectiveLocationId);
+    return inventoryStorageClient.getLocations(Collections.singletonList(effectiveLocationId))
+            .compose(locations -> setEffectiveLocationData(locations, effectiveLocationId, locationAndSpData))
+            .compose(primaryServicePoint -> setRetrievalServicePointData(primaryServicePoint, locationAndSpData))
+            .compose(e -> Future.succeededFuture(locationAndSpData))
+            .onFailure(throwable -> log.info("ItemUpdateProcessorForRequest :: Error while fetching Locations: {}", throwable.toString()));
+  }
+
+  private static Future<String> setEffectiveLocationData(Collection<Location> locations, String effectiveLocationId,
+                                                         Map<String, String> locationAndSpData) {
+    Location effectiveLocation = locations.stream()
+            .filter(l -> l.getId().equals(effectiveLocationId))
+            .findFirst().orElse(null);
+    if (Objects.nonNull(effectiveLocation)) {
+      locationAndSpData.put(ITEM_EFFECTIVE_LOCATION_NAME, effectiveLocation.getName());
+      return succeededFuture(effectiveLocation.getPrimaryServicePoint().toString());
+    }
+    return succeededFuture();
+  }
+
+  private Future<Object> setRetrievalServicePointData(String primaryServicePoint, Map<String, String> locationAndSpData) {
+    if (!StringUtils.isBlank(primaryServicePoint)) {
+      locationAndSpData.put(RETRIEVAL_SERVICE_POINT_ID, primaryServicePoint);
+      return inventoryStorageClient.getServicePoints(Collections.singletonList(primaryServicePoint))
+              .compose(servicePoints -> {
+                Servicepoint retrievalServicePoint = servicePoints.stream()
+                        .filter(sp -> sp.getId().equals(primaryServicePoint))
+                        .findFirst().orElse(null);
+                if (Objects.nonNull(retrievalServicePoint)) {
+                  locationAndSpData.put(RETRIEVAL_SERVICE_POINT_NAME, retrievalServicePoint.getName());
+                }
+                return succeededFuture();
+              }).onFailure(throwable -> log.info("ItemUpdateProcessorForRequest :: Error while fetching ServicePoint: {}",
+                      throwable.toString()));
+    }
+    return succeededFuture();
   }
 
+
   @Override
   protected Criterion criterionForObjectsToBeUpdated(String oldObjectId) {
     log.info("criteriaForObjectsToBeUpdated:: oldObjectId: {}", oldObjectId);

src/main/java/org/folio/service/event/handler/processor/ServicePointDeleteProcessorForRequestPolicy.java

@@ -7,6 +7,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import io.vertx.core.Future;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.folio.persist.RequestPolicyRepository;
@@ -27,7 +28,7 @@ public ServicePointDeleteProcessorForRequestPolicy(
   }
 
   @Override
-  protected List<Change<RequestPolicy>> collectRelevantChanges(JsonObject payload) {
+  protected Future<List<Change<RequestPolicy>>> collectRelevantChanges(JsonObject payload) {
     log.debug("collectRelevantChanges:: payload: {}", payload);
 
     JsonObject oldObject = payload.getJsonObject("old");
@@ -38,7 +39,7 @@ protected List<Change<RequestPolicy>> collectRelevantChanges(JsonObject payload)
     changes.add(new Change<>(requestPolicy -> removeServicePointFromRequestPolicy(requestPolicy,
       deletedServicePointId)));
 
-    return changes;
+    return Future.succeededFuture(changes);
   }
 
   @Override

src/main/java/org/folio/service/event/handler/processor/ServicePointUpdateProcessorForRequest.java

@@ -6,6 +6,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import io.vertx.core.Future;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.folio.persist.RequestRepository;
@@ -24,7 +25,7 @@ public ServicePointUpdateProcessorForRequest(RequestRepository requestRepository
   }
 
   @Override
-  protected List<Change<Request>> collectRelevantChanges(JsonObject payload) {
+  protected Future<List<Change<Request>>> collectRelevantChanges(JsonObject payload) {
     JsonObject oldObject = payload.getJsonObject("old");
     JsonObject newObject = payload.getJsonObject("new");
 
@@ -40,7 +41,7 @@ protected List<Change<Request>> collectRelevantChanges(JsonObject payload) {
         .setPickupServicePointName(newServicePointName)));
     }
 
-    return changes;
+    return Future.succeededFuture(changes);
   }
 
   @Override