从集合中创建元素流

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()。