@Test public void givenWorkspaceExistAndPdoWhenExecuteThenReturnResponseOK() throws Exception { final WorkerParameters paramsObjectGroups = WorkerParametersFactory.newWorkerParameters().setWorkerGUID(GUIDFactory .newGUID().getId()).setContainerName(CONTAINER_NAME).setUrlMetadata("http: .setUrlWorkspace("http: .setObjectNameList(Lists.newArrayList(OBJECT_GROUP_GUID_2 + ".json")) .setObjectName(OBJECT_GROUP_GUID_2 + ".json").setCurrentStep("Store ObjectGroup"); when(workspaceClient.getObject(CONTAINER_NAME, "ObjectGroup/" + OBJECT_GROUP_GUID_2 + ".json")) .thenReturn(Response.status(Status.OK).entity(objectGroup2).build()); doReturn(getStorageResult("aeaaaaaaaaakwtamaaxakak32oqku2iaaaaq", "942bb63cc16bf5ca3ba7fabf40ce9be19c3185a36cd87ad17c63d6fad1aa29d4312d73f2d6a1ba1266c3a71fc4119dd476d2d776cf2ad2acd7a9a3dfa1f80dc7")) .when(storageClient).bulkStoreFilesFromWorkspace(any(), any()); plugin = new StoreObjectGroupActionPlugin(storageClientFactory); saveWorkspacePutObject(); final List<ItemStatus> response = plugin.executeList(paramsObjectGroups, action); assertEquals(StatusCode.OK, response.get(0).getGlobalStatus()); ArgumentCaptor<String> fileNameArgumentCaptor = ArgumentCaptor.forClass(String.class); verify(workspaceClient).putObject(anyString(), ArgumentMatchers.startsWith("ObjectGroup/"), any(InputStream.class)); verify(workspaceClient, atLeastOnce()).putObject(anyString(), fileNameArgumentCaptor.capture(), any(InputStream.class)); String gotFilename = fileNameArgumentCaptor.getAllValues().stream() .filter(str -> str.startsWith("ObjectGroup/")) .findFirst().get(); JsonNode gotJson = getSavedWorkspaceObject(gotFilename); assertNotNull(gotJson); final JsonPointer storagePointer = JsonPointer.compile("/_qualifiers/1/versions/0/_storage"); final JsonPointer storageStrategyPointer = JsonPointer.compile("/_qualifiers/1/versions/0/_storage/strategyId"); final JsonPointer storageOfferIdPointer = JsonPointer.compile("/_qualifiers/1/versions/0/_storage/offerIds"); assertTrue(gotJson.at(storagePointer).isObject()); assertEquals("default", gotJson.at(storageStrategyPointer).asText()); assertTrue(gotJson.at(storageOfferIdPointer).isMissingNode()); }
@Override public List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO) { checkMandatoryParameters(params); final List<ItemStatus> itemStatusList = new ArrayList<>(); final List<Map<String, ItemStatus>> itemStatusByObjectList = new ArrayList<>(); try { checkMandatoryIOParameter(handlerIO); Set<String> strategies = new LinkedHashSet<>(); Map<String, String> strategiesByObjectId = new HashMap<>(); ListValuedMap<String, String> workspaceObjectURIsByStrategies = new ArrayListValuedHashMap<>(); ListValuedMap<String, String> objectNamesByStrategies = new ArrayListValuedHashMap<>(); List<MapOfObjects> mapOfObjectsList = new ArrayList<>(); for (String objectName : params.getObjectNameList()) { MapOfObjects mapOfObjects = getMapOfObjectsIdsAndUris(params.getContainerName(), objectName, handlerIO); Map<String, ItemStatus> itemStatusByObject = new HashMap<>(); for (final Map.Entry<String, String> objectGuid : mapOfObjects.getBinaryObjectsToStore() .entrySet()) { itemStatusByObject.put(objectGuid.getKey(), new ItemStatus(STORING_OBJECT_TASK_ID)); String strategyId = mapOfObjects.getObjectStorageInfos().get(objectGuid.getKey()).get("strategyId").asText(); strategies.add(strategyId); workspaceObjectURIsByStrategies.put(strategyId, SIP + objectGuid.getValue()); objectNamesByStrategies.put(strategyId, objectGuid.getKey()); strategiesByObjectId.put(objectGuid.getKey(), strategyId); } itemStatusByObjectList.add(itemStatusByObject); itemStatusList.add(new ItemStatus(STORING_OBJECT_TASK_ID)); mapOfObjectsList.add(mapOfObjects); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Pre OG: {}", JsonHandler.prettyPrint(mapOfObjects.getJsonOG())); } } if (objectNamesByStrategies.values().isEmpty()) { return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.OK))); } Map<String, BulkObjectStoreResponse> resultByStrategy = new LinkedHashMap<>(); for (String strategy : strategies) { List<String> workspaceObjectURIs = workspaceObjectURIsByStrategies.get(strategy); List<String> objectNames = objectNamesByStrategies.get(strategy); BulkObjectStoreRequest bulkObjectStoreRequest = new BulkObjectStoreRequest(params.getContainerName(), workspaceObjectURIs, DataCategory.OBJECT, objectNames); BulkObjectStoreResponse result = storeObjects(strategy, bulkObjectStoreRequest); resultByStrategy.put(strategy, result); } updateSubTasksAndTasksFromStorageInfos(resultByStrategy, itemStatusByObjectList, itemStatusList); storeStorageInfos(mapOfObjectsList, resultByStrategy, strategiesByObjectId); for (int i = 0; i < mapOfObjectsList.size(); i++) { handlerIO.transferJsonToWorkspace(IngestWorkflowConstants.OBJECT_GROUP_FOLDER, params.getObjectNameList().get(i), mapOfObjectsList.get(i).getJsonOG(), false, false); } } catch (final ProcessingException e) { LOGGER.error(params.getObjectName(), e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } catch (StorageClientException e) { LOGGER.error(e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } return getFinalResult(itemStatusList); }
StoreObjectGroupActionPlugin extends StoreObjectActionHandler { @Override public List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO) { checkMandatoryParameters(params); final List<ItemStatus> itemStatusList = new ArrayList<>(); final List<Map<String, ItemStatus>> itemStatusByObjectList = new ArrayList<>(); try { checkMandatoryIOParameter(handlerIO); Set<String> strategies = new LinkedHashSet<>(); Map<String, String> strategiesByObjectId = new HashMap<>(); ListValuedMap<String, String> workspaceObjectURIsByStrategies = new ArrayListValuedHashMap<>(); ListValuedMap<String, String> objectNamesByStrategies = new ArrayListValuedHashMap<>(); List<MapOfObjects> mapOfObjectsList = new ArrayList<>(); for (String objectName : params.getObjectNameList()) { MapOfObjects mapOfObjects = getMapOfObjectsIdsAndUris(params.getContainerName(), objectName, handlerIO); Map<String, ItemStatus> itemStatusByObject = new HashMap<>(); for (final Map.Entry<String, String> objectGuid : mapOfObjects.getBinaryObjectsToStore() .entrySet()) { itemStatusByObject.put(objectGuid.getKey(), new ItemStatus(STORING_OBJECT_TASK_ID)); String strategyId = mapOfObjects.getObjectStorageInfos().get(objectGuid.getKey()).get("strategyId").asText(); strategies.add(strategyId); workspaceObjectURIsByStrategies.put(strategyId, SIP + objectGuid.getValue()); objectNamesByStrategies.put(strategyId, objectGuid.getKey()); strategiesByObjectId.put(objectGuid.getKey(), strategyId); } itemStatusByObjectList.add(itemStatusByObject); itemStatusList.add(new ItemStatus(STORING_OBJECT_TASK_ID)); mapOfObjectsList.add(mapOfObjects); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Pre OG: {}", JsonHandler.prettyPrint(mapOfObjects.getJsonOG())); } } if (objectNamesByStrategies.values().isEmpty()) { return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.OK))); } Map<String, BulkObjectStoreResponse> resultByStrategy = new LinkedHashMap<>(); for (String strategy : strategies) { List<String> workspaceObjectURIs = workspaceObjectURIsByStrategies.get(strategy); List<String> objectNames = objectNamesByStrategies.get(strategy); BulkObjectStoreRequest bulkObjectStoreRequest = new BulkObjectStoreRequest(params.getContainerName(), workspaceObjectURIs, DataCategory.OBJECT, objectNames); BulkObjectStoreResponse result = storeObjects(strategy, bulkObjectStoreRequest); resultByStrategy.put(strategy, result); } updateSubTasksAndTasksFromStorageInfos(resultByStrategy, itemStatusByObjectList, itemStatusList); storeStorageInfos(mapOfObjectsList, resultByStrategy, strategiesByObjectId); for (int i = 0; i < mapOfObjectsList.size(); i++) { handlerIO.transferJsonToWorkspace(IngestWorkflowConstants.OBJECT_GROUP_FOLDER, params.getObjectNameList().get(i), mapOfObjectsList.get(i).getJsonOG(), false, false); } } catch (final ProcessingException e) { LOGGER.error(params.getObjectName(), e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } catch (StorageClientException e) { LOGGER.error(e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } return getFinalResult(itemStatusList); } }
StoreObjectGroupActionPlugin extends StoreObjectActionHandler { @Override public List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO) { checkMandatoryParameters(params); final List<ItemStatus> itemStatusList = new ArrayList<>(); final List<Map<String, ItemStatus>> itemStatusByObjectList = new ArrayList<>(); try { checkMandatoryIOParameter(handlerIO); Set<String> strategies = new LinkedHashSet<>(); Map<String, String> strategiesByObjectId = new HashMap<>(); ListValuedMap<String, String> workspaceObjectURIsByStrategies = new ArrayListValuedHashMap<>(); ListValuedMap<String, String> objectNamesByStrategies = new ArrayListValuedHashMap<>(); List<MapOfObjects> mapOfObjectsList = new ArrayList<>(); for (String objectName : params.getObjectNameList()) { MapOfObjects mapOfObjects = getMapOfObjectsIdsAndUris(params.getContainerName(), objectName, handlerIO); Map<String, ItemStatus> itemStatusByObject = new HashMap<>(); for (final Map.Entry<String, String> objectGuid : mapOfObjects.getBinaryObjectsToStore() .entrySet()) { itemStatusByObject.put(objectGuid.getKey(), new ItemStatus(STORING_OBJECT_TASK_ID)); String strategyId = mapOfObjects.getObjectStorageInfos().get(objectGuid.getKey()).get("strategyId").asText(); strategies.add(strategyId); workspaceObjectURIsByStrategies.put(strategyId, SIP + objectGuid.getValue()); objectNamesByStrategies.put(strategyId, objectGuid.getKey()); strategiesByObjectId.put(objectGuid.getKey(), strategyId); } itemStatusByObjectList.add(itemStatusByObject); itemStatusList.add(new ItemStatus(STORING_OBJECT_TASK_ID)); mapOfObjectsList.add(mapOfObjects); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Pre OG: {}", JsonHandler.prettyPrint(mapOfObjects.getJsonOG())); } } if (objectNamesByStrategies.values().isEmpty()) { return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.OK))); } Map<String, BulkObjectStoreResponse> resultByStrategy = new LinkedHashMap<>(); for (String strategy : strategies) { List<String> workspaceObjectURIs = workspaceObjectURIsByStrategies.get(strategy); List<String> objectNames = objectNamesByStrategies.get(strategy); BulkObjectStoreRequest bulkObjectStoreRequest = new BulkObjectStoreRequest(params.getContainerName(), workspaceObjectURIs, DataCategory.OBJECT, objectNames); BulkObjectStoreResponse result = storeObjects(strategy, bulkObjectStoreRequest); resultByStrategy.put(strategy, result); } updateSubTasksAndTasksFromStorageInfos(resultByStrategy, itemStatusByObjectList, itemStatusList); storeStorageInfos(mapOfObjectsList, resultByStrategy, strategiesByObjectId); for (int i = 0; i < mapOfObjectsList.size(); i++) { handlerIO.transferJsonToWorkspace(IngestWorkflowConstants.OBJECT_GROUP_FOLDER, params.getObjectNameList().get(i), mapOfObjectsList.get(i).getJsonOG(), false, false); } } catch (final ProcessingException e) { LOGGER.error(params.getObjectName(), e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } catch (StorageClientException e) { LOGGER.error(e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } return getFinalResult(itemStatusList); } StoreObjectGroupActionPlugin();  StoreObjectGroupActionPlugin(StorageClientFactory storageClientFactory); }
StoreObjectGroupActionPlugin extends StoreObjectActionHandler { @Override public List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO) { checkMandatoryParameters(params); final List<ItemStatus> itemStatusList = new ArrayList<>(); final List<Map<String, ItemStatus>> itemStatusByObjectList = new ArrayList<>(); try { checkMandatoryIOParameter(handlerIO); Set<String> strategies = new LinkedHashSet<>(); Map<String, String> strategiesByObjectId = new HashMap<>(); ListValuedMap<String, String> workspaceObjectURIsByStrategies = new ArrayListValuedHashMap<>(); ListValuedMap<String, String> objectNamesByStrategies = new ArrayListValuedHashMap<>(); List<MapOfObjects> mapOfObjectsList = new ArrayList<>(); for (String objectName : params.getObjectNameList()) { MapOfObjects mapOfObjects = getMapOfObjectsIdsAndUris(params.getContainerName(), objectName, handlerIO); Map<String, ItemStatus> itemStatusByObject = new HashMap<>(); for (final Map.Entry<String, String> objectGuid : mapOfObjects.getBinaryObjectsToStore() .entrySet()) { itemStatusByObject.put(objectGuid.getKey(), new ItemStatus(STORING_OBJECT_TASK_ID)); String strategyId = mapOfObjects.getObjectStorageInfos().get(objectGuid.getKey()).get("strategyId").asText(); strategies.add(strategyId); workspaceObjectURIsByStrategies.put(strategyId, SIP + objectGuid.getValue()); objectNamesByStrategies.put(strategyId, objectGuid.getKey()); strategiesByObjectId.put(objectGuid.getKey(), strategyId); } itemStatusByObjectList.add(itemStatusByObject); itemStatusList.add(new ItemStatus(STORING_OBJECT_TASK_ID)); mapOfObjectsList.add(mapOfObjects); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Pre OG: {}", JsonHandler.prettyPrint(mapOfObjects.getJsonOG())); } } if (objectNamesByStrategies.values().isEmpty()) { return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.OK))); } Map<String, BulkObjectStoreResponse> resultByStrategy = new LinkedHashMap<>(); for (String strategy : strategies) { List<String> workspaceObjectURIs = workspaceObjectURIsByStrategies.get(strategy); List<String> objectNames = objectNamesByStrategies.get(strategy); BulkObjectStoreRequest bulkObjectStoreRequest = new BulkObjectStoreRequest(params.getContainerName(), workspaceObjectURIs, DataCategory.OBJECT, objectNames); BulkObjectStoreResponse result = storeObjects(strategy, bulkObjectStoreRequest); resultByStrategy.put(strategy, result); } updateSubTasksAndTasksFromStorageInfos(resultByStrategy, itemStatusByObjectList, itemStatusList); storeStorageInfos(mapOfObjectsList, resultByStrategy, strategiesByObjectId); for (int i = 0; i < mapOfObjectsList.size(); i++) { handlerIO.transferJsonToWorkspace(IngestWorkflowConstants.OBJECT_GROUP_FOLDER, params.getObjectNameList().get(i), mapOfObjectsList.get(i).getJsonOG(), false, false); } } catch (final ProcessingException e) { LOGGER.error(params.getObjectName(), e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } catch (StorageClientException e) { LOGGER.error(e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } return getFinalResult(itemStatusList); } StoreObjectGroupActionPlugin();  StoreObjectGroupActionPlugin(StorageClientFactory storageClientFactory); @Override List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO); @Override ItemStatus execute(WorkerParameters param, HandlerIO handler); @Override void checkMandatoryIOParameter(HandlerIO handler); }
StoreObjectGroupActionPlugin extends StoreObjectActionHandler { @Override public List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO) { checkMandatoryParameters(params); final List<ItemStatus> itemStatusList = new ArrayList<>(); final List<Map<String, ItemStatus>> itemStatusByObjectList = new ArrayList<>(); try { checkMandatoryIOParameter(handlerIO); Set<String> strategies = new LinkedHashSet<>(); Map<String, String> strategiesByObjectId = new HashMap<>(); ListValuedMap<String, String> workspaceObjectURIsByStrategies = new ArrayListValuedHashMap<>(); ListValuedMap<String, String> objectNamesByStrategies = new ArrayListValuedHashMap<>(); List<MapOfObjects> mapOfObjectsList = new ArrayList<>(); for (String objectName : params.getObjectNameList()) { MapOfObjects mapOfObjects = getMapOfObjectsIdsAndUris(params.getContainerName(), objectName, handlerIO); Map<String, ItemStatus> itemStatusByObject = new HashMap<>(); for (final Map.Entry<String, String> objectGuid : mapOfObjects.getBinaryObjectsToStore() .entrySet()) { itemStatusByObject.put(objectGuid.getKey(), new ItemStatus(STORING_OBJECT_TASK_ID)); String strategyId = mapOfObjects.getObjectStorageInfos().get(objectGuid.getKey()).get("strategyId").asText(); strategies.add(strategyId); workspaceObjectURIsByStrategies.put(strategyId, SIP + objectGuid.getValue()); objectNamesByStrategies.put(strategyId, objectGuid.getKey()); strategiesByObjectId.put(objectGuid.getKey(), strategyId); } itemStatusByObjectList.add(itemStatusByObject); itemStatusList.add(new ItemStatus(STORING_OBJECT_TASK_ID)); mapOfObjectsList.add(mapOfObjects); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Pre OG: {}", JsonHandler.prettyPrint(mapOfObjects.getJsonOG())); } } if (objectNamesByStrategies.values().isEmpty()) { return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.OK))); } Map<String, BulkObjectStoreResponse> resultByStrategy = new LinkedHashMap<>(); for (String strategy : strategies) { List<String> workspaceObjectURIs = workspaceObjectURIsByStrategies.get(strategy); List<String> objectNames = objectNamesByStrategies.get(strategy); BulkObjectStoreRequest bulkObjectStoreRequest = new BulkObjectStoreRequest(params.getContainerName(), workspaceObjectURIs, DataCategory.OBJECT, objectNames); BulkObjectStoreResponse result = storeObjects(strategy, bulkObjectStoreRequest); resultByStrategy.put(strategy, result); } updateSubTasksAndTasksFromStorageInfos(resultByStrategy, itemStatusByObjectList, itemStatusList); storeStorageInfos(mapOfObjectsList, resultByStrategy, strategiesByObjectId); for (int i = 0; i < mapOfObjectsList.size(); i++) { handlerIO.transferJsonToWorkspace(IngestWorkflowConstants.OBJECT_GROUP_FOLDER, params.getObjectNameList().get(i), mapOfObjectsList.get(i).getJsonOG(), false, false); } } catch (final ProcessingException e) { LOGGER.error(params.getObjectName(), e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } catch (StorageClientException e) { LOGGER.error(e); return Arrays.asList(new ItemStatus(STORING_OBJECT_TASK_ID) .setItemsStatus(STORING_OBJECT_TASK_ID, new ItemStatus().increment(StatusCode.FATAL))); } return getFinalResult(itemStatusList); } StoreObjectGroupActionPlugin();  StoreObjectGroupActionPlugin(StorageClientFactory storageClientFactory); @Override List<ItemStatus> executeList(WorkerParameters params, HandlerIO handlerIO); @Override ItemStatus execute(WorkerParameters param, HandlerIO handler); @Override void checkMandatoryIOParameter(HandlerIO handler); static final String STORING_OBJECT_TASK_ID; }