KafkaEndpointComponent extends AbstractEndpointComponent { @Override protected Endpoint createEndpoint(String resourcePath, Map<String, String> parameters, TestContext context) { KafkaEndpoint endpoint = new KafkaEndpoint(); if (resourcePath.indexOf(':') > 0) { endpoint.getEndpointConfiguration().setTopic(resourcePath.substring(resourcePath.lastIndexOf(':') + 1)); } else { endpoint.getEndpointConfiguration().setTopic(resourcePath); } enrichEndpointConfiguration(endpoint.getEndpointConfiguration(), parameters, context); return endpoint; } KafkaEndpointComponent();   }
@Test public void testCreateEndpoint() { KafkaEndpointComponent component = new KafkaEndpointComponent(); Endpoint endpoint = component.createEndpoint("kafka:test", context); Assert.assertEquals(endpoint.getClass(), KafkaEndpoint.class); Assert.assertEquals(((KafkaEndpoint)endpoint).getEndpointConfiguration().getTopic(), "test"); Assert.assertEquals(((KafkaEndpoint) endpoint).getEndpointConfiguration().getServer(), "localhost:9092"); Assert.assertEquals(((KafkaEndpoint) endpoint).getEndpointConfiguration().getTimeout(), 5000L); }
@Test public void testCreateEndpointWithParameters() { KafkaEndpointComponent component = new KafkaEndpointComponent(); Endpoint endpoint = component.createEndpoint("kafka:test?server=localhost:9091&timeout=10000", context); Assert.assertEquals(endpoint.getClass(), KafkaEndpoint.class); Assert.assertEquals(((KafkaEndpoint)endpoint).getEndpointConfiguration().getTopic(), "test"); Assert.assertEquals(((KafkaEndpoint) endpoint).getEndpointConfiguration().getServer(), "localhost:9091"); Assert.assertEquals(((KafkaEndpoint) endpoint).getEndpointConfiguration().getTimeout(), 10000L); }
@Test public void testCreateEndpointWithNullParameters() { KafkaEndpointComponent component = new KafkaEndpointComponent(); Endpoint endpoint = component.createEndpoint("kafka:test?server", context); Assert.assertEquals(endpoint.getClass(), KafkaEndpoint.class); Assert.assertEquals(((KafkaEndpoint)endpoint).getEndpointConfiguration().getTopic(), "test"); Assert.assertNull(((KafkaEndpoint) endpoint).getEndpointConfiguration().getServer()); }
@Test public void testInvalidEndpointUri() { KafkaEndpointComponent component = new KafkaEndpointComponent(); try { component.createEndpoint("kafka:test?param1=&param2=value2", context); Assert.fail("Missing exception due to invalid endpoint uri"); } catch (CitrusRuntimeException e) { Assert.assertTrue(e.getMessage().startsWith("Unable to find parameter"), e.getMessage()); } }