MicroPipelineValidator { protected MicroPipelineValidationResult validateComponent(final MicroPipelineComponentConfiguration componentCfg, final Set<String> queueIdentifiers, final Set<String> componentIdentifiers) { if(componentCfg == null) return MicroPipelineValidationResult.MISSING_COMPONENT_CONFIGURATION; if(StringUtils.isBlank(componentCfg.getId())) return MicroPipelineValidationResult.MISSING_COMPONENT_ID; if(StringUtils.isBlank(componentCfg.getName())) return MicroPipelineValidationResult.MISSING_COMPONENT_NAME; if(StringUtils.isBlank(componentCfg.getVersion())) return MicroPipelineValidationResult.MISSING_COMPONENT_VERSION; if(componentCfg.getType() == null) return MicroPipelineValidationResult.MISSING_COMPONENT_TYPE; String tempId = StringUtils.lowerCase(StringUtils.trim(componentCfg.getId())); if(componentIdentifiers.contains(tempId)) return MicroPipelineValidationResult.NON_UNIQUE_COMPONENT_ID; if(componentCfg.getType() == MicroPipelineComponentType.SOURCE) { if(StringUtils.isNotBlank(componentCfg.getFromQueue())) return MicroPipelineValidationResult.NOT_PERMITTED_SOURCE_QUEUE_REF; if(StringUtils.isBlank(componentCfg.getToQueue())) return MicroPipelineValidationResult.MISSING_DESTINATION_QUEUE; String destinationQueueId = StringUtils.lowerCase(StringUtils.trim(componentCfg.getToQueue())); if(!queueIdentifiers.contains(destinationQueueId)) return MicroPipelineValidationResult.UNKNOWN_DESTINATION_QUEUE; } else if(componentCfg.getType() == MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR || componentCfg.getType() == MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR) { if(StringUtils.isBlank(componentCfg.getFromQueue())) return MicroPipelineValidationResult.MISSING_SOURCE_QUEUE; if(StringUtils.isBlank(componentCfg.getToQueue())) return MicroPipelineValidationResult.MISSING_DESTINATION_QUEUE; String sourceQueueId = StringUtils.lowerCase(StringUtils.trim(componentCfg.getFromQueue())); if(!queueIdentifiers.contains(sourceQueueId)) return MicroPipelineValidationResult.UNKNOWN_SOURCE_QUEUE; String destinationQueueId = StringUtils.lowerCase(StringUtils.trim(componentCfg.getToQueue())); if(!queueIdentifiers.contains(destinationQueueId)) return MicroPipelineValidationResult.UNKNOWN_DESTINATION_QUEUE; } else if(componentCfg.getType() == MicroPipelineComponentType.EMITTER) { if(StringUtils.isNotBlank(componentCfg.getToQueue())) return MicroPipelineValidationResult.NOT_PERMITTED_DESTINATION_QUEUE_REF; if(StringUtils.isBlank(componentCfg.getFromQueue())) return MicroPipelineValidationResult.MISSING_SOURCE_QUEUE; String destinationQueueId = StringUtils.lowerCase(StringUtils.trim(componentCfg.getFromQueue())); if(!queueIdentifiers.contains(destinationQueueId)) return MicroPipelineValidationResult.UNKNOWN_SOURCE_QUEUE; } return MicroPipelineValidationResult.OK; }  MicroPipelineValidationResult validate(final MicroPipelineConfiguration configuration);  }
@Test public void testValidateComponent_withMissingComponentVersion() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-id"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setVersion(""); Assert.assertEquals("Missing component version", MicroPipelineValidationResult.MISSING_COMPONENT_VERSION, new MicroPipelineValidator().validateComponent(cfg, queues, Collections.<String>emptySet())); }
@Test public void testValidateComponent_withMissingComponentType() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-id"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(null); cfg.setVersion("test-version"); Assert.assertEquals("Missing component type", MicroPipelineValidationResult.MISSING_COMPONENT_TYPE, new MicroPipelineValidator().validateComponent(cfg, queues, Collections.<String>emptySet())); }
@Test public void testValidateComponent_withDuplicateComponentId() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); components.add("test-component"); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Duplicate component ids are not permitted", MicroPipelineValidationResult.NON_UNIQUE_COMPONENT_ID, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withSourceReferencingSourceQueue() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setVersion("test-version"); Assert.assertEquals("Source queues not permitted for source components", MicroPipelineValidationResult.NOT_PERMITTED_SOURCE_QUEUE_REF, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withSourceMissingDestinationQueue() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue(""); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue(""); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setVersion("test-version"); Assert.assertEquals("Missing destination queue", MicroPipelineValidationResult.MISSING_DESTINATION_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withSourceUnknownDestinationQueue() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue(""); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("unknown-queue"); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.UNKNOWN_DESTINATION_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withValidSource() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue(""); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.OK, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withOperatorMissingSourceQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue(""); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.MISSING_SOURCE_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withOperatorMissingDestinationQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue(""); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.MISSING_DESTINATION_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withOperatorUnknownSourceQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("unknown-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.UNKNOWN_SOURCE_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withOperatorUnknownDestinationQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("unknown-output"); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.UNKNOWN_DESTINATION_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withValidOperator() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.OK, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withEmitterDestinationQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.NOT_PERMITTED_DESTINATION_QUEUE_REF, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withEmitterMissingSourceQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue(""); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue(""); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.MISSING_SOURCE_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withEmitterUnknownSourceQueueRef() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("unknown-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue(""); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.UNKNOWN_SOURCE_QUEUE, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withValidEmitter() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); Set<String> components = new HashSet<String>(); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-component"); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue(""); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setVersion("test-version"); Assert.assertEquals("Unknown destination queue", MicroPipelineValidationResult.OK, new MicroPipelineValidator().validateComponent(cfg, queues, components)); }
@Test public void testValidateComponent_withNullInput() { Assert.assertEquals("Missing component configuration", MicroPipelineValidationResult.MISSING_COMPONENT_CONFIGURATION, new MicroPipelineValidator().validateComponent(null, Collections.<String>emptySet(), Collections.<String>emptySet())); }
@Test public void testValidateComponent_withMissingComponentId() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId(""); cfg.setName("test-name"); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Missing component id", MicroPipelineValidationResult.MISSING_COMPONENT_ID, new MicroPipelineValidator().validateComponent(cfg, queues, Collections.<String>emptySet())); }
@Test public void testValidateComponent_withMissingComponentName() { Set<String> queues = new HashSet<>(); queues.add("test-input"); queues.add("test-output"); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setFromQueue("test-input"); cfg.setId("test-id"); cfg.setName(""); cfg.setSettings(new Properties()); cfg.setToQueue("test-output"); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setVersion("test-version"); Assert.assertEquals("Missing component name", MicroPipelineValidationResult.MISSING_COMPONENT_NAME, new MicroPipelineValidator().validateComponent(cfg, queues, Collections.<String>emptySet())); }