added buffering supprot to GenericTypeahead via KratiBufferedInts

LinkedInAttic · Sep 17, 2012 · baaca5b · baaca5b
1 parent 0b61706
commit baaca5b
Show file tree

Hide file tree

Showing 24 changed files with 660 additions and 87 deletions.
diff --git a/pom.xml b/pom.xml
@@ -10,7 +10,7 @@
   <groupId>com.sna-projects.cleo</groupId>
   <artifactId>cleo</artifactId>
   <packaging>bundle</packaging>
-  <version>1.2.4</version>
+  <version>1.2.5-SNAPSHOT</version>
 
   <name>cleo</name>
   <description>Flexible, partial, out-of-order real-time typeahead search</description>
@@ -148,7 +148,7 @@
     <dependency>
       <groupId>com.sna-projects.krati</groupId>
       <artifactId>krati</artifactId>
-      <version>0.4.7</version>
+      <version>0.4.7.1</version>
       <scope>compile</scope>
     </dependency>
 

diff --git a/src/main/java/cleo/search/store/KratiArrayStoreConnections.java b/src/main/java/cleo/search/store/KratiArrayStoreConnections.java
@@ -16,6 +16,8 @@
 
 package cleo.search.store;
 
+import java.io.IOException;
+
 /**
  * KratiArrayStoreConnections
  * 
@@ -25,68 +27,113 @@
  * <p>
  * 09/18, 2011 - Added readBytes to support partial reads <br/>
  */
-public final class KratiArrayStoreConnections extends KratiArrayStoreInts implements ArrayStoreConnections {
-
-  public KratiArrayStoreConnections(KratiArrayStore store) {
-    this(store, 0);
-  }
+public final class KratiArrayStoreConnections implements ArrayStoreConnections {
+  /**
+   * The underlying ArrayStore for list of integers.
+   */
+  protected final KratiArrayStoreInts storeInts;
 
-  public KratiArrayStoreConnections(KratiArrayStore store, int indexStart) {
-    super(store);
+  public KratiArrayStoreConnections(KratiArrayStoreInts storeInts) {
+    this.storeInts = storeInts;
   }
 
   @Override
   public int[] getConnections(int source) {
-    return get(source);
+    return storeInts.get(source);
   }
 
   @Override
   public void setConnections(int source, int[] connections, long scn) throws Exception {
-    set(source, connections, scn);
+    storeInts.set(source, connections, scn);
   }
 
   @Override
   public void deleteConnections(int source, long scn) throws Exception {
-    delete(source, scn);
+    storeInts.delete(source, scn);
   }
 
   @Override
   public void addConnection(int source, int connection, long scn) throws Exception {
-    add(source, connection, scn);
+    storeInts.add(source, connection, scn);
   }
 
   @Override
   public void removeConnection(int source, int connection, long scn) throws Exception {
-    remove(source, connection, scn);
+    storeInts.remove(source, connection, scn);
   }
 
   @Override
   public byte[] getBytes(int index) {
-    return store.get(index);
+    return storeInts.getUnderlyingStore().get(index);
   }
 
   @Override
   public int getBytes(int index, byte[] dst) {
-    return store.get(index, dst);
+    return storeInts.getUnderlyingStore().get(index, dst);
   }
 
   @Override
   public int getBytes(int index, byte[] dst, int offset) {
-    return store.get(index, dst, offset);
+    return storeInts.getUnderlyingStore().get(index, dst, offset);
   }
 
   @Override
   public int readBytes(int index, byte[] dst) {
-    return store.read(index, dst);
+    return storeInts.getUnderlyingStore().read(index, dst);
   }
 
   @Override
   public int readBytes(int index, int offset, byte[] dst) {
-    return store.read(index, offset, dst);
+    return storeInts.getUnderlyingStore().read(index, offset, dst);
   }
 
   @Override
   public int getLength(int index) {
-    return store.getLength(index);
+    return storeInts.getUnderlyingStore().getLength(index);
+  }
+
+  @Override
+  public void sync() throws IOException {
+    storeInts.sync();
+  }
+
+  @Override
+  public void persist() throws IOException {
+    storeInts.persist();
+  }
+
+  @Override
+  public long getLWMark() {
+    return storeInts.getLWMark();
+  }
+
+  @Override
+  public long getHWMark() {
+    return storeInts.getHWMark();
+  }
+
+  @Override
+  public void saveHWMark(long endOfPeriod) throws Exception {
+    storeInts.saveHWMark(endOfPeriod);
+  }
+
+  @Override
+  public void clear() {
+    storeInts.clear();
+  }
+
+  @Override
+  public int length() {
+    return storeInts.length();
+  }
+
+  @Override
+  public boolean hasIndex(int index) {
+    return storeInts.hasIndex(index);
+  }
+
+  @Override
+  public Type getType() {
+    return storeInts.getType();
   }
 }
diff --git a/src/main/java/cleo/search/store/KratiArrayStoreInts.java b/src/main/java/cleo/search/store/KratiArrayStoreInts.java
@@ -37,6 +37,13 @@ public KratiArrayStoreInts(KratiArrayStore store) {
     this.store = store;
   }
 
+  /**
+   * @return the underlying {@link KratiArrayStore}.
+   */
+  public final KratiArrayStore getUnderlyingStore() {
+    return store;
+  }
+
   @Override
   public String getStatus() {
     return store.getStatus();
@@ -158,6 +165,43 @@ public synchronized void add(int index, int elemId, long scn) throws Exception {
     internalPersist(scn);
   }
 
+  /**
+   * Adds an array of element IDs to the specified <code>index</code>
+   * 
+   * @param index   - the array store index
+   * @param elemIds - the array of element IDs
+   * @param scn     - the monotonically increasing System Change Number (SCN)
+   * @throws Exception
+   */
+  public synchronized void add(int index, int[] elemIds, long scn) throws Exception {
+    store.expandCapacity(index);
+
+    byte[] upd = null;
+    byte[] dat = store.get(index);
+
+    if (dat == null) {
+      upd = new byte[NUM_BYTES_IN_INT * elemIds.length];
+      ByteBuffer bb = ByteBuffer.wrap(upd);
+      for(int i = 0; i < elemIds.length; i++) {
+        bb.putInt(elemIds[i]);
+      }
+    } else {
+      ByteBuffer bb = ByteBuffer.wrap(dat);
+
+      int safeLen = dat.length - (dat.length % NUM_BYTES_IN_INT);
+      upd = new byte[safeLen + NUM_BYTES_IN_INT * elemIds.length];
+      bb = ByteBuffer.wrap(upd);
+      bb.put(dat, 0, safeLen);
+      for(int i = 0; i < elemIds.length; i++) {
+        bb.putInt(elemIds[i]);
+      }
+    }
+
+    // Update store
+    store.set(index, upd, scn);
+    internalPersist(scn);
+  }
+
   @Override
   public synchronized void remove(int index, int elemId, long scn) throws Exception {
     store.expandCapacity(index);