FlowRefresherTask implements Runnable { void refreshAgentClass(final String agentClassName, final AtomicInteger refreshedCount) throws FlowMapperException, IOException, FlowRetrievalException, SerializerException, FlowTransformException { final Optional<FlowUri> optionalFlowUri = flowMapper.getFlowMapping(agentClassName); if (!optionalFlowUri.isPresent()) { LOGGER.debug("No flow mapping present for '{}'", new Object[] {agentClassName}); return; } final FlowUri flowUri = optionalFlowUri.get(); LOGGER.debug("Found flow {} mapped to '{}'", new Object[]{flowUri.getFlowId(), agentClassName}); final VersionedFlowSnapshot latestFlow = flowRetrievalService.getLatestFlow(flowUri); if (latestFlow == null) { LOGGER.debug("No flow versions exist for '{}'", new Object[] {agentClassName}); return; } final Optional<FlowMappingEntity> optionalFlowMapping = flowMappingRepository.findById(agentClassName); if (optionalFlowMapping.isPresent()) { final FlowMappingEntity existingFlowMapping = optionalFlowMapping.get(); final String existingRegistryUrl = existingFlowMapping.getFlowEntity().getRegistryUrl(); final String existingBucketId = existingFlowMapping.getFlowEntity().getRegistryBucketId(); final String existingFlowId = existingFlowMapping.getFlowEntity().getRegistryFlowId(); if (flowUri.getRegistryUrl().equals(existingRegistryUrl) && flowUri.getBucketId().equals(existingBucketId) && flowUri.getFlowId().equals(existingFlowId)) { final FlowEntity existingFlow = existingFlowMapping.getFlowEntity(); if (latestFlow.getSnapshotMetadata().getVersion() > existingFlow.getRegistryFlowVersion()) { updateFlowMapping(existingFlowMapping, flowUri, latestFlow); refreshedCount.incrementAndGet(); } else { LOGGER.debug("'{}' already mapped to latest version, nothing to do...", new Object[]{agentClassName}); } } else { updateFlowMapping(existingFlowMapping, flowUri, latestFlow); refreshedCount.incrementAndGet(); } } else { createNewFlowMapping(agentClassName, flowUri, latestFlow); refreshedCount.incrementAndGet(); } } @Autowired  FlowRefresherTask(
            final AgentClassRepository agentClassRepository,
            final FlowMapper flowMapper,
            final FlowRetrievalService flowRetrievalService,
            final FlowTransformService<ConfigSchema> flowTransformService,
            final Serializer<ConfigSchema> configSchemaSerializer,
            final FlowRepository flowRepository,
            final FlowMappingRepository flowMappingRepository,
            final TransactionTemplate transactionTemplate); @Override void run();  }
@Test public void testRefreshWhenRegistryInfoChanged() throws FlowRetrievalException, FlowMapperException, SerializerException, FlowTransformException, IOException { final String className = "Test Class"; final AtomicInteger refreshedCounter = new AtomicInteger(0); final FlowUri flowUri = new FlowUri("http: when(flowMapper.getFlowMapping(className)).thenReturn(Optional.of(flowUri)); final Integer flowVersion = new Integer(1); final VersionedFlowSnapshot flowSnapshot = FlowTestUtils.createSnapshot(flowUri.getBucketId(), flowUri.getFlowId(), flowVersion); when(flowRetrievalService.getLatestFlow(flowUri)).thenReturn(flowSnapshot); final FlowMappingEntity flowMappingEntity = FlowTestUtils.createFlowMappingEntity(className, flowUri, flowVersion, "YAML"); flowMappingEntity.getFlowEntity().setRegistryUrl("http: when(flowMappingRepository.findById(className)).thenReturn(Optional.of(flowMappingEntity)); flowRefresherTask.refreshAgentClass(className, refreshedCounter); assertEquals(1, refreshedCounter.get()); }
@Test public void testRefreshWhenNoFlowMappingExists() throws FlowMapperException, IOException, FlowRetrievalException, SerializerException, FlowTransformException { final String className = "Test Class"; final AtomicInteger refreshedCounter = new AtomicInteger(0); final FlowUri flowUri = new FlowUri("http: when(flowMapper.getFlowMapping(className)).thenReturn(Optional.of(flowUri)); final Integer flowVersion = new Integer(1); final VersionedFlowSnapshot flowSnapshot = FlowTestUtils.createSnapshot(flowUri.getBucketId(), flowUri.getFlowId(), flowVersion); when(flowRetrievalService.getLatestFlow(flowUri)).thenReturn(flowSnapshot); when(flowMappingRepository.findById(className)).thenReturn(Optional.empty()); flowRefresherTask.refreshAgentClass(className, refreshedCounter); assertEquals(1, refreshedCounter.get()); verify(flowTransformService, times(1)).transform(flowSnapshot); }
@Test public void testRefreshWhenNewerVersionExists() throws FlowRetrievalException, FlowMapperException, SerializerException, FlowTransformException, IOException { final String className = "Test Class"; final AtomicInteger refreshedCounter = new AtomicInteger(0); final FlowUri flowUri = new FlowUri("http: when(flowMapper.getFlowMapping(className)).thenReturn(Optional.of(flowUri)); final VersionedFlowSnapshot flowSnapshot = FlowTestUtils.createSnapshot(flowUri.getBucketId(), flowUri.getFlowId(), 2); when(flowRetrievalService.getLatestFlow(flowUri)).thenReturn(flowSnapshot); final FlowMappingEntity flowMappingEntity = FlowTestUtils.createFlowMappingEntity(className, flowUri, 1, "YAML"); when(flowMappingRepository.findById(className)).thenReturn(Optional.of(flowMappingEntity)); flowRefresherTask.refreshAgentClass(className, refreshedCounter); assertEquals(1, refreshedCounter.get()); }
@Test public void testRefreshWhenSameVersionFound() throws FlowRetrievalException, FlowMapperException, SerializerException, FlowTransformException, IOException { final String className = "Test Class"; final AtomicInteger refreshedCounter = new AtomicInteger(0); final FlowUri flowUri = new FlowUri("http: when(flowMapper.getFlowMapping(className)).thenReturn(Optional.of(flowUri)); final Integer flowVersion = new Integer(1); final VersionedFlowSnapshot flowSnapshot = FlowTestUtils.createSnapshot(flowUri.getBucketId(), flowUri.getFlowId(), flowVersion); when(flowRetrievalService.getLatestFlow(flowUri)).thenReturn(flowSnapshot); final FlowMappingEntity flowMappingEntity = FlowTestUtils.createFlowMappingEntity(className, flowUri, flowVersion, "YAML"); when(flowMappingRepository.findById(className)).thenReturn(Optional.of(flowMappingEntity)); flowRefresherTask.refreshAgentClass(className, refreshedCounter); assertEquals(0, refreshedCounter.get()); }