diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java index 40de95764..b7d32b056 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java @@ -3,10 +3,12 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.fisco.bcos.sdk.jni.common.JniException; import org.fisco.bcos.sdk.v3.amop.Amop; @@ -46,59 +48,48 @@ public void amopAsyncSubTest() throws ConfigException, JniException, Interrupted amopBroadCast.start(); subAmop.start(); - ThreadPoolService threadPoolService = new ThreadPoolService("amop", 1000); + final int pubCount = 5; + CountDownLatch receiveLatch = new CountDownLatch(pubCount); + AtomicReference recvError = new AtomicReference<>(); - threadPoolService - .getThreadPool() - .execute( - () -> { - int count = 5; - while (count-- > 0) { - System.out.println( - " ====== AMOP broadcast, topic: " - + topic - + " ,msg: " - + message); - amopBroadCast.broadcastAmopMsg(topic, message.getBytes()); - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - }); + // subscribe once; the callback stays registered for the whole test + subAmop.subscribeTopic( + topic, + (endpoint, seq, data) -> { + System.out.println(" ==> receive message from client"); + System.out.println(" \t==> endpoint: " + endpoint); + System.out.println(" \t==> seq: " + seq); + System.out.println(" \t==> data: " + new String(data)); + if (!message.equals(new String(data))) { + recvError.compareAndSet( + null, "unexpected message: " + new String(data)); + } + subAmop.sendResponse(endpoint, seq, data); + receiveLatch.countDown(); + }); - threadPoolService - .getThreadPool() - .execute( - () -> { - int count = 5; - while (count-- > 0) { - CompletableFuture future = new CompletableFuture<>(); - subAmop.subscribeTopic( - topic, - (endpoint, seq, data) -> { - System.out.println(" ==> receive message from client"); - System.out.println(" \t==> endpoint: " + endpoint); - System.out.println(" \t==> seq: " + seq); - System.out.println(" \t==> data: " + new String(data)); - Assert.assertEquals(new String(data), message); - subAmop.sendResponse(endpoint, seq, data); - future.complete(false); - }); - try { - future.get(10, TimeUnit.SECONDS); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - e.printStackTrace(); - } - } - }); - Thread.sleep(10000); - threadPoolService.stop(); + // the subscription is pushed to the node asynchronously; broadcasts sent + // before it takes effect are dropped silently, so wait for it first + Thread.sleep(3000); + + for (int i = 0; i < pubCount; i++) { + System.out.println(" ====== AMOP broadcast, topic: " + topic + " ,msg: " + message); + amopBroadCast.broadcastAmopMsg(topic, message.getBytes()); + Thread.sleep(1000); + } + + // fail the test if not all broadcasts are received, instead of swallowing + // a TimeoutException + boolean allReceived = receiveLatch.await(15, TimeUnit.SECONDS); amopBroadCast.stop(); subAmop.stop(); amopBroadCast.destroy(); subAmop.destroy(); + Assert.assertNull(recvError.get(), recvError.get()); + Assert.assertTrue( + "only received " + (pubCount - receiveLatch.getCount()) + "/" + pubCount + + " broadcast messages", + allReceived); } @Test @@ -142,7 +133,7 @@ public void amopSubAsyncTest() endpoint, seq, message2.getBytes()); })); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); @@ -199,7 +190,7 @@ public void amopSubTest() subAmop.sendResponse(endpoint, seq, message2.getBytes()); }); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); @@ -258,7 +249,7 @@ public void amopUnsubTest() subAmop.getSubTopics(); subAmop.unsubscribeTopic(topic); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java index 2eaf48a4c..140e59f01 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java @@ -301,7 +301,7 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { client.getTotalTransactionCount() .getTotalTransactionCount() .getTransactionCount()); - for (int i = 0; i < 100; i++) { + for (int i = 0; i < 20; i++) { Integer index = i; threadPool.execute( () -> { @@ -334,19 +334,43 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { .getTotalTransactionCount() .getTransactionCount()); System.out.println("orgTxCount: " + orgTxCount + ", currentTxCount:" + currentTxCount); - Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(300))) >= 0); + Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(60))) >= 0); client.stop(); client.destroy(); } class FakeTransactionCallback implements PrecompiledCallback { public TransactionReceipt receipt; + private final ExecutorService executor; + private final Runnable onSuccess; - // wait until get the transactionReceipt + FakeTransactionCallback(ExecutorService executor, Runnable onSuccess) { + this.executor = executor; + this.onSuccess = onSuccess; + } + + // wait until get the transactionReceipt; count by the receipt status only: + // for CRUD precompiled calls a successful retCode carries the affected row + // count (e.g. 1 for a successful insert), not 0 @Override public void onResponse(RetCode retCode) { this.receipt = retCode.getTransactionReceipt(); - PrecompiledTest.this.receiptCount.addAndGet(1); + if (this.receipt != null && this.receipt.isStatusOK()) { + PrecompiledTest.this.receiptCount.addAndGet(1); + if (onSuccess != null) { + // the next CRUD call resolves the table address with a blocking + // call, it must not run on the sdk callback thread + executor.execute(onSuccess); + } + } else { + System.out.println( + "async crud failed, code: " + + retCode.getCode() + + ", message: " + + retCode.getMessage() + + ", receipt status: " + + (this.receipt == null ? "null" : this.receipt.getStatus())); + } } } @@ -376,29 +400,52 @@ public void test52AsyncCRUDService() client.getTotalTransactionCount() .getTotalTransactionCount() .getTransactionCount()); - for (int i = 0; i < 100; i++) { + for (int i = 0; i < 20; i++) { int index = i; threadPool.execute( () -> { try { LinkedHashMap value = new LinkedHashMap<>(); value.put("field", "field" + index); - // insert - FakeTransactionCallback callback = new FakeTransactionCallback(); + // chain insert -> update -> remove per key through the callbacks, + // firing them together races and fails with "Key not exist" crudService.asyncInsert( tableName, new Entry(valueFiled, "key" + index, value), - callback); - // update - value.clear(); - value.put("field", "field" + index + 100); - UpdateFields updateFields = new UpdateFields(value); - FakeTransactionCallback callback2 = new FakeTransactionCallback(); - crudService.asyncUpdate( - tableName, "key" + index, updateFields, callback2); - // remove - FakeTransactionCallback callback3 = new FakeTransactionCallback(); - crudService.asyncRemove(tableName, "key" + index, callback3); + new FakeTransactionCallback( + threadPool, + () -> { + try { + LinkedHashMap newValue = + new LinkedHashMap<>(); + newValue.put("field", "field" + index + 100); + crudService.asyncUpdate( + tableName, + "key" + index, + new UpdateFields(newValue), + new FakeTransactionCallback( + threadPool, + () -> { + try { + crudService.asyncRemove( + tableName, + "key" + index, + new FakeTransactionCallback( + threadPool, + null)); + } catch (ContractException + e) { + System.out.println( + "asyncRemove failed: " + + e.getMessage()); + } + })); + } catch (ContractException e) { + System.out.println( + "asyncUpdate failed: " + + e.getMessage()); + } + })); } catch (ContractException e) { System.out.println( "call crudService failed, error information: " @@ -406,8 +453,10 @@ public void test52AsyncCRUDService() } }); } - while (this.receiptCount.get() != 300) { - Thread.sleep(1000); + // wait for all async callbacks, but fail instead of hanging forever + long deadline = System.currentTimeMillis() + 60000; + while (this.receiptCount.get() != 60 && System.currentTimeMillis() < deadline) { + Thread.sleep(100); } ThreadPoolService.stopThreadPool(threadPool); BigInteger currentTxCount = @@ -416,7 +465,8 @@ public void test52AsyncCRUDService() .getTotalTransactionCount() .getTransactionCount()); System.out.println("orgTxCount: " + orgTxCount + ", currentTxCount:" + currentTxCount); - Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(300))) >= 0); + Assert.assertEquals(60, this.receiptCount.get()); + Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(60))) >= 0); client.stop(); client.destroy(); } diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java index a986e47c2..f886c21fe 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java @@ -17,6 +17,7 @@ import java.math.BigInteger; import java.util.*; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.collections4.ListUtils; import org.apache.commons.lang3.StringUtils; @@ -190,8 +191,9 @@ public void test11HelloWorldAsync() throws Exception { }); System.out.println("--- finish deploy with CompletableFuture ---"); - // wait for the async thread - Thread.sleep(1000); + // wait for the async deploy deterministically instead of a fixed sleep + TransactionReceipt receipt = future.get(10, TimeUnit.SECONDS); + Assert.assertEquals(0, receipt.getStatus()); } @Test diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java index ea2f57294..e124e16ec 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java @@ -18,6 +18,7 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; import org.fisco.bcos.sdk.jni.utilities.tx.TransactionBuilderJniObj; @@ -212,7 +213,8 @@ public void test2HelloWorldAsync() throws Exception { return null; }); - // wait for the async thread - Thread.sleep(1000); + // wait for the async operations deterministically instead of a fixed sleep + Assert.assertEquals(0, future.get(10, TimeUnit.SECONDS).getStatus()); + Assert.assertEquals(0, future2.get(10, TimeUnit.SECONDS).getStatus()); } }