@Test(expected = IllegalArgumentException.class) public void givenWorkerImplementWhenEmptyActionsInStepThenThrowsIllegalArgumentException() throws IllegalArgumentException, HandlerNotFoundException, ProcessingException, ContentAddressableStorageServerException { workerImpl = WorkerFactory.getInstance(pluginLoader).create(); workerImpl.run( WorkerParametersFactory.newWorkerParameters().setUrlWorkspace("http: .setUrlMetadata("http: .setObjectNameList(Lists.newArrayList("objectName.json")).setCurrentStep("currentStep").setContainerName("containerName"), new Step()); }
@Override public ItemStatus run(WorkerParameters workParams, Step step) throws IllegalArgumentException, ProcessingException { ParametersChecker.checkNullOrEmptyParameters(workParams); if (step == null) { throw new IllegalArgumentException(STEP_NULL); } if (step.getActions() == null || step.getActions().isEmpty()) { throw new IllegalArgumentException(EMPTY_LIST); } final ItemStatus responses = new ItemStatus(step.getStepName()); workerId = GUIDFactory.newGUID().toString(); try (final HandlerIO handlerIO = new HandlerIOImpl(workParams.getContainerName(), workerId, workParams.getObjectNameList()); LogbookLifeCyclesClient logbookLfcClient = LogbookLifeCyclesClientFactory.getInstance().getClient()) { LifecycleFromWorker lifecycleFromWorker = new LifecycleFromWorker(logbookLfcClient); for (final Action action : step.getActions()) { Stopwatch stopwatch = Stopwatch.createStarted(); handlerIO.reset(); ActionDefinition actionDefinition = action.getActionDefinition(); if (actionDefinition.getIn() != null) { handlerIO.addInIOParameters(actionDefinition.getIn()); } if (actionDefinition.getOut() != null) { handlerIO.addOutIOParameters(actionDefinition.getOut()); } String handlerName = actionDefinition.getActionKey(); List<ItemStatus> pluginResponse; ItemStatus aggregateItemStatus = new ItemStatus(); if (pluginLoader.contains(handlerName)) { try (ActionHandler actionPlugin = pluginLoader.newInstance(handlerName)) { LOGGER.debug("START plugin ", actionDefinition.getActionKey(), step.getStepName()); if (action.getActionDefinition().lifecycleEnabled()) { pluginResponse = actionPlugin.executeList(workParams, handlerIO); lifecycleFromWorker .generateLifeCycle(pluginResponse, workParams, action, step.getDistribution().getType(), aggregateItemStatus); aggregateItemStatus.setItemId(handlerName); aggregateItemStatus = getActionResponse(handlerName, aggregateItemStatus); } else { pluginResponse = actionPlugin.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } aggregateItemStatus.setItemId(handlerName); } responses.setItemsStatus(aggregateItemStatus); } } else { final ActionHandler actionHandler = getActionHandler(handlerName); if (actionHandler == null) { throw new HandlerNotFoundException(actionDefinition.getActionKey() + HANDLER_NOT_FOUND); } LOGGER.debug("START handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); pluginResponse = actionHandler.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } responses.setItemsStatus(aggregateItemStatus); } if (FLUSH_LFC.equals(action.getActionDefinition().getLifecycleState())) { lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } LOGGER.debug("STOP handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); PERFORMANCE_LOGGER.log(step.getStepName(), actionDefinition.getActionKey(), elapsed); if (responses.shallStop(ProcessBehavior.BLOCKING.equals(actionDefinition.getBehavior()))) { break; } } lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } catch (Exception e) { throw new ProcessingException(e); } LOGGER.debug("step name :" + step.getStepName()); return responses; }
WorkerImpl implements Worker { @Override public ItemStatus run(WorkerParameters workParams, Step step) throws IllegalArgumentException, ProcessingException { ParametersChecker.checkNullOrEmptyParameters(workParams); if (step == null) { throw new IllegalArgumentException(STEP_NULL); } if (step.getActions() == null || step.getActions().isEmpty()) { throw new IllegalArgumentException(EMPTY_LIST); } final ItemStatus responses = new ItemStatus(step.getStepName()); workerId = GUIDFactory.newGUID().toString(); try (final HandlerIO handlerIO = new HandlerIOImpl(workParams.getContainerName(), workerId, workParams.getObjectNameList()); LogbookLifeCyclesClient logbookLfcClient = LogbookLifeCyclesClientFactory.getInstance().getClient()) { LifecycleFromWorker lifecycleFromWorker = new LifecycleFromWorker(logbookLfcClient); for (final Action action : step.getActions()) { Stopwatch stopwatch = Stopwatch.createStarted(); handlerIO.reset(); ActionDefinition actionDefinition = action.getActionDefinition(); if (actionDefinition.getIn() != null) { handlerIO.addInIOParameters(actionDefinition.getIn()); } if (actionDefinition.getOut() != null) { handlerIO.addOutIOParameters(actionDefinition.getOut()); } String handlerName = actionDefinition.getActionKey(); List<ItemStatus> pluginResponse; ItemStatus aggregateItemStatus = new ItemStatus(); if (pluginLoader.contains(handlerName)) { try (ActionHandler actionPlugin = pluginLoader.newInstance(handlerName)) { LOGGER.debug("START plugin ", actionDefinition.getActionKey(), step.getStepName()); if (action.getActionDefinition().lifecycleEnabled()) { pluginResponse = actionPlugin.executeList(workParams, handlerIO); lifecycleFromWorker .generateLifeCycle(pluginResponse, workParams, action, step.getDistribution().getType(), aggregateItemStatus); aggregateItemStatus.setItemId(handlerName); aggregateItemStatus = getActionResponse(handlerName, aggregateItemStatus); } else { pluginResponse = actionPlugin.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } aggregateItemStatus.setItemId(handlerName); } responses.setItemsStatus(aggregateItemStatus); } } else { final ActionHandler actionHandler = getActionHandler(handlerName); if (actionHandler == null) { throw new HandlerNotFoundException(actionDefinition.getActionKey() + HANDLER_NOT_FOUND); } LOGGER.debug("START handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); pluginResponse = actionHandler.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } responses.setItemsStatus(aggregateItemStatus); } if (FLUSH_LFC.equals(action.getActionDefinition().getLifecycleState())) { lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } LOGGER.debug("STOP handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); PERFORMANCE_LOGGER.log(step.getStepName(), actionDefinition.getActionKey(), elapsed); if (responses.shallStop(ProcessBehavior.BLOCKING.equals(actionDefinition.getBehavior()))) { break; } } lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } catch (Exception e) { throw new ProcessingException(e); } LOGGER.debug("step name :" + step.getStepName()); return responses; } }
WorkerImpl implements Worker { @Override public ItemStatus run(WorkerParameters workParams, Step step) throws IllegalArgumentException, ProcessingException { ParametersChecker.checkNullOrEmptyParameters(workParams); if (step == null) { throw new IllegalArgumentException(STEP_NULL); } if (step.getActions() == null || step.getActions().isEmpty()) { throw new IllegalArgumentException(EMPTY_LIST); } final ItemStatus responses = new ItemStatus(step.getStepName()); workerId = GUIDFactory.newGUID().toString(); try (final HandlerIO handlerIO = new HandlerIOImpl(workParams.getContainerName(), workerId, workParams.getObjectNameList()); LogbookLifeCyclesClient logbookLfcClient = LogbookLifeCyclesClientFactory.getInstance().getClient()) { LifecycleFromWorker lifecycleFromWorker = new LifecycleFromWorker(logbookLfcClient); for (final Action action : step.getActions()) { Stopwatch stopwatch = Stopwatch.createStarted(); handlerIO.reset(); ActionDefinition actionDefinition = action.getActionDefinition(); if (actionDefinition.getIn() != null) { handlerIO.addInIOParameters(actionDefinition.getIn()); } if (actionDefinition.getOut() != null) { handlerIO.addOutIOParameters(actionDefinition.getOut()); } String handlerName = actionDefinition.getActionKey(); List<ItemStatus> pluginResponse; ItemStatus aggregateItemStatus = new ItemStatus(); if (pluginLoader.contains(handlerName)) { try (ActionHandler actionPlugin = pluginLoader.newInstance(handlerName)) { LOGGER.debug("START plugin ", actionDefinition.getActionKey(), step.getStepName()); if (action.getActionDefinition().lifecycleEnabled()) { pluginResponse = actionPlugin.executeList(workParams, handlerIO); lifecycleFromWorker .generateLifeCycle(pluginResponse, workParams, action, step.getDistribution().getType(), aggregateItemStatus); aggregateItemStatus.setItemId(handlerName); aggregateItemStatus = getActionResponse(handlerName, aggregateItemStatus); } else { pluginResponse = actionPlugin.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } aggregateItemStatus.setItemId(handlerName); } responses.setItemsStatus(aggregateItemStatus); } } else { final ActionHandler actionHandler = getActionHandler(handlerName); if (actionHandler == null) { throw new HandlerNotFoundException(actionDefinition.getActionKey() + HANDLER_NOT_FOUND); } LOGGER.debug("START handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); pluginResponse = actionHandler.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } responses.setItemsStatus(aggregateItemStatus); } if (FLUSH_LFC.equals(action.getActionDefinition().getLifecycleState())) { lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } LOGGER.debug("STOP handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); PERFORMANCE_LOGGER.log(step.getStepName(), actionDefinition.getActionKey(), elapsed); if (responses.shallStop(ProcessBehavior.BLOCKING.equals(actionDefinition.getBehavior()))) { break; } } lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } catch (Exception e) { throw new ProcessingException(e); } LOGGER.debug("step name :" + step.getStepName()); return responses; } WorkerImpl(PluginLoader pluginLoader); }
WorkerImpl implements Worker { @Override public ItemStatus run(WorkerParameters workParams, Step step) throws IllegalArgumentException, ProcessingException { ParametersChecker.checkNullOrEmptyParameters(workParams); if (step == null) { throw new IllegalArgumentException(STEP_NULL); } if (step.getActions() == null || step.getActions().isEmpty()) { throw new IllegalArgumentException(EMPTY_LIST); } final ItemStatus responses = new ItemStatus(step.getStepName()); workerId = GUIDFactory.newGUID().toString(); try (final HandlerIO handlerIO = new HandlerIOImpl(workParams.getContainerName(), workerId, workParams.getObjectNameList()); LogbookLifeCyclesClient logbookLfcClient = LogbookLifeCyclesClientFactory.getInstance().getClient()) { LifecycleFromWorker lifecycleFromWorker = new LifecycleFromWorker(logbookLfcClient); for (final Action action : step.getActions()) { Stopwatch stopwatch = Stopwatch.createStarted(); handlerIO.reset(); ActionDefinition actionDefinition = action.getActionDefinition(); if (actionDefinition.getIn() != null) { handlerIO.addInIOParameters(actionDefinition.getIn()); } if (actionDefinition.getOut() != null) { handlerIO.addOutIOParameters(actionDefinition.getOut()); } String handlerName = actionDefinition.getActionKey(); List<ItemStatus> pluginResponse; ItemStatus aggregateItemStatus = new ItemStatus(); if (pluginLoader.contains(handlerName)) { try (ActionHandler actionPlugin = pluginLoader.newInstance(handlerName)) { LOGGER.debug("START plugin ", actionDefinition.getActionKey(), step.getStepName()); if (action.getActionDefinition().lifecycleEnabled()) { pluginResponse = actionPlugin.executeList(workParams, handlerIO); lifecycleFromWorker .generateLifeCycle(pluginResponse, workParams, action, step.getDistribution().getType(), aggregateItemStatus); aggregateItemStatus.setItemId(handlerName); aggregateItemStatus = getActionResponse(handlerName, aggregateItemStatus); } else { pluginResponse = actionPlugin.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } aggregateItemStatus.setItemId(handlerName); } responses.setItemsStatus(aggregateItemStatus); } } else { final ActionHandler actionHandler = getActionHandler(handlerName); if (actionHandler == null) { throw new HandlerNotFoundException(actionDefinition.getActionKey() + HANDLER_NOT_FOUND); } LOGGER.debug("START handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); pluginResponse = actionHandler.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } responses.setItemsStatus(aggregateItemStatus); } if (FLUSH_LFC.equals(action.getActionDefinition().getLifecycleState())) { lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } LOGGER.debug("STOP handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); PERFORMANCE_LOGGER.log(step.getStepName(), actionDefinition.getActionKey(), elapsed); if (responses.shallStop(ProcessBehavior.BLOCKING.equals(actionDefinition.getBehavior()))) { break; } } lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } catch (Exception e) { throw new ProcessingException(e); } LOGGER.debug("step name :" + step.getStepName()); return responses; } WorkerImpl(PluginLoader pluginLoader); @Override @VisibleForTesting WorkerImpl addActionHandler(String actionName, ActionHandler actionHandler); @Override ItemStatus run(WorkerParameters workParams, Step step); @Override String getWorkerId(); @Override void close(); }
WorkerImpl implements Worker { @Override public ItemStatus run(WorkerParameters workParams, Step step) throws IllegalArgumentException, ProcessingException { ParametersChecker.checkNullOrEmptyParameters(workParams); if (step == null) { throw new IllegalArgumentException(STEP_NULL); } if (step.getActions() == null || step.getActions().isEmpty()) { throw new IllegalArgumentException(EMPTY_LIST); } final ItemStatus responses = new ItemStatus(step.getStepName()); workerId = GUIDFactory.newGUID().toString(); try (final HandlerIO handlerIO = new HandlerIOImpl(workParams.getContainerName(), workerId, workParams.getObjectNameList()); LogbookLifeCyclesClient logbookLfcClient = LogbookLifeCyclesClientFactory.getInstance().getClient()) { LifecycleFromWorker lifecycleFromWorker = new LifecycleFromWorker(logbookLfcClient); for (final Action action : step.getActions()) { Stopwatch stopwatch = Stopwatch.createStarted(); handlerIO.reset(); ActionDefinition actionDefinition = action.getActionDefinition(); if (actionDefinition.getIn() != null) { handlerIO.addInIOParameters(actionDefinition.getIn()); } if (actionDefinition.getOut() != null) { handlerIO.addOutIOParameters(actionDefinition.getOut()); } String handlerName = actionDefinition.getActionKey(); List<ItemStatus> pluginResponse; ItemStatus aggregateItemStatus = new ItemStatus(); if (pluginLoader.contains(handlerName)) { try (ActionHandler actionPlugin = pluginLoader.newInstance(handlerName)) { LOGGER.debug("START plugin ", actionDefinition.getActionKey(), step.getStepName()); if (action.getActionDefinition().lifecycleEnabled()) { pluginResponse = actionPlugin.executeList(workParams, handlerIO); lifecycleFromWorker .generateLifeCycle(pluginResponse, workParams, action, step.getDistribution().getType(), aggregateItemStatus); aggregateItemStatus.setItemId(handlerName); aggregateItemStatus = getActionResponse(handlerName, aggregateItemStatus); } else { pluginResponse = actionPlugin.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } aggregateItemStatus.setItemId(handlerName); } responses.setItemsStatus(aggregateItemStatus); } } else { final ActionHandler actionHandler = getActionHandler(handlerName); if (actionHandler == null) { throw new HandlerNotFoundException(actionDefinition.getActionKey() + HANDLER_NOT_FOUND); } LOGGER.debug("START handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); pluginResponse = actionHandler.executeList(workParams, handlerIO); for (ItemStatus itemStatus : pluginResponse) { aggregateItemStatus.setItemId(itemStatus.getItemId()); aggregateItemStatus.setItemsStatus(itemStatus); } responses.setItemsStatus(aggregateItemStatus); } if (FLUSH_LFC.equals(action.getActionDefinition().getLifecycleState())) { lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } LOGGER.debug("STOP handler {} in step {}", actionDefinition.getActionKey(), step.getStepName()); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); PERFORMANCE_LOGGER.log(step.getStepName(), actionDefinition.getActionKey(), elapsed); if (responses.shallStop(ProcessBehavior.BLOCKING.equals(actionDefinition.getBehavior()))) { break; } } lifecycleFromWorker.saveLifeCycles(step.getDistribution().getType()); } catch (Exception e) { throw new ProcessingException(e); } LOGGER.debug("step name :" + step.getStepName()); return responses; } WorkerImpl(PluginLoader pluginLoader); @Override @VisibleForTesting WorkerImpl addActionHandler(String actionName, ActionHandler actionHandler); @Override ItemStatus run(WorkerParameters workParams, Step step); @Override String getWorkerId(); @Override void close();  }