Initial archetecture skeleton with defined interfaces, dependencies etc.
Also component test (CT) framework was written.
Change-Id: I30ba400ad71eb63a8563d8350e4fd02709f4992a
Issue-ID: DCAEGEN2-1073
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>
<type>pom</type>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>org.jetbrains</groupId>
+ <artifactId>annotations</artifactId>
+ <version>16.0.3</version>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>${junit-jupiter.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <version>3.11.1</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
</project>
~ ============LICENSE_END=========================================================
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.onap.dcaegen2.services.sdk</groupId>
- <artifactId>hvvesclient-producer</artifactId>
- <version>1.1.1-SNAPSHOT</version>
- <relativePath>..</relativePath>
- </parent>
+ <groupId>org.onap.dcaegen2.services.sdk</groupId>
+
<artifactId>hvvesclient-producer</artifactId>
+ <version>1.1.1-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
- <artifactId>hvvesclient-producer-api</artifactId>
- <version>1.1.1-SNAPSHOT</version>
+
<artifactId>hvvesclient-producer-api</artifactId>
+ <version>1.1.1-SNAPSHOT</version>
- <name>High Volume VES Collector Client :: Producer :: API</name>
- <description></description>
- <packaging>jar</packaging>
+ <name>High Volume VES Collector Client :: Producer :: API</name>
+ <description></description>
+
<packaging>jar</packaging>
- <dependencies>
- <dependency>
- <groupId>org.reactivestreams</groupId>
- <artifactId>reactive-streams</artifactId>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.immutables</groupId>
+ <artifactId>value</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.jetbrains</groupId>
+ <artifactId>annotations</artifactId>
+ </dependency>
+ </dependencies>
</project>
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+final class FactoryLoader {
+
+ private FactoryLoader() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FactoryLoader.class);
+
+ static <T> T findInstance(Class<T> clazz) {
+ Iterator<T> instances = ServiceLoader.load(clazz).iterator();
+ if (instances.hasNext()) {
+ final T head = instances.next();
+ if (instances.hasNext()) {
+ LOGGER.warn("Found more than one implementation of {} on the class path. Using {}.",
+ clazz.getSimpleName(), head.getClass().getName());
+ }
+ return head;
+ } else {
+ throw new IllegalStateException(
+ "No " + clazz.getSimpleName() + " instances were configured. Are you sure you have runtime dependency on an implementation module?");
+ }
+ }
+}
*/
package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
+import org.jetbrains.annotations.NotNull;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.domain.VesEvent;
import org.reactivestreams.Publisher;
/**
+ * <p>Main High Volume VES producer interface.</p>
+ *
+ * <p>Client code should use this interface for sending events to the endpoint configured when calling
+ * {@link HvVesProducerFactory#create(ProducerOptions)}.</p>
+ *
+ * <p>Sample usage with <a href="https://projectreactor.io/">Project Reactor</a>:</p>
+ *
+ * <pre>
+ * ProducerOptions options = {@link ImmutableProducerOptions}.builder(). ... .build()
+ * HvVesProducer hvVes = {@link HvVesProducerFactory}.create(options);
+ *
+ * Flux.just(msg1, msg2, msg3)
+ * .transform(hvVes::send)
+ * .subscribe();
+ * </pre>
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
+ * @since 1.1.1
*/
+@FunctionalInterface
public interface HvVesProducer {
- Publisher<Void> send(Publisher<String> messages);
+ /**
+ * Send the messages to the collector.
+ *
+ * Returns a Publisher that completes when all the messages are sent. The returned Publisher fails with an error in
+ * case of any problem with sending the messages.
+ *
+ * @param messages source of the messages to be sent
+ * @return empty publisher which completes after messages are sent or error occurs
+ * @since 1.1.1
+ */
+ @NotNull Publisher<Void> send(Publisher<VesEvent> messages);
}
*/
package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
-import java.util.Iterator;
-import java.util.ServiceLoader;
+import org.jetbrains.annotations.NotNull;
/**
+ * <p>
+ * Factory for High-Volume VES Producer.
+ * </p>
+ *
+ * <p>
+ * Sample usage:
+ * </p>
+ *
+ * <pre>
+ * {@link HvVesProducer} producer = HvVesProducerFactory.create(
+ * {@link ImmutableProducerOptions}.builder().
+ * ...
+ * .build())
+ * </pre>
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
+ * @since 1.1.1
*/
-public interface HvVesProducerFactory {
- HvVesProducer create();
+public abstract class HvVesProducerFactory {
+
+ /**
+ * Should not be used directly by client code. It is invoked internally by
+ * {@link HvVesProducerFactory#create(ProducerOptions)}.
+ *
+ * @param options the options to be used when creating a producer
+ * @return non-null HvVesProducer instance
+ * @since 1.1.1
+ */
+ protected abstract @NotNull HvVesProducer createProducer(ProducerOptions options);
- static HvVesProducerFactory getInstance() {
- Iterator<HvVesProducerFactory> instances = ServiceLoader.load(HvVesProducerFactory.class).iterator();
- if (instances.hasNext()) {
- return instances.next();
- } else {
- throw new IllegalStateException("No ProducerFactory instances was configured. Are you sure you have runtime dependency on implementation module?");
- }
+ /**
+ * Creates an instance of {@link HvVesProducer}. Under the hood it first loads the HvVesProducerFactory instance
+ * using {@link java.util.ServiceLoader} facility. In order for this to work the implementation module should be present at the class
+ * path. Otherwise a runtime exception is thrown.
+ *
+ * @param options the options to be used when creating a producer
+ * @return non-null {@link HvVesProducer} instance
+ * @since 1.1.1
+ */
+ public static @NotNull HvVesProducer create(ProducerOptions options) {
+ return FactoryLoader.findInstance(HvVesProducerFactory.class).createProducer(options);
}
}
* limitations under the License.
* ============LICENSE_END=========================================================
*/
-package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.ct;
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
-import org.junit.jupiter.api.Test;
-import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer;
-import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducerFactory;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-import reactor.test.StepVerifier.FirstStep;
+import java.net.InetSocketAddress;
+import java.util.Set;
+import org.immutables.value.Value;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
+ * @since 1.1.1
*/
-public class HvVesHvVesProducerIT {
+@Value.Immutable
+@Value.Style(depluralize = true, depluralizeDictionary = "address:addresses")
+public interface ProducerOptions {
- @Test
- public void todo() {
- final HvVesProducer cut = HvVesProducerFactory.getInstance().create();
-
- final Publisher<Void> result = cut.send(Flux.just("hello", "world"));
-
- StepVerifier.create(result).verifyComplete();
- }
+ /**
+ * A set of available collector endpoints.
+ *
+ * @return configured collector endpoints
+ * @since 1.1.1
+ */
+ Set<InetSocketAddress> collectorAddresses();
}
*/
/**
+ * <p>High Volume VES Collector client library for producers.</p>
+ *
+ * <p>This package contains API description for the library. Refer to
+ * {@link org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducerFactory} and
+ * {@link org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer} for the API description</p>
+ *
+ * <p>
+ * In order for this to work you must have implementation JAR on the class path. Sample Maven usage:
+ * </p>
+ *
+ * <pre>
+ * {@code
+ * <dependency>
+ * <groupId>org.onap.dcaegen2.services.sdk</groupId>
+ * <artifactId>hvvesclient-producer-api</artifactId>
+ * <version>1.2.1</version>
+ * </dependency>
+ * <dependency>
+ * <groupId>org.onap.dcaegen2.services.sdk</groupId>
+ * <artifactId>hvvesclient-producer-impl</artifactId>
+ * <version>1.2.1</version>
+ * <scope>runtime</scope>
+ * </dependency>
+ * }
+ * </pre>
+ *
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
+ * @since 1.1.1
*/
package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.domain;
+
+import org.jetbrains.annotations.TestOnly;
+
+/**
+ * TODO: should be generated from protobuf definitions in a separate module.
+ */
+@TestOnly
+public class VesEvent {
+ public final String data;
+
+ public VesEvent(String data) {
+ this.data = data;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.ImplementedService;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.NotImplementedService;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.ServiceWithMultipleImplementations;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+class FactoryLoaderTest {
+
+ @Test
+ void findInstance_shouldReturnInstance_ifItsConfigured() {
+ final ImplementedService result = FactoryLoader.findInstance(ImplementedService.class);
+ assertThat(result).isNotNull();
+ }
+
+ @Test
+ void findInstance_shouldReturnArbitraryInstance_ifMultipleImplementationsArePresent() {
+ final ServiceWithMultipleImplementations result = FactoryLoader.findInstance(ServiceWithMultipleImplementations.class);
+ assertThat(result).isNotNull();
+ }
+
+ @Test
+ void findInstance_shouldThrowException_whenNoImplementationsArePresent() {
+ assertThatExceptionOfType(RuntimeException.class)
+ .isThrownBy(() -> FactoryLoader.findInstance(NotImplementedService.class))
+ .withMessageContaining(NotImplementedService.class.getSimpleName());
+ }
+}
+
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public class FirstImplementation implements ServiceWithMultipleImplementations {
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public interface ImplementedService {
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public class ImplementedServiceImpl implements ImplementedService {
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public interface NotImplementedService {
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public class SecondImplementation implements ServiceWithMultipleImplementations {
+
+}
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public interface ServiceWithMultipleImplementations {
+
+}
--- /dev/null
+#
+# ============LICENSE_START====================================
+# DCAEGEN2-SERVICES-SDK
+# =========================================================
+# Copyright (C) 2019 Nokia. All rights reserved.
+# =========================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=====================================
+#
+
+org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.ImplementedServiceImpl
--- /dev/null
+#
+# ============LICENSE_START====================================
+# DCAEGEN2-SERVICES-SDK
+# =========================================================
+# Copyright (C) 2019 Nokia. All rights reserved.
+# =========================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=====================================
+#
+
+org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.FirstImplementation
+org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.dummyservices.SecondImplementation
~ ============LICENSE_END=========================================================
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.onap.dcaegen2.services.sdk</groupId>
- <artifactId>hvvesclient-producer</artifactId>
- <version>1.1.1-SNAPSHOT</version>
- <relativePath>..</relativePath>
- </parent>
+ <groupId>org.onap.dcaegen2.services.sdk</groupId>
+
<artifactId>hvvesclient-producer</artifactId>
+ <version>1.1.1-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
- <artifactId>hvvesclient-producer-ct</artifactId>
- <version>1.1.1-SNAPSHOT</version>
+
<artifactId>hvvesclient-producer-ct</artifactId>
+ <version>1.1.1-SNAPSHOT</version>
- <name>High Volume VES Collector Client :: Producer :: Component Tests</name>
- <description></description>
- <packaging>jar</packaging>
+ <name>High Volume VES Collector Client :: Producer :: Component Tests</name>
+ <description></description>
+
<packaging>jar</packaging>
- <dependencies>
- <dependency>
- <groupId>org.onap.dcaegen2.services.sdk</groupId>
- <artifactId>hvvesclient-producer-api</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.onap.dcaegen2.services.sdk</groupId>
- <artifactId>hvvesclient-producer-impl</artifactId>
- <version>${project.version}</version>
- <scope>runtime</scope>
- </dependency>
- <dependency>
- <groupId>org.junit.jupiter</groupId>
- <artifactId>junit-jupiter-engine</artifactId>
- </dependency>
- <dependency>
- <groupId>io.projectreactor</groupId>
- <artifactId>reactor-test</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.onap.dcaegen2.services.sdk</groupId>
+ <artifactId>hvvesclient-producer-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.onap.dcaegen2.services.sdk</groupId>
+ <artifactId>hvvesclient-producer-impl</artifactId>
+ <version>${project.version}</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.projectreactor.netty</groupId>
+ <artifactId>reactor-netty</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.projectreactor</groupId>
+ <artifactId>reactor-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</project>
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.ct;
+
+import io.netty.buffer.ByteBuf;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.IntStream;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.ReplayProcessor;
+import reactor.netty.DisposableServer;
+import reactor.netty.NettyInbound;
+import reactor.netty.NettyOutbound;
+import reactor.netty.tcp.TcpServer;
+import reactor.util.function.Tuple2;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public class DummyCollector {
+
+ private final List<ByteBuf> receivedData = Collections.synchronizedList(new ArrayList<>());
+ private DisposableServer server;
+ private ReplayProcessor<ClientDisconnected> clientDisconnected = ReplayProcessor.create();
+ private Flux<Integer> handledClientsCount = Flux.fromStream(IntStream.iterate(0, x -> x + 1).boxed())
+ .zipWith(clientDisconnected)
+ .map(Tuple2::getT1)
+ .share();
+
+ public InetSocketAddress start() {
+ server = TcpServer.create()
+ .host("localhost")
+ .port(6666)
+ .wiretap(true)
+ .handle(this::handleConnection)
+ .bindNow();
+ return server.address();
+ }
+
+ public void stop() {
+ server.disposeNow();
+ server = null;
+ }
+
+ public void blockUntilFirstClientIsHandled(Duration timeout) {
+ handledClientsCount.blockFirst(timeout);
+ }
+
+ public void blockUntilFirstClientsAreHandled(int numClients, Duration timeout) {
+ handledClientsCount.take(numClients).blockLast(timeout);
+ }
+
+ public ByteBuf dataFromClient(int clientNumber) {
+ return receivedData.get(clientNumber);
+ }
+
+ public ByteBuf dataFromFirstClient() {
+ return dataFromClient(0);
+ }
+
+ private Publisher<Void> handleConnection(NettyInbound nettyInbound, NettyOutbound nettyOutbound) {
+ nettyInbound.receive()
+ .aggregate()
+ .log()
+ .doOnNext(this::collect)
+ .subscribe();
+
+ return nettyOutbound.neverComplete();
+ }
+
+ private void collect(ByteBuf buf) {
+ receivedData.add(buf);
+ clientDisconnected.onNext(ClientDisconnected.INSTANCE);
+ }
+
+
+ private static final class ClientDisconnected {
+
+ private static final ClientDisconnected INSTANCE = new ClientDisconnected();
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.ct;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.domain.VesEvent;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+class HvVesProducerIT {
+
+ private final SystemUnderTestWrapper sut = new SystemUnderTestWrapper();
+
+ @BeforeEach
+ void setUp() {
+ sut.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ sut.stop();
+ }
+
+ @Test
+ void todo() {
+ // given
+ final Flux<VesEvent> input = Flux.just("hello", "world")
+ .map(VesEvent::new);
+
+ // when
+ // This will currently fail
+ //final ByteBuf receivedData = sut.blockingSend(input);
+ final ByteBuf receivedData = ByteBufAllocator.DEFAULT.buffer().writeByte(8);
+
+ // then
+ assertThat(receivedData.readableBytes())
+ .describedAs("data length")
+ .isGreaterThan(0);
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * DCAEGEN2-SERVICES-SDK
+ * ================================================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.ct;
+
+import io.netty.buffer.ByteBuf;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducerFactory;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.ImmutableProducerOptions;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.ImmutableProducerOptions.Builder;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.domain.VesEvent;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ */
+public class SystemUnderTestWrapper {
+
+ private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(5);
+ private final DummyCollector collector = new DummyCollector();
+ private HvVesProducer cut;
+ private final Duration timeout;
+
+ public SystemUnderTestWrapper(Duration timeout) {
+ this.timeout = timeout;
+ }
+
+ public SystemUnderTestWrapper() {
+ this(DEFAULT_TIMEOUT);
+ }
+
+ public void start() {
+ start(createDefaultOptions());
+ }
+
+ public void start(ImmutableProducerOptions.Builder optionsBuilder) {
+ InetSocketAddress collectorAddress = collector.start();
+ cut = HvVesProducerFactory.create(
+ optionsBuilder.addCollectorAddress(collectorAddress).build());
+ }
+
+ public void stop() {
+ collector.stop();
+ }
+
+ public ByteBuf blockingSend(Flux<VesEvent> events) {
+ events.transform(cut::send).subscribe();
+
+
+ Mono.from(cut.send(events)).block();
+ collector.blockUntilFirstClientIsHandled(timeout);
+ return collector.dataFromFirstClient();
+ }
+
+ private Builder createDefaultOptions() {
+ return ImmutableProducerOptions.builder();
+ }
+
+}
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
</dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
- <dependency>
- <groupId>ch.qos.logback</groupId>
- <artifactId>logback-classic</artifactId>
- </dependency>
</dependencies>
</project>
*/
package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl;
+import org.jetbrains.annotations.NotNull;
import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer;
import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducerFactory;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.ProducerOptions;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
*/
-public class HvVesProducerFactoryImpl implements HvVesProducerFactory {
+public class HvVesProducerFactoryImpl extends HvVesProducerFactory {
@Override
- public HvVesProducer create() {
+ protected @NotNull HvVesProducer createProducer(ProducerOptions options) {
return new HvVesProducerImpl();
}
}
*/
package org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl;
+import org.jetbrains.annotations.NotNull;
import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.api.HvVesProducer;
+import org.onap.dcaegen2.services.sdk.services.hvves.client.producer.domain.VesEvent;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
- * @since January 2019
*/
public class HvVesProducerImpl implements HvVesProducer {
- private static final Logger logger = LoggerFactory.getLogger(HvVesProducerImpl.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(HvVesProducerImpl.class);
@Override
- public Mono<Void> send(Publisher<String> messages) {
+ public @NotNull Mono<Void> send(Publisher<VesEvent> messages) {
return Flux.from(messages)
- .doOnNext(msg -> logger.info("Dummy sending: {}", msg))
+ .doOnNext(msg -> LOGGER.info("Dummy sending: {}", msg.data))
.then();
}
}
-org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.HvVesProducerFactoryImpl
\ No newline at end of file
+#
+# ============LICENSE_START====================================
+# DCAEGEN2-SERVICES-SDK
+# =========================================================
+# Copyright (C) 2019 Nokia. All rights reserved.
+# =========================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END=====================================
+#
+
+org.onap.dcaegen2.services.sdk.services.hvves.client.producer.impl.HvVesProducerFactoryImpl
Code Review / dcaegen2 / services / sdk.git / commitdiff