The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
+## [1.5.2] - 16/03/2021
+
+- Implement singleton cache for events being processed
+- Add JUnit tests
+
## [1.5.1] - 03/02/2021
- Add configuration number of threads and limit rate in files processing
-- Fix vulnerability - update undertow from 2.0.3.Final to 2.2.3.Final and commons.io from 2.6.0 to 2.8.0
\ No newline at end of file
+- Fix vulnerability - update undertow from 2.0.3.Final to 2.2.3.Final and commons.io from 2.6.0 to 2.8.0
<groupId>org.onap.dcaegen2.services</groupId>
<artifactId>pm-mapper</artifactId>
- <version>1.5.1-SNAPSHOT</version>
+ <version>1.5.2-SNAPSHOT</version>
<name>dcaegen2-services-pm-mapper</name>
import org.onap.dcaegen2.services.pmmapper.model.ServerResource;
import org.onap.dcaegen2.services.pmmapper.ssl.SSLContextFactory;
import org.onap.dcaegen2.services.pmmapper.utils.DataRouterUtils;
+import org.onap.dcaegen2.services.pmmapper.utils.IncomingEventsCache;
import org.onap.dcaegen2.services.pmmapper.utils.MeasConverter;
import org.onap.dcaegen2.services.pmmapper.utils.MeasSplitter;
import org.onap.dcaegen2.services.pmmapper.utils.XMLValidator;
private static final int HTTPS_PORT = 8443;
private static final int INITIAL_RECONFIGURATION_PERIOD = 60;
private static final int RECONFIGURATION_PERIOD = 60;
+ private static final IncomingEventsCache eventsCache = IncomingEventsCache.INSTANCE;
private static Path templates = Paths.get("/opt/app/pm-mapper/etc/templates/");
private static Path schemas = Paths.get("/opt/app/pm-mapper/etc/schemas/");
this.flux.onBackpressureDrop(App::handleBackPressure)
.doOnNext(App::receiveRequest)
+ .filter(event -> !isCached(event.getPublishIdentity()))
+ .doOnNext(App::addToCache)
.limitRate(processingConfig.getLimitRate())
.parallel(processingThreads)
.runOn(Schedulers.newParallel("Thread", processingThreads), PREFETCH_ONE_PER_THREAD)
}
}
+ private boolean isCached(String id) {
+ boolean isPresent = eventsCache.contains(id);
+ if(isPresent) {
+ logger.unwrap().info("Skipping. This event is already waiting in cache to be processed: " + id);
+ }
+ return isPresent;
+ }
+
public static void main(String[] args) throws EnvironmentConfigException {
FilesProcessingConfig processingConfig = new FilesProcessingConfig(new EnvironmentReader());
new App(templates, schemas, HTTP_PORT, HTTPS_PORT, new ConfigHandler(), processingConfig).start();
public static void sendEventProcessed(MapperConfig config, Event event) {
try {
DataRouterUtils.processEvent(config, event);
+ eventsCache.remove(event.getPublishIdentity());
} catch (ProcessEventException exception) {
logger.unwrap().error("Process event failure", exception);
}
.unDispatch();
}
+ /**
+ * Adds received event to cache
+ * @param event to be cached.
+ */
+ public static void addToCache(@NonNull Event event) {
+ eventsCache.add(event.getPublishIdentity());
+ }
+
/**
* Takes the exchange from an event, responds with a 200 and un-dispatches the exchange.
* @param event to be received.
/*-
* ============LICENSE_START=======================================================
- * Copyright (C) 2020 Nordix Foundation.
+ * Copyright (C) 2020 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2021 Nokia.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcaegen2.services.pmmapper.utils;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public enum IncomingEventsCache {
+ INSTANCE;
+
+ private Set<String> eventsIds = ConcurrentHashMap.newKeySet();
+
+ /**
+ * Gets thread safe, single instance of this enum
+ * @return the single instance of cache
+ */
+ public IncomingEventsCache getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Adds publishIdentity of an event to cache
+ * @param id to be added to cache
+ */
+ public void add(String id) {
+ eventsIds.add(id);
+ }
+
+ /**
+ * Remove publishIdentity of an event from cache
+ * @param id to be removed from cache
+ */
+ public void remove(String id) {
+ eventsIds.remove(id);
+ }
+
+ /**
+ * Checks if the cache contains a publishIdentity
+ * @param id to be found in cache
+ * @return true when the id exists, false otherwise
+ */
+ public boolean contains(String id) {
+ return eventsIds.contains(id);
+ }
+
+ Set<String> getCachedEvents() {
+ return new HashSet<>(eventsIds);
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2021 Nokia.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcaegen2.services.pmmapper.utils;
+
+import org.junit.After;
+import org.junit.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class IncomingEventsCacheTest {
+
+ private static final String id1 = "123.dmaap-dr-node";
+ private static final String id2 = "987.dmaap-dr-node";
+
+ @After
+ public void resetCache() {
+ IncomingEventsCache cache = IncomingEventsCache.INSTANCE;
+ for(String id: cache.getCachedEvents()) {
+ cache.remove(id);
+ }
+ }
+
+ @Test
+ public void shouldContainEventAfterAddingItToCache() {
+ IncomingEventsCache eventsCache = IncomingEventsCache.INSTANCE;
+ eventsCache.add(id1);
+ assertEquals(1, IncomingEventsCache.INSTANCE.getCachedEvents().size());
+ assertTrue(IncomingEventsCache.INSTANCE.getCachedEvents().contains(id1));
+ }
+
+ @Test
+ public void shouldRemoveEventFromCache() {
+ IncomingEventsCache eventsCache = IncomingEventsCache.INSTANCE;
+
+ assertEquals(0, IncomingEventsCache.INSTANCE.getCachedEvents().size());
+ eventsCache.add(id1);
+ eventsCache.add(id1);
+ eventsCache.add(id2);
+ assertEquals(2, IncomingEventsCache.INSTANCE.getCachedEvents().size());
+ eventsCache.remove(id1);
+ assertEquals(1, IncomingEventsCache.INSTANCE.getCachedEvents().size());
+ assertTrue(IncomingEventsCache.INSTANCE.getCachedEvents().contains(id2));
+ }
+}
major=1
minor=5
-patch=1
+patch=2
base_version=${major}.${minor}.${patch}
release_version=${base_version}
snapshot_version=${base_version}-SNAPSHOT
Code Review / dcaegen2 / services / pm-mapper.git / commitdiff