added more fix and unit test

LinkedInAttic · Dec 31, 2011 · 250696f · 250696f
1 parent 96b6033
commit 250696f
Show file tree

Hide file tree

Showing 3 changed files with 73 additions and 7 deletions.
diff --git a/zoie-core/src/main/java/proj/zoie/impl/indexing/internal/LuceneIndexDataLoader.java b/zoie-core/src/main/java/proj/zoie/impl/indexing/internal/LuceneIndexDataLoader.java
@@ -259,7 +259,7 @@ public void loadFromIndex(RAMSearchIndex<R> ramIndex) throws ZoieException
 
         //V newVersion = idx.getVersion().compareTo(ramIndex.getVersion()) < 0 ? ramIndex.getVersion(): idx.getVersion();
         String newVersion = idx.getVersion() == null ? ramIndex.getVersion() : (_versionComparator.compare(idx.getVersion(), ramIndex.getVersion()) < 0 ? ramIndex.getVersion(): idx.getVersion());
-        idx.setVersion(newVersion);
+        idx.setVersion(newVersion);      
         //System.out.println("disk verson from the signature" + newVersion.toString());        
 
         //idx.setVersion(Math.max(idx.getVersion(), ramIndex.getVersion()));

diff --git a/zoie-core/src/main/java/proj/zoie/impl/indexing/internal/RealtimeIndexDataLoader.java b/zoie-core/src/main/java/proj/zoie/impl/indexing/internal/RealtimeIndexDataLoader.java
@@ -173,8 +173,9 @@ protected synchronized void processBatch()
       long t1=System.currentTimeMillis();
       try
       {
-        if(readOnlyMemIndex != null)
+        if(readOnlyMemIndex != null){
           _luceneDataLoader.loadFromIndex(readOnlyMemIndex);
+        }
       }
       catch (ZoieException e)
       {
@@ -191,14 +192,18 @@ protected synchronized void processBatch()
         {
           segmentCount = _idxMgr.getDiskSegmentCount();
           segmentInfo = _idxMgr.getDiskSegmentInfo();
+
+          IndexUpdatedEvent evt = new IndexUpdatedEvent(eventCount,t1,t2,_eventCount);
+          fireIndexingEvent(evt);
+          fireNewVersionEvent(readOnlyMemIndex.getVersion());
         } catch (IOException e)
         {
-          log.error("error getting new segment count after disk flush", e);
+          log.error("error getting disk information after disk flush", e);
+        }
+        if (log.isInfoEnabled()){
+          log.info("flushed batch of "+eventCount+" events to disk indexer, took: "+(t2-t1)+" current event count: "+_eventCount + ", current disk segment count: " + segmentCount);
+          log.info("post-flush segment info: " + segmentInfo);
         }
-        log.info("flushed batch of "+eventCount+" events to disk indexer, took: "+(t2-t1)+" current event count: "+_eventCount + ", current disk segment count: " + segmentCount);
-        log.info("post-flush segment info: " + segmentInfo);
-        IndexUpdatedEvent evt = new IndexUpdatedEvent(eventCount,t1,t2,_eventCount);
-        fireIndexingEvent(evt);
         notifyAll();
       }
     }

diff --git a/zoie-core/src/test/java/proj/zoie/test/ZoieTest.java b/zoie-core/src/test/java/proj/zoie/test/ZoieTest.java
@@ -50,6 +50,8 @@
 import proj.zoie.api.ZoieIndexReader;
 import proj.zoie.api.impl.DocIDMapperImpl;
 import proj.zoie.api.impl.InRangeDocIDMapperFactory;
+import proj.zoie.api.indexing.IndexingEventListener;
+import proj.zoie.api.indexing.IndexingEventListener.IndexingEvent;
 import proj.zoie.impl.indexing.AsyncDataConsumer;
 import proj.zoie.impl.indexing.MemoryStreamDataProvider;
 import proj.zoie.impl.indexing.ZoieConfig;
@@ -311,6 +313,65 @@ public int nextDoc() throws IOException {
 
 	}
 
+
+  @Test
+  public void testIndexEventListener() throws Exception {
+    File idxDir = getIdxDir();
+    final int[] flushNum = {0};
+    final String[] flushVersion = {null};
+
+    ZoieSystem<IndexReader, String> idxSystem = createZoie(
+        idxDir, true, ZoieConfig.DEFAULT_VERSION_COMPARATOR,true);
+
+    idxSystem.start();
+
+    idxSystem.addIndexingEventListener(new IndexingEventListener() {
+
+      @Override
+      public void handleUpdatedDiskVersion(String version) {
+        flushVersion[0] = version;
+      }
+
+      @Override
+      public void handleIndexingEvent(IndexingEvent evt) {
+        flushNum[0]++;
+      }
+    });
+
+    MemoryStreamDataProvider<String> memoryProvider = new MemoryStreamDataProvider<String>(ZoieConfig.DEFAULT_VERSION_COMPARATOR);
+    memoryProvider.setMaxEventsPerMinute(Long.MAX_VALUE);
+    memoryProvider.setDataConsumer(idxSystem);
+    memoryProvider.start();
+
+    try {
+      int count = DataForTests.testdata.length;
+      List<DataEvent<String>> list = new ArrayList<DataEvent<String>>(
+          count);
+      for (int i = 0; i < count; ++i) {
+        list.add(new DataEvent<String>(
+            DataForTests.testdata[i], ""+i));
+      }
+      memoryProvider.addEvents(list);
+      memoryProvider.flush();
+
+
+      idxSystem.flushEvents(10000);
+      String diskVersion = null;
+      while(!"9".equals(diskVersion)){
+        diskVersion = idxSystem.getCurrentDiskVersion();
+        Thread.sleep(500);
+      }
+
+    } finally {
+      memoryProvider.stop();
+      idxSystem.shutdown();
+      deleteDirectory(idxDir);
+    }
+
+    assertTrue(flushNum[0]>0);
+    assertEquals("9", flushVersion[0]);
+  }
+
 	@Test
   public void testPurgeFilter() throws Exception {
     File idxDir = getIdxDir();