https://issues.apache.org/jira/browse/AMQ-4094 - ensure list is split down to single entry before page over flow for a value, ensures the smallest marshall size for a page

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1399300 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubsOfflineSelectorIndexUseTest.java

@@ -167,6 +167,12 @@ public class DurableSubsOfflineSelectorIndexUseTest extends org.apache.activemq.
         assertEquals(messageCount / 2, listenerT.count);
         assertEquals(messageCount / 2, listenerF.count);
 
+        consumerTrue.close();
+        session.unsubscribe("true");
+
+        consumerFalse.close();
+        session.unsubscribe("false");
+
         session.close();
         con.close();
 
@@ -195,14 +201,6 @@ public class DurableSubsOfflineSelectorIndexUseTest extends org.apache.activemq.
 
     }
 
-    private void dumpstats() throws Exception {
-        //TimeUnit.SECONDS.sleep(2);
-        //ThreadTracker.result();
-        TimeUnit.SECONDS.sleep(4);
-
-
-    }
-
     public static class Listener implements MessageListener {
         int count = 0;
         String id = null;

kahadb/src/main/java/org/apache/kahadb/index/ListNode.java

@@ -309,19 +309,14 @@ public final class ListNode<Key, Value> {
     }
 
     public void storeUpdate(Transaction tx) throws IOException {
-        getContainingList().storeNode(tx, this, true);
+        store(tx, ADD_LAST);
     }
 
     private void store(Transaction tx, boolean addFirst) throws IOException {
         try {
-            // When we split to a node of one element we can span multiple pages for that
+            // keeping splitting till we get down to a single entry
-            // entry, otherwise we keep the entries on one page to avoid fragmented reads
+            // then we need to overflow the value
-            // and segment the list traversal.
+            getContainingList().storeNode(tx, this, entries.size() == 1);
-            if (this.entries.size() == 1) {
-                getContainingList().storeNode(tx, this, true);
-            } else {
-                getContainingList().storeNode(tx, this, false);
-            }
 
             if (this.next == -1) {
                 getContainingList().setTailPageId(getPageId());
@@ -347,6 +342,7 @@ public final class ListNode<Key, Value> {
             this.setNext(extension.getPageId());
         } else {
             extension.setEntries(entries.getTail().getPrevious().splitAfter());
+            extension.setNext(this.getNext());
             extension.store(tx, isAddFirst);
             getContainingList().setTailPageId(extension.getPageId());
             this.setNext(extension.getPageId());

kahadb/src/main/java/org/apache/kahadb/util/SequenceSet.java

@@ -107,7 +107,14 @@ public class SequenceSet extends LinkedNodeList<Sequence> implements Iterable<Lo
             return true;
         }
 
-        Sequence sequence = getHead();
+        // check for append
+        Sequence sequence = getTail();
+        if (sequence.isAdjacentToLast(value)) {
+            sequence.last = value;
+            return true;
+        }
+
+        sequence = getHead();
         while (sequence != null) {
 
             if (sequence.isAdjacentToLast(value)) {

kahadb/src/test/java/org/apache/kahadb/index/ListIndexTest.java

@@ -516,7 +516,7 @@ public class ListIndexTest extends IndexTestSupport {
         int expectedListEntries = 0;
         int nextSequenceId = 0;
 
-        LOG.info("Loading up the ListIndex with "+NUM_ITERATIONS+" entires and sparsely populating the sequences.");
+        LOG.info("Loading up the ListIndex with "+NUM_ITERATIONS+" entries and sparsely populating the sequences.");
 
         for (int i = 0; i < NUM_ITERATIONS; ++i) {
             test.add(tx, String.valueOf(expectedListEntries++), new SequenceSet());