從集合中創建元素流

Question

我正在使用Junit 5動態測試。 我的目的是從集合中創建一個元素流，並將其傳遞給JUnit5進行測試。 但是，通過此代碼，我只能運行1000條記錄。我如何使這個工作無縫地無阻塞。

MongoCollection<Document> collection = mydatabase.getCollection("mycoll"); 
    final List<Document> cache = Collections.synchronizedList(new ArrayList<Document>()); 

    FindIterable<Document> f = collection.find().batchSize(1000); 
    f.batchCursor(new SingleResultCallback<AsyncBatchCursor<Document>>() { 

     @Override 
     public void onResult(AsyncBatchCursor<Document> t, Throwable thrwbl) { 
      t.next(new SingleResultCallback<List<Document>>() { 

       @Override 
       public void onResult(List<Document> t, Throwable thrwbl) { 
        if (thrwbl != null) { 
         th.set(thrwbl); 
        } 
        cache.addAll(t); 
        latch.countDown();; 

       } 
      }); 
     } 
    }); 
    latch.await(); 
    return cache.stream().map(batch->process(batch)); 

更新的代碼

@ParameterizedTest 
@MethodSource("setUp") 
void cacheTest(MyClazz myclass) throws Exception { 
    assertTrue(doTest(myclass)); 
} 
public static MongoClient getMongoClient() { 
// get client here 
} 

private static Stream<MyClazz> setUp() throws Exception { 
    MongoDatabase mydatabase = getMongoClient().getDatabase("test"); 
    List<Throwable> failures = new ArrayList<>(); 
    CountDownLatch latch = new CountDownLatch(1); 
    List<MyClazz> list = Collections.synchronizedList(new ArrayList<>()); 
      mydatabase.getCollection("testcollection").find() 
      .toObservable().subscribe(
      document -> { 
       list.add(process(document)); 
      }, 
      throwable -> { 
       failures.add(throwable); 
      }, 
      () -> { 
       latch.countDown(); 
      }); 
    latch.await(); 
    return list.stream(); 
} 

public boolean doTest(MyClazz myclass) { 
// processing goes here 
} 
public MyClazz process(Document doc) { 
// doc gets converted to MyClazz 
    return MyClazz; 
} 

即使是現在，我看到它的單元測試發生後，所有的數據被加載。 我認爲這是因爲latch.await（）。但是，如果我刪除它，則有可能沒有運行測試用例，因爲db可能正在加載集合。

我的用例是：我有mongo中的百萬條記錄，並且正在運行它們的集成測試用例。將所有內容加載到內存中是不可行的，因此我正在嘗試流式傳輸解決方案。

Answer 1

我不認爲我完全理解你的使用情況，但考慮到你的問題被打上java和mongo-asyc-driver這個要求肯定是可以實現的：

從集合創建元素的流上傳遞測試...使這項工作無縫無阻塞

下面的代碼：

• 用途將M ongoDB RxJava驅動程序查詢集合
• 從集合創建的Rx Observable
• 訂閱該Observable
• 記錄異常

• 標記完成

  CountDownLatch latch = new CountDownLatch(1); 
  List<Throwable> failures = new ArrayList<>(); 
  collection.find() 
       .toObservable().subscribe(
       // on next, this is invoked for each document returned by your find call 
       document -> { 
        // presumably you'll want to do something here to meet this requirement: "pass it on to test in JUnit5" 
        System.out.println(document); 
       }, 
       /// on error 
       throwable -> { 
        failures.add(throwable); 
       }, 
       // on completion 
       () -> { 
        latch.countDown(); 
       }); 
  // await the completion event 
  latch.await(); 
  

注：

• 這需要使用MongoDB RxJava driver（即類在com.mongodb.rx.client命名空間......在org.mongodb::mongodb-driver-rx Maven構件）
• 在你的問題要調用collection.find().batchSize()這清楚地表明，你是不是目前使用的Rx驅動程序（因爲batchSize不能是一個友好的Rx概念:)
• 上面的代碼進行了驗證MongoDB的RxJava司機的V1.4.0和io.reactive::rxjava

更新v1.1.10 1：基於改變你的問題（下面我原來的答覆），你問：「我看到單元測試發生之後所有數據都被加載。我認爲這是因爲latch.await（）「？我認爲你只是在可觀察到的數據流後，才從可觀察數據流中彈出一個列表，然後你是否開始調用doTest()。這種方法涉及（1）從MongoDB流式傳輸結果; （2）將這些結果存儲在內存中; （3）針對每個存儲的結果運行doTest()。如果您確實想要全程流式傳輸，那麼您應該在您可觀察的訂閱中調用doTest()。例如：

mydatabase.getCollection("testcollection").find() 
     .toObservable().subscribe(
     document -> { 
      doTest(process(document)); 
     }, 
     throwable -> { 
      failures.add(throwable); 
     }, 
     () -> { 
      latch.countDown(); 
     }); 
latch.await(); 

，因爲它從接收的MongoDB每個文檔和當整個觀察到的耗盡時鎖存器將被遞減，並且代碼將完成上面的代碼將調用doTest()。