* limitations under the License.
* ============LICENSE_END=========================================================
*/
-
package org.onap.so.bpmn.infrastructure.pnf.dmaap;
import java.io.IOException;
-import java.util.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
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.onap.so.client.kafka.KafkaConsumerImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;
-import org.onap.so.client.aai.entities.uri.AAIResourceUri;
-import org.onap.so.client.aai.entities.uri.AAIUriFactory;
-import org.onap.so.client.aai.AAIResourcesClient;
-import org.onap.so.client.aai.AAIObjectType;
@Component
public class PnfEventReadyDmaapClient implements DmaapClient {
-
private static final Logger logger = LoggerFactory.getLogger(PnfEventReadyDmaapClient.class);
-
- private HttpClient httpClient;
private Map<String, Runnable> pnfCorrelationIdToThreadMap;
- private HttpGet getRequest;
private int topicListenerDelayInSeconds;
private volatile ScheduledThreadPoolExecutor executor;
private volatile boolean dmaapThreadListenerIsRunning;
+ private KafkaConsumerImpl consumerForPnfReady;
+ private KafkaConsumerImpl consumerForPnfUpdate;
+ private String pnfReadyTopic;
+ private String pnfUpdateTopic;
+ private String consumerGroup;
+ private String consumerId;
+ private String consumerIdUpdate;
+
- public volatile List<HashMap<String, String>> updateInfoMap;
@Autowired
- public PnfEventReadyDmaapClient(Environment env) {
- httpClient = HttpClientBuilder.create().build();
+ public PnfEventReadyDmaapClient(Environment env) throws IOException {
pnfCorrelationIdToThreadMap = new ConcurrentHashMap<>();
topicListenerDelayInSeconds = env.getProperty("pnf.dmaap.topicListenerDelayInSeconds", Integer.class);
executor = null;
- getRequest = new HttpGet(UriBuilder.fromUri(env.getProperty("pnf.dmaap.uriPathPrefix"))
- .scheme(env.getProperty("pnf.dmaap.protocol")).host(env.getProperty("pnf.dmaap.host"))
- .port(env.getProperty("pnf.dmaap.port", Integer.class)).path(env.getProperty("pnf.dmaap.topicName"))
- .path(env.getProperty("pnf.dmaap.consumerGroup")).path(env.getProperty("pnf.dmaap.consumerId"))
- .build());
- updateInfoMap = new ArrayList<>();
+ try {
+ consumerForPnfReady = new KafkaConsumerImpl(env.getProperty("pnf.kafka.kafkaBootstrapServers"));
+ consumerForPnfUpdate = new KafkaConsumerImpl(env.getProperty("pnf.kafka.kafkaBootstrapServers"));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ pnfReadyTopic = env.getProperty("pnf.kafka.pnfReadyTopicName");
+ pnfUpdateTopic = env.getProperty("pnf.kafka.pnfUpdateTopicName");
+ consumerGroup = env.getProperty("pnf.kafka.consumerGroup");
+ consumerId = env.getProperty("pnf.kafka.consumerId");
+ consumerIdUpdate = env.getProperty("pnf.kafka.consumerIdUpdate");
}
+
@Override
- public synchronized void registerForUpdate(String pnfCorrelationId, Runnable informConsumer,
- Optional<HashMap<String, String>> updateInfo) {
- logger.debug("registering for pnf ready dmaap event for pnf correlation id: {}", pnfCorrelationId);
- HashMap<String, String> map = updateInfo.get();
- if (map != null && map.size() > 0) {
- synchronized (updateInfoMap) {
- updateInfoMap.add(map);
- }
- }
+ public synchronized void registerForUpdate(String pnfCorrelationId, Runnable informConsumer) {
+ logger.debug("registering for pnf ready kafka event for pnf correlation id: {}", pnfCorrelationId);
pnfCorrelationIdToThreadMap.put(pnfCorrelationId, informConsumer);
if (!dmaapThreadListenerIsRunning) {
startDmaapThreadListener();
@Override
public synchronized Runnable unregister(String pnfCorrelationId) {
- logger.debug("unregistering from pnf ready dmaap event for pnf correlation id: {}", pnfCorrelationId);
+ logger.debug("unregistering from pnf ready kafka event for pnf correlation id: {}", pnfCorrelationId);
Runnable runnable = pnfCorrelationIdToThreadMap.remove(pnfCorrelationId);
- synchronized (updateInfoMap) {
- for (int i = updateInfoMap.size() - 1; i >= 0; i--) {
- if (!updateInfoMap.get(i).containsKey("pnfCorrelationId"))
- continue;
- String id = updateInfoMap.get(i).get("pnfCorrelationId");
- if (id != pnfCorrelationId)
- continue;
- updateInfoMap.remove(i);
- }
- }
if (pnfCorrelationIdToThreadMap.isEmpty()) {
+ consumerForPnfUpdate.close();
+ consumerForPnfReady.close();
stopDmaapThreadListener();
}
return runnable;
}
class DmaapTopicListenerThread implements Runnable {
-
@Override
public void run() {
try {
- logger.debug("dmaap listener starts listening pnf ready dmaap topic");
- HttpResponse response = httpClient.execute(getRequest);
- List<String> idList = getPnfCorrelationIdListFromResponse(response);
-
- // idList is never null
- if (!idList.isEmpty()) {
- // send only body of response
- registerClientResponse(idList.get(0), EntityUtils.toString(response.getEntity(), "UTF-8"));
- }
-
- if (idList != null) {
- idList.forEach(this::informAboutPnfReadyIfPnfCorrelationIdFound);
+ List<String> response;
+ System.out.println(pnfUpdateTopic + " " + consumerGroup);
+ response = consumerForPnfUpdate.get(pnfUpdateTopic, consumerGroup, consumerIdUpdate);
+ if (response.isEmpty()) {
+ response = consumerForPnfReady.get(pnfReadyTopic, consumerGroup, consumerId);
+ getPnfCorrelationIdListFromResponse(response)
+ .forEach(this::informAboutPnfReadyIfPnfCorrelationIdFound);
+ } else {
+ getPnfCorrelationIdListFromResponse(response)
+ .forEach(this::informAboutPnfReadyIfPnfCorrelationIdFound);
}
} catch (IOException e) {
- logger.error("Exception caught during sending rest request to dmaap for listening event topic", e);
- } finally {
- getRequest.reset();
+ logger.error("Exception caught during sending rest request to kafka for listening event topic", e);
}
}
- private List<String> getPnfCorrelationIdListFromResponse(HttpResponse response) throws IOException {
- if (response.getStatusLine().getStatusCode() == 200) {
- String responseString = EntityUtils.toString(response.getEntity(), "UTF-8");
- if (responseString != null) {
- return JsonUtilForPnfCorrelationId.parseJsonToGelAllPnfCorrelationId(responseString);
- }
+ private List<String> getPnfCorrelationIdListFromResponse(List<String> response) throws IOException {
+ if (response != null) {
+ return JsonUtilForPnfCorrelationId.parseJsonToGelAllPnfCorrelationId(response);
}
return Collections.emptyList();
}
runnable.run();
}
}
-
- private void registerClientResponse(String pnfCorrelationId, String response) {
-
- String customerId = null;
- String serviceType = null;
- String serId = null;
- synchronized (updateInfoMap) {
- for (HashMap<String, String> map : updateInfoMap) {
- if (!map.containsKey("pnfCorrelationId"))
- continue;
- if (pnfCorrelationId != map.get("pnfCorrelationId"))
- continue;
- if (!map.containsKey("globalSubscriberID"))
- continue;
- if (!map.containsKey("serviceType"))
- continue;
- if (!map.containsKey("serviceInstanceId"))
- continue;
- customerId = map.get("pnfCorrelationId");
- serviceType = map.get("serviceType");
- serId = map.get("serviceInstanceId");
- }
- }
- if (customerId == null || serviceType == null || serId == null)
- return;
- AAIResourcesClient client = new AAIResourcesClient();
- AAIResourceUri uri = AAIUriFactory.createResourceUri(AAIObjectType.SERVICE_INSTANCE_METADATA, customerId,
- serviceType, serId);
- client.update(uri, response);
- }
-
}
}
+