DatabaseFDFlowManager extends BaseService implements FDFlowManager { @Override public FDFlowEvent addFlowEvent(final FDFlowEvent flowEvent, final VersionedProcessGroup flowContent, final NiFiUser user) { if (flowEvent == null) { throw new IllegalArgumentException("Flow event cannot be null or blank"); } if (StringUtils.isBlank(flowEvent.getFlowIdentifier())) { throw new IllegalArgumentException("Flow identifier of flow event cannot be null or blank"); } if (flowContent == null) { throw new IllegalArgumentException("Flow content cannot be null or blank"); } if (StringUtils.isBlank(flowContent.getIdentifier()) || StringUtils.isBlank(flowContent.getName())) { throw new IllegalArgumentException("Flow content must contain a root process group with an identifier and name"); } if (user == null || StringUtils.isBlank(user.getIdentity())) { throw new IllegalArgumentException("A user identity must be provided"); } final Optional<FDFlowEntity> flowEntity = flowRepository.findById(flowEvent.getFlowIdentifier()); if (!flowEntity.isPresent()) { throw new ResourceNotFoundException("No flow exists with the provided flow id"); } final String previousRootGroupId = flowEntity.get().getRootProcessGroupId(); final String incomingRootGroupId = flowContent.getIdentifier(); if (!previousRootGroupId.equals(incomingRootGroupId)) { throw new IllegalArgumentException("Cannot change the id of the root process group"); } final Optional<FDFlowEventEntity> latestFlowEvent = flowEventRepository.findFirstByFlowOrderByFlowRevisionDesc(flowEntity.get()); if (!latestFlowEvent.isPresent()) { throw new IllegalStateException("Could not find any flow content for the given flow"); } final BigInteger currRevision = latestFlowEvent.get().getFlowRevision(); final BigInteger nextRevision = currRevision.add(BigInteger.ONE); flowEvent.setIdentifier(UUID.randomUUID().toString()); flowEvent.setCreated(System.currentTimeMillis()); flowEvent.setUpdated(System.currentTimeMillis()); flowEvent.setFlowRevision(nextRevision); flowEvent.setFlowFormat(CURRENT_FLOW_FORMAT); flowEvent.setUserIdentity(user.getIdentity()); if (flowEvent.getEventType() != FDFlowEventType.FLOW_PUBLISHED) { flowEvent.setRegistryUrl(latestFlowEvent.get().getRegistryUrl()); flowEvent.setRegistryBucketId(latestFlowEvent.get().getRegistryBucketId()); flowEvent.setRegistryFlowId(latestFlowEvent.get().getRegistryFlowId()); flowEvent.setRegistryVersion(latestFlowEvent.get().getRegistryVersion()); final Date lastPublished = latestFlowEvent.get().getLastPublished(); flowEvent.setLastPublished(lastPublished == null ? null : lastPublished.getTime()); flowEvent.setLastPublishedUserIdentity(latestFlowEvent.get().getLastPublishedUserIdentity()); } if (flowEvent.getEventType() == FDFlowEventType.FLOW_PUBLISHED && flowEvent.getLastPublished() == null) { throw new IllegalArgumentException("Publish event must contain a published date"); } validate(flowEvent, "Cannot add flow event due to invalid event"); final FDFlowEventEntity flowEventEntity = modelMapper.map(flowEvent, FDFlowEventEntity.class); flowEventEntity.setFlowContent(serializeFlowContent(flowContent)); flowEventEntity.setFlow(flowEntity.get()); final FDFlowEventEntity savedFlowEventEntity = flowEventRepository.save(flowEventEntity); final String flowId = flowEntity.get().getId(); if (flowsById.containsKey(flowId)) { final FDVersionInfo versionInfo = createVersionInfo(savedFlowEventEntity); final FDFlow cachedFlow = flowsById.get(flowId); cachedFlow.setVersionInfo(versionInfo); cachedFlow.setLocalFlowRevision(savedFlowEventEntity.getFlowRevision()); } return modelMapper.map(savedFlowEventEntity, FDFlowEvent.class); } @Autowired  DatabaseFDFlowManager(final FDFlowRepository flowRepository,
                                 final FDFlowEventRepository flowEventRepository,
                                 final OptionalModelMapper modelMapper,
                                 final ObjectMapper objectMapper,
                                 final Validator validator); @Override Optional<FDFlowMetadata> getFlowMetadata(final String flowId); @Override Optional<FDFlow> getFlow(final String flowId); @Override FDFlowMetadata createFlow(final String agentClass, final NiFiUser user); @Override List<FDFlowMetadata> getAvailableFlows(); @Override List<FDFlowSummary> getFlowSummaries(); @Override FDFlowMetadata deleteFlow(final String flowId); @Override List<FDFlowEvent> getFlowEvents(final String flowId); @Override Optional<FDFlowEvent> getLatestFlowEvent(final String flowId); @Override Optional<FDFlowEvent> getLatestPublishFlowEvent(final String flowId); @Override Optional<FDFlowEvent> getFlowEvent(final String flowEventId); @Override FDFlowEvent addFlowEvent(final FDFlowEvent flowEvent, final VersionedProcessGroup flowContent, final NiFiUser user); @Override FDFlowEvent deleteFlowEvent(final String flowEventId); @Override void retainPublishEvents(final String flowId); @Override void revertToFlowRevision(final String flowId, final BigInteger flowRevision); static final FDFlowFormat CURRENT_FLOW_FORMAT; }
@Test(expected = IllegalArgumentException.class) public void testAddFlowEventAndTryToChangeRootGroupId() { final FDFlowEvent flowEvent = new FDFlowEvent(); flowEvent.setFlowIdentifier("2"); flowEvent.setComponentId("proc-1"); flowEvent.setEventType(FDFlowEventType.COMPONENT_ADDED); flowEvent.setEventDescription("Added TailFile processor"); flowEvent.setUserIdentity("test-user"); final VersionedProcessGroup pg = new VersionedProcessGroup(); pg.setIdentifier(UUID.randomUUID().toString()); pg.setName("root"); final NiFiUser user = new StandardNiFiUser.Builder().identity("system").build(); flowManager.addFlowEvent(flowEvent, pg, user); }
@Test(expected = IllegalArgumentException.class) public void testAddFlowEventWhenMissingFlowId() { final FDFlowEvent flowEvent = new FDFlowEvent(); flowEvent.setComponentId("proc-1"); flowEvent.setEventType(FDFlowEventType.COMPONENT_ADDED); flowEvent.setEventDescription("Added TailFile processor"); flowEvent.setUserIdentity("test-user"); final VersionedProcessGroup pg = new VersionedProcessGroup(); pg.setIdentifier(UUID.randomUUID().toString()); pg.setName("NEW ROOT GROUP"); final NiFiUser user = new StandardNiFiUser.Builder().identity("system").build(); flowManager.addFlowEvent(flowEvent, pg, user); }