From e540d3894d2f662a843e5c51535d44b876b9889f Mon Sep 17 00:00:00 2001 From: Shiwei Tian Date: Sat, 14 Oct 2017 10:45:09 +0800 Subject: modify not publish messages to DMaaP Issue-ID: HOLMES-71 Change-Id: Iaf87f0250d043aae84e9afc6aaec0f6cdc0a529c Signed-off-by: Shiwei Tian --- .../org/onap/holmes/engine/EngineDActiveApp.java | 9 ++++ .../dcaepolling/DcaeConfigurationPolling.java | 63 ++++++++++++++++++++++ .../engine/dmaappolling/SubscriberAction.java | 14 ++--- .../holmes/engine/request/DeployRuleRequest.java | 3 ++ .../holmes/engine/resources/EngineResources.java | 4 +- 5 files changed, 86 insertions(+), 7 deletions(-) create mode 100644 engine-d/src/main/java/org/onap/holmes/engine/dcaepolling/DcaeConfigurationPolling.java (limited to 'engine-d/src/main/java/org') diff --git a/engine-d/src/main/java/org/onap/holmes/engine/EngineDActiveApp.java b/engine-d/src/main/java/org/onap/holmes/engine/EngineDActiveApp.java index 5bb3dbe..69e4c7a 100644 --- a/engine-d/src/main/java/org/onap/holmes/engine/EngineDActiveApp.java +++ b/engine-d/src/main/java/org/onap/holmes/engine/EngineDActiveApp.java @@ -20,11 +20,15 @@ import static jdk.nashorn.internal.runtime.regexp.joni.Config.log; import io.dropwizard.setup.Environment; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; import org.onap.holmes.common.config.MicroServiceConfig; import org.onap.holmes.common.dropwizard.ioc.bundle.IOCApplication; import org.onap.holmes.common.exception.CorrelationException; import org.onap.holmes.common.utils.MSBRegisterUtil; +import org.onap.holmes.engine.dcaepolling.DcaeConfigurationPolling; import org.onap.holmes.engine.resources.EngineResources; import org.onap.msb.sdk.discovery.entity.MicroServiceInfo; import org.onap.msb.sdk.discovery.entity.Node; @@ -40,6 +44,11 @@ public class EngineDActiveApp extends IOCApplication { super.run(configuration, environment); environment.jersey().register(new EngineResources()); + + ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); + service.scheduleAtFixedRate(new DcaeConfigurationPolling("holmes-rule-mgmt"), 0, + DcaeConfigurationPolling.POLLING_PERIOD, TimeUnit.MILLISECONDS); + try { new MSBRegisterUtil().register2Msb(createMicroServiceInfo()); } catch (CorrelationException e) { diff --git a/engine-d/src/main/java/org/onap/holmes/engine/dcaepolling/DcaeConfigurationPolling.java b/engine-d/src/main/java/org/onap/holmes/engine/dcaepolling/DcaeConfigurationPolling.java new file mode 100644 index 0000000..23030d2 --- /dev/null +++ b/engine-d/src/main/java/org/onap/holmes/engine/dcaepolling/DcaeConfigurationPolling.java @@ -0,0 +1,63 @@ +/** + * Copyright 2017 ZTE Corporation. + * + * 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. + */ +package org.onap.holmes.engine.dcaepolling; + +import lombok.extern.slf4j.Slf4j; +import org.onap.holmes.common.dcae.DcaeConfigurationQuery; +import org.onap.holmes.common.dcae.DcaeConfigurationsCache; +import org.onap.holmes.common.dcae.entity.DcaeConfigurations; +import org.onap.holmes.common.dropwizard.ioc.utils.ServiceLocatorHolder; +import org.onap.holmes.common.exception.CorrelationException; +import org.onap.holmes.dsa.dmaappolling.Subscriber; +import org.onap.holmes.engine.dmaappolling.SubscriberAction; + +@Slf4j +public class DcaeConfigurationPolling implements Runnable{ + + private String hostname; + + private String subscriberKey = "sec_fault_unsecure"; + + public static long POLLING_PERIOD = 10 * 1000L; + + public DcaeConfigurationPolling(String hostname) { + this.hostname = hostname; + } + + @Override + public void run() { + DcaeConfigurations dcaeConfigurations = null; + try { + dcaeConfigurations = DcaeConfigurationQuery + .getDcaeConfigurations(hostname); + } catch (CorrelationException e) { + log.error("Failed to polling dcae configurations" + e.getMessage()); + } + if (dcaeConfigurations != null) { + DcaeConfigurationsCache.setDcaeConfigurations(dcaeConfigurations); + addSubscriber(dcaeConfigurations); + } + } + + private void addSubscriber(DcaeConfigurations dcaeConfigurations) { + SubscriberAction subscriberAction = ServiceLocatorHolder.getLocator() + .getService(SubscriberAction.class); + Subscriber subscriber = new Subscriber(); + subscriber.setUrl(dcaeConfigurations.getSubSecInfo(subscriberKey).getDmaapInfo() + .getTopicUrl()); + subscriberAction.addSubscriber(subscriber); + } +} diff --git a/engine-d/src/main/java/org/onap/holmes/engine/dmaappolling/SubscriberAction.java b/engine-d/src/main/java/org/onap/holmes/engine/dmaappolling/SubscriberAction.java index 1e71899..ef585d5 100644 --- a/engine-d/src/main/java/org/onap/holmes/engine/dmaappolling/SubscriberAction.java +++ b/engine-d/src/main/java/org/onap/holmes/engine/dmaappolling/SubscriberAction.java @@ -35,17 +35,19 @@ public class SubscriberAction { private ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(); public void addSubscriber(Subscriber subscriber) { - DMaaPPollingRequest pollingTask = new DMaaPPollingRequest(subscriber, droolsEngine); - ScheduledFuture future = service - .scheduleAtFixedRate(pollingTask, 0, subscriber.getPeriod(), TimeUnit.MILLISECONDS); - pollingRequests.put(subscriber.getTopic(), future); + if (!pollingRequests.containsKey(subscriber.getUrl())) { + DMaaPPollingRequest pollingTask = new DMaaPPollingRequest(subscriber, droolsEngine); + ScheduledFuture future = service + .scheduleAtFixedRate(pollingTask, 0, subscriber.getPeriod(), TimeUnit.MILLISECONDS); + pollingRequests.put(subscriber.getUrl(), future); + } } public void removeSubscriber(Subscriber subscriber) { - ScheduledFuture future = pollingRequests.get(subscriber.getTopic()); + ScheduledFuture future = pollingRequests.get(subscriber.getUrl()); if (future != null) { future.cancel(true); } - pollingRequests.remove(subscriber.getTopic()); + pollingRequests.remove(subscriber.getUrl()); } } diff --git a/engine-d/src/main/java/org/onap/holmes/engine/request/DeployRuleRequest.java b/engine-d/src/main/java/org/onap/holmes/engine/request/DeployRuleRequest.java index de9d773..5b8951b 100644 --- a/engine-d/src/main/java/org/onap/holmes/engine/request/DeployRuleRequest.java +++ b/engine-d/src/main/java/org/onap/holmes/engine/request/DeployRuleRequest.java @@ -30,4 +30,7 @@ public class DeployRuleRequest { @JsonProperty(value = "engineid") private String engineId; + + @JsonProperty(value = "loopcontrolname") + private String loopControlName; } diff --git a/engine-d/src/main/java/org/onap/holmes/engine/resources/EngineResources.java b/engine-d/src/main/java/org/onap/holmes/engine/resources/EngineResources.java index 8f9a271..45754e2 100644 --- a/engine-d/src/main/java/org/onap/holmes/engine/resources/EngineResources.java +++ b/engine-d/src/main/java/org/onap/holmes/engine/resources/EngineResources.java @@ -33,6 +33,7 @@ import javax.ws.rs.core.Context; import javax.ws.rs.core.MediaType; import lombok.extern.slf4j.Slf4j; import org.jvnet.hk2.annotations.Service; +import org.onap.holmes.common.dmaap.DmaapService; import org.onap.holmes.common.exception.CorrelationException; import org.onap.holmes.common.utils.ExceptionUtil; import org.onap.holmes.common.utils.LanguageUtil; @@ -65,8 +66,9 @@ public class EngineResources { CorrelationRuleResponse crResponse = new CorrelationRuleResponse(); Locale locale = LanguageUtil.getLocale(httpRequest); try { - String packageName = droolsEngine.deployRule(deployRuleRequest, locale); + DmaapService.loopControlNames + .put(packageName, deployRuleRequest.getLoopControlName()); log.info("Rule deployed. Package name: " + packageName); crResponse.setPackageName(packageName); -- cgit 1.2.3-korg