NIFI-1442: This closes #306. Use CircularFifoQueue instead of Set to store nodes' bulletins

Joint effort by Toivo Adams from PR306 and and Mark Payne

Signed-off-by: joewitt <joewitt@apache.org>

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java

@@ -17,50 +17,26 @@
 package org.apache.nifi.events;
 
 import java.util.HashSet;
-import java.util.LinkedHashSet;
 import java.util.Set;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
+import org.apache.commons.collections4.queue.CircularFifoQueue;
 import org.apache.nifi.reporting.Bulletin;
 
 /**
  *
  */
 public class NodeBulletinProcessingStrategy implements BulletinProcessingStrategy {
-
+    static final int MAX_ENTRIES = 5;
-    private final Lock lock;
+    private final CircularFifoQueue<Bulletin> ringBuffer = new CircularFifoQueue<>(MAX_ENTRIES);
-    private final Set<Bulletin> bulletins;
-
-    public NodeBulletinProcessingStrategy() {
-        lock = new ReentrantLock();
-        bulletins = new LinkedHashSet<>();
-    }
 
     @Override
-    public void update(final Bulletin bulletin) {
+    public synchronized void update(final Bulletin bulletin) {
-        lock.lock();
+        ringBuffer.add(bulletin);
-        try {
-            bulletins.add(bulletin);
-        } finally {
-            lock.unlock();
-        }
     }
 
-    public Set<Bulletin> getBulletins() {
+    public synchronized Set<Bulletin> getBulletins() {
-        final Set<Bulletin> response = new HashSet<>();
+        final Set<Bulletin> response = new HashSet<>(ringBuffer);
-
+        ringBuffer.clear();
-        lock.lock();
-        try {
-            // get all the bulletins currently stored
-            response.addAll(bulletins);
-
-            // remove the bulletins
-            bulletins.clear();
-        } finally {
-            lock.unlock();
-        }
-
         return response;
     }
 }

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/events/TestNodeBulletinProcessingStrategy.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.nifi.events;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+public class TestNodeBulletinProcessingStrategy {
+
+    @Test
+    public void testUpdate() {
+
+        NodeBulletinProcessingStrategy nBulletinProcessingStrategy = new NodeBulletinProcessingStrategy();
+
+        nBulletinProcessingStrategy.update(new ComponentBulletin(1));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(2));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(3));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(4));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(5));
+        assertEquals(5, nBulletinProcessingStrategy.getBulletins().size());
+
+        nBulletinProcessingStrategy.update(new ComponentBulletin(1));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(2));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(3));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(4));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(5));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(6));
+        nBulletinProcessingStrategy.update(new ComponentBulletin(7));
+        assertEquals(NodeBulletinProcessingStrategy.MAX_ENTRIES, nBulletinProcessingStrategy.getBulletins().size());
+
+    }
+
+}