From 7c73254c00e5dc598f68e074d160b2d263872cda Mon Sep 17 00:00:00 2001 From: Lukasz Muszkieta Date: Wed, 25 Apr 2018 11:54:47 +0200 Subject: PnfReadyEventConsumer implementation Change-Id: Ic8d5814c555bad436bfcbe1b4e212637a6800947 Issue-ID: SO-466 Signed-off-by: Lukasz Muszkieta --- .../pnf/dmaap/PnfEventReadyConsumer.java | 173 -------------------- .../pnf/dmaap/PnfEventReadyDmaapClient.java | 173 ++++++++++++++++++++ .../src/main/webapp/WEB-INF/applicationContext.xml | 6 +- .../pnf/dmaap/PnfEventReadyConsumerTest.java | 173 -------------------- .../pnf/dmaap/PnfEventReadyDmaapClientTest.java | 176 +++++++++++++++++++++ 5 files changed, 352 insertions(+), 349 deletions(-) delete mode 100644 bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumer.java create mode 100644 bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java delete mode 100644 bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumerTest.java create mode 100644 bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java diff --git a/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumer.java b/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumer.java deleted file mode 100644 index 1fd2de97f2..0000000000 --- a/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumer.java +++ /dev/null @@ -1,173 +0,0 @@ -/*- - * ============LICENSE_START======================================================= - * ONAP - SO - * ================================================================================ - * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.mso.bpmn.infrastructure.pnf.dmaap; - -import com.google.common.annotations.VisibleForTesting; -import java.io.IOException; -import java.net.URI; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import javax.ws.rs.core.UriBuilder; -import org.apache.http.HttpResponse; -import org.apache.http.client.HttpClient; -import org.apache.http.client.methods.HttpGet; -import org.apache.http.impl.client.HttpClientBuilder; -import org.apache.http.util.EntityUtils; -import org.openecomp.mso.jsonpath.JsonPathUtil; -import org.openecomp.mso.logger.MsoLogger; - -public class PnfEventReadyConsumer implements DmaapClient { - - private static final MsoLogger LOGGER = MsoLogger.getMsoLogger(MsoLogger.Catalog.RA); - - private static final String JSON_PATH_CORRELATION_ID = "$.pnfRegistrationFields.correlationId"; - private HttpClient httpClient; - private String dmaapHost; - private int dmaapPort; - private String dmaapProtocol; - private String dmaapUriPathPrefix; - private String dmaapTopicName; - private String consumerId; - private String consumerGroup; - private Map pnfCorrelationIdToThreadMap; - private HttpGet getRequest; - private ScheduledExecutorService executor; - private int dmaapClientDelayInSeconds; - private volatile boolean dmaapThreadListenerIsRunning; - - public PnfEventReadyConsumer() { - httpClient = HttpClientBuilder.create().build(); - pnfCorrelationIdToThreadMap = new ConcurrentHashMap<>(); - executor = null; - } - - public void init() { - getRequest = new HttpGet(buildURI()); - } - - //TODO: extract this logic to separate class and test it there to avoid using VisibleForTesting - @VisibleForTesting - void sendRequest() { - try { - HttpResponse response = httpClient.execute(getRequest); - getCorrelationIdFromResponse(response).ifPresent(this::informAboutPnfReadyIfCorrelationIdFound); - } catch (IOException e) { - LOGGER.error("Exception caught during sending rest request to dmaap for listening event topic", e); - } - } - - @Override - public synchronized void registerForUpdate(String correlationId, Runnable informConsumer) { - pnfCorrelationIdToThreadMap.put(correlationId, informConsumer); - if (!dmaapThreadListenerIsRunning) { - startDmaapThreadListener(); - } - } - - @Override - public synchronized Runnable unregister(String correlationId) { - Runnable runnable = pnfCorrelationIdToThreadMap.remove(correlationId); - if (pnfCorrelationIdToThreadMap.isEmpty()) { - stopDmaapThreadListener(); - } - return runnable; - } - - private synchronized void startDmaapThreadListener() { - if (!dmaapThreadListenerIsRunning) { - executor = Executors.newScheduledThreadPool(1); - executor.scheduleWithFixedDelay(this::sendRequest, 0, - dmaapClientDelayInSeconds, TimeUnit.SECONDS); - dmaapThreadListenerIsRunning = true; - } - } - - private synchronized void stopDmaapThreadListener() { - if (dmaapThreadListenerIsRunning) { - executor.shutdownNow(); - dmaapThreadListenerIsRunning = false; - executor = null; - } - } - - private URI buildURI() { - return UriBuilder.fromUri(dmaapUriPathPrefix) - .scheme(dmaapProtocol) - .host(dmaapHost) - .port(dmaapPort).path(dmaapTopicName) - .path(consumerGroup).path(consumerId).build(); - } - - private Optional getCorrelationIdFromResponse(HttpResponse response) throws IOException { - if (response.getStatusLine().getStatusCode() == 200) { - String responseString = EntityUtils.toString(response.getEntity(), "UTF-8"); - if (responseString != null) { - return JsonPathUtil.getInstance().locateResult(responseString, JSON_PATH_CORRELATION_ID); - } - } - return Optional.empty(); - } - - - private synchronized void informAboutPnfReadyIfCorrelationIdFound(String correlationId) { - Runnable runnable = unregister(correlationId); - if (runnable != null) { - runnable.run(); - } - } - - public void setDmaapHost(String dmaapHost) { - this.dmaapHost = dmaapHost; - } - - public void setDmaapPort(int dmaapPort) { - this.dmaapPort = dmaapPort; - } - - public void setDmaapProtocol(String dmaapProtocol) { - this.dmaapProtocol = dmaapProtocol; - } - - public void setDmaapUriPathPrefix(String dmaapUriPathPrefix) { - this.dmaapUriPathPrefix = dmaapUriPathPrefix; - } - - public void setDmaapTopicName(String dmaapTopicName) { - this.dmaapTopicName = dmaapTopicName; - } - - public void setConsumerId(String consumerId) { - this.consumerId = consumerId; - } - - public void setConsumerGroup(String consumerGroup) { - this.consumerGroup = consumerGroup; - } - - public void setDmaapClientDelayInSeconds(int dmaapClientDelayInSeconds) { - this.dmaapClientDelayInSeconds = dmaapClientDelayInSeconds; - } - -} diff --git a/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java b/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java new file mode 100644 index 0000000000..830574bad4 --- /dev/null +++ b/bpmn/MSOInfrastructureBPMN/src/main/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClient.java @@ -0,0 +1,173 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - SO + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.mso.bpmn.infrastructure.pnf.dmaap; + +import java.io.IOException; +import java.net.URI; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import javax.ws.rs.core.UriBuilder; +import org.apache.http.HttpResponse; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.util.EntityUtils; +import org.openecomp.mso.jsonpath.JsonPathUtil; +import org.openecomp.mso.logger.MsoLogger; + +public class PnfEventReadyDmaapClient implements DmaapClient { + + private static final MsoLogger LOGGER = MsoLogger.getMsoLogger(MsoLogger.Catalog.RA); + + private static final String JSON_PATH_CORRELATION_ID = "$.pnfRegistrationFields.correlationId"; + private HttpClient httpClient; + private String dmaapHost; + private int dmaapPort; + private String dmaapProtocol; + private String dmaapUriPathPrefix; + private String dmaapTopicName; + private String consumerId; + private String consumerGroup; + private Map pnfCorrelationIdToThreadMap; + private HttpGet getRequest; + private ScheduledExecutorService executor; + private int dmaapClientDelayInSeconds; + private volatile boolean dmaapThreadListenerIsRunning; + + public PnfEventReadyDmaapClient() { + httpClient = HttpClientBuilder.create().build(); + pnfCorrelationIdToThreadMap = new ConcurrentHashMap<>(); + executor = null; + } + + public void init() { + getRequest = new HttpGet(buildURI()); + } + + @Override + public synchronized void registerForUpdate(String correlationId, Runnable informConsumer) { + pnfCorrelationIdToThreadMap.put(correlationId, informConsumer); + if (!dmaapThreadListenerIsRunning) { + startDmaapThreadListener(); + } + } + + @Override + public synchronized Runnable unregister(String correlationId) { + Runnable runnable = pnfCorrelationIdToThreadMap.remove(correlationId); + if (pnfCorrelationIdToThreadMap.isEmpty()) { + stopDmaapThreadListener(); + } + return runnable; + } + + private synchronized void startDmaapThreadListener() { + if (!dmaapThreadListenerIsRunning) { + executor = Executors.newScheduledThreadPool(1); + executor.scheduleWithFixedDelay(new DmaapTopicListenerThread(), 0, + dmaapClientDelayInSeconds, TimeUnit.SECONDS); + dmaapThreadListenerIsRunning = true; + } + } + + private synchronized void stopDmaapThreadListener() { + if (dmaapThreadListenerIsRunning) { + executor.shutdownNow(); + dmaapThreadListenerIsRunning = false; + executor = null; + } + } + + private URI buildURI() { + return UriBuilder.fromUri(dmaapUriPathPrefix) + .scheme(dmaapProtocol) + .host(dmaapHost) + .port(dmaapPort).path(dmaapTopicName) + .path(consumerGroup).path(consumerId).build(); + } + + public void setDmaapHost(String dmaapHost) { + this.dmaapHost = dmaapHost; + } + + public void setDmaapPort(int dmaapPort) { + this.dmaapPort = dmaapPort; + } + + public void setDmaapProtocol(String dmaapProtocol) { + this.dmaapProtocol = dmaapProtocol; + } + + public void setDmaapUriPathPrefix(String dmaapUriPathPrefix) { + this.dmaapUriPathPrefix = dmaapUriPathPrefix; + } + + public void setDmaapTopicName(String dmaapTopicName) { + this.dmaapTopicName = dmaapTopicName; + } + + public void setConsumerId(String consumerId) { + this.consumerId = consumerId; + } + + public void setConsumerGroup(String consumerGroup) { + this.consumerGroup = consumerGroup; + } + + public void setDmaapClientDelayInSeconds(int dmaapClientDelayInSeconds) { + this.dmaapClientDelayInSeconds = dmaapClientDelayInSeconds; + } + + class DmaapTopicListenerThread implements Runnable { + + @Override + public void run() { + try { + HttpResponse response = httpClient.execute(getRequest); + getCorrelationIdFromResponse(response).ifPresent(this::informAboutPnfReadyIfCorrelationIdFound); + } catch (IOException e) { + LOGGER.error("Exception caught during sending rest request to dmaap for listening event topic", e); + } + } + + private Optional getCorrelationIdFromResponse(HttpResponse response) throws IOException { + if (response.getStatusLine().getStatusCode() == 200) { + String responseString = EntityUtils.toString(response.getEntity(), "UTF-8"); + if (responseString != null) { + return JsonPathUtil.getInstance().locateResult(responseString, JSON_PATH_CORRELATION_ID); + } + } + return Optional.empty(); + } + + private synchronized void informAboutPnfReadyIfCorrelationIdFound(String correlationId) { + Runnable runnable = unregister(correlationId); + if (runnable != null) { + runnable.run(); + } + } + } + +} diff --git a/bpmn/MSOInfrastructureBPMN/src/main/webapp/WEB-INF/applicationContext.xml b/bpmn/MSOInfrastructureBPMN/src/main/webapp/WEB-INF/applicationContext.xml index 4ddeba16be..13ab4f8f17 100644 --- a/bpmn/MSOInfrastructureBPMN/src/main/webapp/WEB-INF/applicationContext.xml +++ b/bpmn/MSOInfrastructureBPMN/src/main/webapp/WEB-INF/applicationContext.xml @@ -15,14 +15,14 @@ - + - + - diff --git a/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumerTest.java b/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumerTest.java deleted file mode 100644 index 73b8247ebc..0000000000 --- a/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyConsumerTest.java +++ /dev/null @@ -1,173 +0,0 @@ -/*- - * ============LICENSE_START======================================================= - * ONAP - SO - * ================================================================================ - * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.mso.bpmn.infrastructure.pnf.dmaap; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Matchers.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyZeroInteractions; -import static org.mockito.Mockito.when; - -import java.io.IOException; -import java.io.UnsupportedEncodingException; -import java.lang.reflect.Field; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ScheduledExecutorService; -import org.apache.http.HttpEntity; -import org.apache.http.HttpResponse; -import org.apache.http.ProtocolVersion; -import org.apache.http.client.HttpClient; -import org.apache.http.client.methods.HttpGet; -import org.apache.http.entity.StringEntity; -import org.apache.http.message.BasicHttpResponse; -import org.junit.Before; -import org.junit.Test; -import org.mockito.ArgumentCaptor; - -public class PnfEventReadyConsumerTest { - - private static final String CORRELATION_ID = "corrTestId"; - private static final String CORRELATION_ID_NOT_FOUND_IN_MAP = "otherCorrId"; - private static final String JSON_EXAMPLE_WITH_CORRELATION_ID = - "{\"pnfRegistrationFields\":{\"correlationId\":\"%s\"}}"; - private static final String JSON_EXAMPLE_WITH_NO_CORRELATION_ID = - "{\"pnfRegistrationFields\":{\"field\":\"value\"}}"; - - private static final String HOST = "hostTest"; - private static final int PORT = 1234; - private static final String PROTOCOL = "http"; - private static final String URI_PATH_PREFIX = "eventsForTesting"; - private static final String EVENT_TOPIC_TEST = "eventTopicTest"; - private static final String CONSUMER_ID = "consumerTestId"; - private static final String CONSUMER_GROUP = "consumerGroupTest"; - - private PnfEventReadyConsumer testedObject; - private HttpClient httpClientMock; - private Runnable threadMockToNotifyCamundaFlow; - private ScheduledExecutorService executorMock; - - @Before - public void init() throws NoSuchFieldException, IllegalAccessException { - testedObject = new PnfEventReadyConsumer(); - testedObject.setDmaapHost(HOST); - testedObject.setDmaapPort(PORT); - testedObject.setDmaapProtocol(PROTOCOL); - testedObject.setDmaapUriPathPrefix(URI_PATH_PREFIX); - testedObject.setDmaapTopicName(EVENT_TOPIC_TEST); - testedObject.setConsumerId(CONSUMER_ID); - testedObject.setConsumerGroup(CONSUMER_GROUP); - testedObject.setDmaapClientDelayInSeconds(1); - testedObject.init(); - httpClientMock = mock(HttpClient.class); - threadMockToNotifyCamundaFlow = mock(Runnable.class); - executorMock = mock(ScheduledExecutorService.class); - setPrivateField(); - } - - /** - * Test run method, where the are following conditions: - *

- DmaapThreadListener is running, flag is set to true - *

- map is filled with one entry with the key that we get from response - *

run method should invoke thread from map to notify camunda process, remove element from the map (map is empty) - * and shutdown the executor because of empty map - */ - @Test - public void correlationIdIsFoundInHttpResponse_notifyAboutPnfReady() - throws IOException { - when(httpClientMock.execute(any(HttpGet.class))). - thenReturn(createResponse(String.format(JSON_EXAMPLE_WITH_CORRELATION_ID, CORRELATION_ID))); - testedObject.sendRequest(); - ArgumentCaptor captor1 = ArgumentCaptor.forClass(HttpGet.class); - verify(httpClientMock).execute(captor1.capture()); - assertThat(captor1.getValue().getURI()).hasHost(HOST).hasPort(PORT).hasScheme(PROTOCOL) - .hasPath( - "/" + URI_PATH_PREFIX + "/" + EVENT_TOPIC_TEST + "/" + CONSUMER_GROUP + "/" + CONSUMER_ID + ""); - verify(threadMockToNotifyCamundaFlow).run(); - verify(executorMock).shutdownNow(); - } - - /** - * Test run method, where the are following conditions: - *

- DmaapThreadListener is running, flag is set to true - *

- map is filled with one entry with the correlationId that does not match to correlationId - * taken from http response. run method should not do anything with the map not run any thread to - * notify camunda process - */ - @Test - public void correlationIdIsFoundInHttpResponse_NotFoundInMap() - throws IOException { - when(httpClientMock.execute(any(HttpGet.class))). - thenReturn(createResponse( - String.format(JSON_EXAMPLE_WITH_CORRELATION_ID, CORRELATION_ID_NOT_FOUND_IN_MAP))); - testedObject.sendRequest(); - verifyZeroInteractions(threadMockToNotifyCamundaFlow, executorMock); - } - - /** - * Test run method, where the are following conditions: - *

- DmaapThreadListener is running, flag is set to true - *

- map is filled with one entry with the correlationId but no correlation id is taken from HttpResponse - * run method should not do anything with the map and not run any thread to notify camunda process - */ - @Test - public void correlationIdIsNotFoundInHttpResponse() throws IOException { - when(httpClientMock.execute(any(HttpGet.class))). - thenReturn(createResponse(JSON_EXAMPLE_WITH_NO_CORRELATION_ID)); - testedObject.sendRequest(); - verifyZeroInteractions(threadMockToNotifyCamundaFlow, executorMock); - } - - private void setPrivateField() throws NoSuchFieldException, IllegalAccessException { - Field httpClientField = testedObject.getClass().getDeclaredField("httpClient"); - httpClientField.setAccessible(true); - httpClientField.set(testedObject, httpClientMock); - httpClientField.setAccessible(false); - - Field executorField = testedObject.getClass().getDeclaredField("executor"); - executorField.setAccessible(true); - executorField.set(testedObject, executorMock); - executorField.setAccessible(false); - - Field pnfCorrelationToThreadMapField = testedObject.getClass() - .getDeclaredField("pnfCorrelationIdToThreadMap"); - pnfCorrelationToThreadMapField.setAccessible(true); - Map pnfCorrelationToThreadMap = new ConcurrentHashMap<>(); - pnfCorrelationToThreadMap.put(CORRELATION_ID, threadMockToNotifyCamundaFlow); - pnfCorrelationToThreadMapField.set(testedObject, pnfCorrelationToThreadMap); - - Field threadRunFlag = testedObject.getClass().getDeclaredField("dmaapThreadListenerIsRunning"); - threadRunFlag.setAccessible(true); - threadRunFlag.set(testedObject, true); - threadRunFlag.setAccessible(false); - } - - private HttpResponse createResponse(String json) throws UnsupportedEncodingException { - HttpEntity entity = new StringEntity(json); - ProtocolVersion protocolVersion = new ProtocolVersion("", 1, 1); - HttpResponse response = new BasicHttpResponse(protocolVersion, 1, ""); - response.setEntity(entity); - response.setStatusCode(200); - return response; - } - -} diff --git a/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java b/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java new file mode 100644 index 0000000000..393730ed43 --- /dev/null +++ b/bpmn/MSOInfrastructureBPMN/src/test/java/org/openecomp/mso/bpmn/infrastructure/pnf/dmaap/PnfEventReadyDmaapClientTest.java @@ -0,0 +1,176 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - SO + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.mso.bpmn.infrastructure.pnf.dmaap; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.lang.reflect.Field; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import org.apache.http.HttpEntity; +import org.apache.http.HttpResponse; +import org.apache.http.ProtocolVersion; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.entity.StringEntity; +import org.apache.http.message.BasicHttpResponse; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.openecomp.mso.bpmn.infrastructure.pnf.dmaap.PnfEventReadyDmaapClient.DmaapTopicListenerThread; + +public class PnfEventReadyDmaapClientTest { + + private static final String CORRELATION_ID = "corrTestId"; + private static final String CORRELATION_ID_NOT_FOUND_IN_MAP = "otherCorrId"; + private static final String JSON_EXAMPLE_WITH_CORRELATION_ID = + "{\"pnfRegistrationFields\":{\"correlationId\":\"%s\"}}"; + private static final String JSON_EXAMPLE_WITH_NO_CORRELATION_ID = + "{\"pnfRegistrationFields\":{\"field\":\"value\"}}"; + + private static final String HOST = "hostTest"; + private static final int PORT = 1234; + private static final String PROTOCOL = "http"; + private static final String URI_PATH_PREFIX = "eventsForTesting"; + private static final String EVENT_TOPIC_TEST = "eventTopicTest"; + private static final String CONSUMER_ID = "consumerTestId"; + private static final String CONSUMER_GROUP = "consumerGroupTest"; + + private PnfEventReadyDmaapClient testedObject; + private DmaapTopicListenerThread testedObjectInnerClassThread; + private HttpClient httpClientMock; + private Runnable threadMockToNotifyCamundaFlow; + private ScheduledExecutorService executorMock; + + @Before + public void init() throws NoSuchFieldException, IllegalAccessException { + testedObject = new PnfEventReadyDmaapClient(); + testedObject.setDmaapHost(HOST); + testedObject.setDmaapPort(PORT); + testedObject.setDmaapProtocol(PROTOCOL); + testedObject.setDmaapUriPathPrefix(URI_PATH_PREFIX); + testedObject.setDmaapTopicName(EVENT_TOPIC_TEST); + testedObject.setConsumerId(CONSUMER_ID); + testedObject.setConsumerGroup(CONSUMER_GROUP); + testedObject.setDmaapClientDelayInSeconds(1); + testedObject.init(); + testedObjectInnerClassThread = testedObject.new DmaapTopicListenerThread(); + httpClientMock = mock(HttpClient.class); + threadMockToNotifyCamundaFlow = mock(Runnable.class); + executorMock = mock(ScheduledExecutorService.class); + setPrivateField(); + } + + /** + * Test run method, where the are following conditions: + *

- DmaapThreadListener is running, flag is set to true + *

- map is filled with one entry with the key that we get from response + *

run method should invoke thread from map to notify camunda process, remove element from the map (map is empty) + * and shutdown the executor because of empty map + */ + @Test + public void correlationIdIsFoundInHttpResponse_notifyAboutPnfReady() + throws IOException { + when(httpClientMock.execute(any(HttpGet.class))). + thenReturn(createResponse(String.format(JSON_EXAMPLE_WITH_CORRELATION_ID, CORRELATION_ID))); + testedObjectInnerClassThread.run(); + ArgumentCaptor captor1 = ArgumentCaptor.forClass(HttpGet.class); + verify(httpClientMock).execute(captor1.capture()); + assertThat(captor1.getValue().getURI()).hasHost(HOST).hasPort(PORT).hasScheme(PROTOCOL) + .hasPath( + "/" + URI_PATH_PREFIX + "/" + EVENT_TOPIC_TEST + "/" + CONSUMER_GROUP + "/" + CONSUMER_ID + ""); + verify(threadMockToNotifyCamundaFlow).run(); + verify(executorMock).shutdownNow(); + } + + /** + * Test run method, where the are following conditions: + *

- DmaapThreadListener is running, flag is set to true + *

- map is filled with one entry with the correlationId that does not match to correlationId + * taken from http response. run method should not do anything with the map not run any thread to + * notify camunda process + */ + @Test + public void correlationIdIsFoundInHttpResponse_NotFoundInMap() + throws IOException { + when(httpClientMock.execute(any(HttpGet.class))). + thenReturn(createResponse( + String.format(JSON_EXAMPLE_WITH_CORRELATION_ID, CORRELATION_ID_NOT_FOUND_IN_MAP))); + testedObjectInnerClassThread.run(); + verifyZeroInteractions(threadMockToNotifyCamundaFlow, executorMock); + } + + /** + * Test run method, where the are following conditions: + *

- DmaapThreadListener is running, flag is set to true + *

- map is filled with one entry with the correlationId but no correlation id is taken from HttpResponse + * run method should not do anything with the map and not run any thread to notify camunda process + */ + @Test + public void correlationIdIsNotFoundInHttpResponse() throws IOException { + when(httpClientMock.execute(any(HttpGet.class))). + thenReturn(createResponse(JSON_EXAMPLE_WITH_NO_CORRELATION_ID)); + testedObjectInnerClassThread.run(); + verifyZeroInteractions(threadMockToNotifyCamundaFlow, executorMock); + } + + private void setPrivateField() throws NoSuchFieldException, IllegalAccessException { + Field httpClientField = testedObject.getClass().getDeclaredField("httpClient"); + httpClientField.setAccessible(true); + httpClientField.set(testedObject, httpClientMock); + httpClientField.setAccessible(false); + + Field executorField = testedObject.getClass().getDeclaredField("executor"); + executorField.setAccessible(true); + executorField.set(testedObject, executorMock); + executorField.setAccessible(false); + + Field pnfCorrelationToThreadMapField = testedObject.getClass() + .getDeclaredField("pnfCorrelationIdToThreadMap"); + pnfCorrelationToThreadMapField.setAccessible(true); + Map pnfCorrelationToThreadMap = new ConcurrentHashMap<>(); + pnfCorrelationToThreadMap.put(CORRELATION_ID, threadMockToNotifyCamundaFlow); + pnfCorrelationToThreadMapField.set(testedObject, pnfCorrelationToThreadMap); + + Field threadRunFlag = testedObject.getClass().getDeclaredField("dmaapThreadListenerIsRunning"); + threadRunFlag.setAccessible(true); + threadRunFlag.set(testedObject, true); + threadRunFlag.setAccessible(false); + } + + private HttpResponse createResponse(String json) throws UnsupportedEncodingException { + HttpEntity entity = new StringEntity(json); + ProtocolVersion protocolVersion = new ProtocolVersion("", 1, 1); + HttpResponse response = new BasicHttpResponse(protocolVersion, 1, ""); + response.setEntity(entity); + response.setStatusCode(200); + return response; + } + +} -- cgit 1.2.3-korg