Upgraded to Eventuate Local 0.2.0.RELEASE

_build-and-test-all.sh

@@ -19,6 +19,12 @@ else
   ${DOCKER_COMPOSE?} rm -v --force
 fi
 
+NO_RM=false
+
+if [ "$1" = "--no-rm" ] ; then
+  NO_RM=true
+  shift
+fi
 
 ${DOCKER_COMPOSE?} up -d mongodb
 
@@ -54,5 +60,7 @@ set -e
 
 ./gradlew -a $* :e2e-test:cleanTest :e2e-test:test -P ignoreE2EFailures=false
 
-${DOCKER_COMPOSE?} stop
+if [ $NO_RM = false ] ; then
-${DOCKER_COMPOSE?} rm -v --force
+  ${DOCKER_COMPOSE?} stop
+  ${DOCKER_COMPOSE?} rm -v --force
+fi

java-spring/accounts-query-side-backend/src/main/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/AccountChangeInfo.java

@@ -1,5 +1,8 @@
 package net.chrisrichardson.eventstore.javaexamples.banking.backend.queryside.accounts;
 
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+
 public class AccountChangeInfo {
 
   private String changeId;
@@ -15,4 +18,14 @@ public class AccountChangeInfo {
     this.amount = amount;
     this.balanceDelta = balanceDelta;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    return EqualsBuilder.reflectionEquals(this, o);
+  }
+
+  @Override
+  public int hashCode() {
+    return HashCodeBuilder.reflectionHashCode(this);
+  }
 }

java-spring/accounts-query-side-backend/src/main/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/AccountInfo.java

@@ -1,6 +1,9 @@
 package net.chrisrichardson.eventstore.javaexamples.banking.backend.queryside.accounts;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Created by cer on 11/21/14.
@@ -13,13 +16,13 @@ public class AccountInfo {
   private String description;
   private long balance;
   private List<AccountChangeInfo> changes;
-  private List<AccountTransactionInfo> transactions;
+  private Map<String, AccountTransactionInfo> transactions;
   private String version;
 
   private AccountInfo() {
   }
 
-  public AccountInfo(String id, String customerId, String title, String description, long balance, List<AccountChangeInfo> changes, List<AccountTransactionInfo> transactions, String version) {
+  public AccountInfo(String id, String customerId, String title, String description, long balance, List<AccountChangeInfo> changes, Map<String, AccountTransactionInfo> transactions, String version) {
 
     this.id = id;
     this.customerId = customerId;
@@ -52,11 +55,11 @@ public class AccountInfo {
   }
 
   public List<AccountChangeInfo> getChanges() {
-    return changes;
+    return changes == null ? Collections.EMPTY_LIST : changes;
   }
 
   public List<AccountTransactionInfo> getTransactions() {
-    return transactions;
+    return transactions == null ? Collections.EMPTY_LIST : new ArrayList<>(transactions.values());
   }
 
   public String getVersion() {

java-spring/accounts-query-side-backend/src/main/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/AccountInfoUpdateService.java

@@ -16,11 +16,9 @@ import static org.springframework.data.mongodb.core.query.Criteria.where;
 public class AccountInfoUpdateService  {
   private Logger logger = LoggerFactory.getLogger(getClass());
 
-  private AccountInfoRepository accountInfoRepository;
   private MongoTemplate mongoTemplate;
 
-  public AccountInfoUpdateService(AccountInfoRepository accountInfoRepository, MongoTemplate mongoTemplate) {
+  public AccountInfoUpdateService(MongoTemplate mongoTemplate) {
-    this.accountInfoRepository = accountInfoRepository;
     this.mongoTemplate = mongoTemplate;
   }
 
@@ -28,29 +26,26 @@ public class AccountInfoUpdateService  {
 
   public void create(String accountId, String customerId, String title,  BigDecimal initialBalance, String description, String version) {
     try {
-      accountInfoRepository.save(new AccountInfo(
+      WriteResult x = mongoTemplate.upsert(new Query(where("id").is(accountId).and("version").exists(false)),
-              accountId,
+              new Update()
-              customerId,
+                      .set("customerId", customerId)
-              title,
+                      .set("title", title)
-              description,
+                      .set("description", description)
-              toIntegerRepr(initialBalance),
+                      .set("balance", toIntegerRepr(initialBalance))
-              Collections.<AccountChangeInfo>emptyList(),
+                      .set("version", version),
-              Collections.<AccountTransactionInfo>emptyList(),
+              AccountInfo.class);
-              version));
       logger.info("Saved in mongo");
     } catch (Throwable t) {
-      logger.error("Error during saving: ");
       logger.error("Error during saving: ", t);
       throw new RuntimeException(t);
     }
   }
 
 
-  public void addTransaction(String eventId, String fromAccountId, AccountTransactionInfo ti) {
+  public void addTransaction(String eventId, String accountId, AccountTransactionInfo ti) {
-    mongoTemplate.updateMulti(new Query(where("id").is(fromAccountId).and("version").lt(eventId)),
+    mongoTemplate.upsert(new Query(where("id").is(accountId)),
             new Update().
-                    push("transactions", ti).
+                    set("transactions." + eventId, ti),
-                    set("version", eventId),
             AccountInfo.class);
   }
 

java-spring/accounts-query-side-backend/src/main/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/AccountTransactionInfo.java

@@ -2,6 +2,7 @@ package net.chrisrichardson.eventstore.javaexamples.banking.backend.queryside.ac
 
 import org.apache.commons.lang.builder.EqualsBuilder;
 import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.commons.lang.builder.ToStringBuilder;
 
 import java.util.Date;
 
@@ -21,6 +22,12 @@ public class AccountTransactionInfo {
     this(transactionId, fromAccountId, toAccountId, amount, new Date(),  "");
   }
 
+  @Override
+  public String toString() {
+    return ToStringBuilder.reflectionToString(this);
+  }
+
+
   public AccountTransactionInfo(String transactionId, String fromAccountId, String toAccountId, long amount, Date date, String description) {
     this.transactionId = transactionId;
     this.fromAccountId = fromAccountId;

java-spring/accounts-query-side-backend/src/main/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/QuerySideAccountConfiguration.java

@@ -18,8 +18,8 @@ public class QuerySideAccountConfiguration {
   }
 
   @Bean
-  public AccountInfoUpdateService accountInfoUpdateService(AccountInfoRepository accountInfoRepository, MongoTemplate mongoTemplate) {
+  public AccountInfoUpdateService accountInfoUpdateService(MongoTemplate mongoTemplate) {
-    return new AccountInfoUpdateService(accountInfoRepository, mongoTemplate);
+    return new AccountInfoUpdateService(mongoTemplate);
   }
 
   @Bean

java-spring/accounts-query-side-backend/src/test/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/queryside/accounts/AccountInfoUpdateServiceTest.java

@@ -0,0 +1,96 @@
+package net.chrisrichardson.eventstore.javaexamples.banking.backend.queryside.accounts;
+
+import io.eventuate.javaclient.spring.jdbc.EventuateJdbcEventStoreConfiguration;
+import io.eventuate.javaclient.spring.jdbc.IdGenerator;
+import io.eventuate.javaclient.spring.jdbc.IdGeneratorImpl;
+import net.chrisrichardson.eventstore.javaexamples.banking.backend.common.accounts.AccountCreditedEvent;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.IntegrationTest;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.Assert.*;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(classes = AccountInfoUpdateServiceTest.AccountInfoUpdateServiceTestConfiguration.class)
+@IntegrationTest
+public class AccountInfoUpdateServiceTest {
+
+  @Configuration
+  @EnableAutoConfiguration
+  @Import({QuerySideAccountConfiguration.class, EventuateJdbcEventStoreConfiguration.class})
+  public static class AccountInfoUpdateServiceTestConfiguration {
+
+  }
+
+  @Autowired
+  private AccountInfoUpdateService accountInfoUpdateService;
+
+  @Autowired
+  private AccountQueryService accountQueryService;
+
+  @Test
+  public void shouldSaveAccountInfo() throws ExecutionException, InterruptedException {
+    IdGenerator x = new IdGeneratorImpl();
+    String accountId = x.genId().asString();
+    String customerId = x.genId().asString();
+    String version = x.genId().asString();
+
+    String title = "Checking account";
+    BigDecimal initialBalance = new BigDecimal("1345");
+    String description = "Some account";
+
+    accountInfoUpdateService.create(accountId, customerId, title, initialBalance, description, version);
+
+    AccountInfo accountInfo = accountQueryService.findByAccountId(accountId).get();
+
+    assertEquals(accountId, accountInfo.getId());
+    assertEquals(customerId, accountInfo.getCustomerId());
+    assertEquals(title, accountInfo.getTitle());
+    assertEquals(description, accountInfo.getDescription());
+    assertEquals(initialBalance.longValue() * 100, accountInfo.getBalance());
+    assertTrue(accountInfo.getChanges().isEmpty());
+    assertTrue(accountInfo.getTransactions().isEmpty());
+    assertEquals(version, accountInfo.getVersion());
+
+
+    String changeId = x.genId().asString();
+
+    String transactionId = x.genId().asString();
+
+    AccountChangeInfo change = new AccountChangeInfo(changeId, transactionId, AccountCreditedEvent.class.getSimpleName(),
+            500, +1);
+
+    accountInfoUpdateService.updateBalance(accountId, changeId, 500,
+            change);
+
+    accountInfo = accountQueryService.findByAccountId(accountId).get();
+    assertEquals(initialBalance.add(new BigDecimal(5)).longValue() * 100, accountInfo.getBalance());
+    assertFalse(accountInfo.getChanges().isEmpty());
+
+    assertEquals(change, accountInfo.getChanges().get(0));
+
+    String eventId = x.genId().asString();
+
+    AccountTransactionInfo ti = new AccountTransactionInfo(transactionId, accountId, accountId, 5, new Date(), "A transfer");
+
+    accountInfoUpdateService.addTransaction(eventId, accountId, ti);
+
+    accountInfo = accountQueryService.findByAccountId(accountId).get();
+    assertFalse(accountInfo.getTransactions().isEmpty());
+
+    assertEquals(ti, accountInfo.getTransactions().get(0));
+  }
+
+}

java-spring/backend-integration-tests/src/test/java/net/chrisrichardson/eventstore/javaexamples/banking/backend/BankingTestConfiguration.java

@@ -3,11 +3,13 @@ package net.chrisrichardson.eventstore.javaexamples.banking.backend;
 import io.eventuate.javaclient.spring.jdbc.EventuateJdbcEventStoreConfiguration;
 import net.chrisrichardson.eventstore.javaexamples.banking.backend.commandside.accounts.AccountConfiguration;
 import net.chrisrichardson.eventstore.javaexamples.banking.backend.commandside.transactions.MoneyTransferConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 @Configuration
 @Import({AccountConfiguration.class, MoneyTransferConfiguration.class, EventuateJdbcEventStoreConfiguration.class})
+@EnableAutoConfiguration
 public class BankingTestConfiguration {
 
 }

java-spring/buildSrc/src/main/groovy/EventuateDependencyPlugin.groovy

@@ -6,9 +6,10 @@ class EventuateDependencyPlugin implements Plugin<Project> {
     @Override
     void apply(Project project) {
         project.dependencies {
-            if (project.hasProperty("eventuateLocal"))
+            if (project.hasProperty("eventuateLocal")) {
                 compile "io.eventuate.local.java:eventuate-local-java-jdbc:${project.eventuateLocalVersion}"
-            else
+                compile "io.eventuate.local.java:eventuate-local-java-embedded-cdc-autoconfigure:${project.eventuateLocalVersion}"
+            } else
                 compile "io.eventuate.client.java:eventuate-client-java-http-stomp-spring:${project.eventuateClientVersion}"
         }
     }

java-spring/common-backend/src/main/resources/logback.xml

@@ -16,7 +16,7 @@
   <logger name="org.springframework" level='info'>
   </logger>
 
-  <logger name="net.chrisrichardson.eventstore.client" level='info'>
+  <logger name="io.eventuate" level='debug'>
   </logger>
 
 </configuration>

java-spring/docker-compose-eventuate-local.yml

@@ -18,8 +18,11 @@ apigateway:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
     SPRING_DATA_MONGODB_URI: mongodb://mongodb/mydb
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 accountscommandside:
   image: java:openjdk-8u91-jdk
@@ -34,7 +37,10 @@ accountscommandside:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 transactionscommandside:
   image: java:openjdk-8u91-jdk
@@ -49,7 +55,10 @@ transactionscommandside:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 
 accountsqueryside:
@@ -67,8 +76,11 @@ accountsqueryside:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
     SPRING_DATA_MONGODB_URI: mongodb://mongodb/mydb
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 customerscommandside:
   image: java:openjdk-8u91-jdk
@@ -83,7 +95,10 @@ customerscommandside:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 customersqueryside:
   image: java:openjdk-8u91-jdk
@@ -100,8 +115,11 @@ customersqueryside:
     SPRING_DATASOURCE_USERNAME:
     SPRING_DATASOURCE_PASSWORD:
     SPRING_DATASOURCE_DRIVER_CLASS_NAME:
-    EVENTUATELOCAL_AGGREGATESTORE_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_KAFKA_BOOTSTRAP_SERVERS:
+    EVENTUATELOCAL_ZOOKEEPER_CONNECTION_STRING:
     SPRING_DATA_MONGODB_URI: mongodb://mongodb/mydb
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_USER_NAME:
+    EVENTUATELOCAL_EMBEDDED_CDC_DB_PASSWORD:
 
 mongodb:
   image: mongo:3.0.4

java-spring/e2e-test/src/test/java/net/chrisrichardson/eventstore/examples/bank/web/EndToEndTest.java

@@ -28,6 +28,7 @@ import java.util.concurrent.CompletableFuture;
 import static net.chrisrichardson.eventstorestore.javaexamples.testutil.TestUtil.eventually;
 import static net.chrisrichardson.eventstorestore.javaexamples.testutil.customers.CustomersTestUtils.generateCustomerInfo;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 public class EndToEndTest {
 
@@ -115,15 +116,26 @@ public class EndToEndTest {
 
     // TOOD - check state of money transfer
 
-    List<AccountTransactionInfo> transactionInfoList = restTemplate.exchange(accountsQuerySideBaseUrl("/accounts/"+fromAccountId+"/history"),
+    eventually(
-            HttpMethod.GET,
+            () -> CompletableFuture.completedFuture(restTemplate.exchange(accountsQuerySideBaseUrl("/accounts/"+fromAccountId+"/history"),
-            new HttpEntity(BasicAuthUtils.basicAuthHeaders("test_user@mail.com")),
+                    HttpMethod.GET,
-            new ParameterizedTypeReference<List<AccountTransactionInfo>>() {}).getBody();
+                    new HttpEntity(BasicAuthUtils.basicAuthHeaders("test_user@mail.com")),
+                    new ParameterizedTypeReference<List<AccountTransactionInfo>>() {}).getBody()),
+            transactionInfoList -> {
+              if (!(transactionInfoList.stream().filter(ti -> ti.getTransactionId().equals(moneyTransfer.getMoneyTransferId()) &&
+                      ti.getFromAccountId().equals(fromAccountId) &&
+                      ti.getToAccountId().equals(toAccountId) &&
+                      ti.getAmount() == toCents(amountToTransfer).longValue()).findFirst().isPresent())) {
+                fail(String.format("%s does not contain %s %s %s",
+                        moneyTransfer.getMoneyTransferId(),
+                        fromAccount,
+                        toAccount,
+                        toCents(amountToTransfer).longValue()));
+              }
+
+            }
+    );
 
-    assertTrue(transactionInfoList.stream().filter(ti -> ti.getTransactionId().equals(moneyTransfer.getMoneyTransferId()) &&
-            ti.getFromAccountId().equals(fromAccountId) &&
-            ti.getToAccountId().equals(toAccountId) &&
-            ti.getAmount() == toCents(amountToTransfer).longValue()).findFirst().isPresent());
   }
 
   @Test
@@ -163,14 +175,14 @@ public class EndToEndTest {
     return dollarAmount.multiply(new BigDecimal(100));
   }
 
-  private void assertAccountBalance(final String fromAccountId, final BigDecimal expectedBalanceInDollars) {
+  private void assertAccountBalance(final String accountId, final BigDecimal expectedBalanceInDollars) {
     final BigDecimal inCents = toCents(expectedBalanceInDollars);
-    eventually(
+      eventually(
             new Producer<GetAccountResponse>() {
               @Override
               public CompletableFuture<GetAccountResponse> produce() {
                   return CompletableFuture.completedFuture(BasicAuthUtils.doBasicAuthenticatedRequest(restTemplate,
-                          accountsQuerySideBaseUrl("/accounts/" + fromAccountId),
+                          accountsQuerySideBaseUrl("/accounts/" + accountId),
                           HttpMethod.GET,
                           GetAccountResponse.class));
               }
@@ -178,8 +190,8 @@ public class EndToEndTest {
             new Verifier<GetAccountResponse>() {
               @Override
               public void verify(GetAccountResponse accountInfo) {
-                Assert.assertEquals(fromAccountId, accountInfo.getAccountId());
+                Assert.assertEquals(accountId, accountInfo.getAccountId());
-                Assert.assertEquals(inCents, accountInfo.getBalance());
+                Assert.assertEquals(accountId, inCents, accountInfo.getBalance());
               }
             });
   }

java-spring/gradle.properties

@@ -5,6 +5,6 @@ eventuateMavenRepoUrl=http://mavenrepo.eventuate.io/release
 
 springBootVersion=1.3.5.RELEASE
 
-eventuateClientVersion=0.6.0.RELEASE
+eventuateClientVersion=0.8.0.RELEASE
-eventuateLocalVersion=0.1.0.RELEASE
+eventuateLocalVersion=0.2.0.RELEASE
 

java-spring/monolithic-service/src/test/java/net/chrisrichardson/eventstore/javaexamples/banking/web/BankingWebIntegrationTest.java

@@ -35,6 +35,7 @@ import static net.chrisrichardson.eventstorestore.javaexamples.testutil.customer
 import static net.chrisrichardson.eventstorestore.javaexamples.testutil.customers.CustomersTestUtils.generateToAccountInfo;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringApplicationConfiguration(classes = BankingWebTestConfiguration.class)
@@ -102,23 +103,31 @@ public class BankingWebIntegrationTest {
         assertAccountBalance(fromAccountId, finalFromAccountBalance);
         assertAccountBalance(toAccountId, finalToAccountBalance);
 
-        List<AccountTransactionInfo> transactionInfoList = restTemplate.exchange(baseUrl("/accounts/"+fromAccountId+"/history"),
-                HttpMethod.GET,
-                new HttpEntity(BasicAuthUtils.basicAuthHeaders("test_user@mail.com")),
-                new ParameterizedTypeReference<List<AccountTransactionInfo>>() {}).getBody();
 
+        eventually(
+                () -> CompletableFuture.completedFuture(restTemplate.exchange(baseUrl("/accounts/"+fromAccountId+"/history"),
+                        HttpMethod.GET,
+                        new HttpEntity(BasicAuthUtils.basicAuthHeaders("test_user@mail.com")),
+                        new ParameterizedTypeReference<List<AccountTransactionInfo>>() {}).getBody()),
+                transactionInfoList -> {
+                    Optional<AccountTransactionInfo> txn = transactionInfoList.stream()
+                            .filter(ti -> ti.getTransactionId().equals(moneyTransfer.getMoneyTransferId()))
+                            .findFirst();
 
-        Optional<AccountTransactionInfo> first = transactionInfoList.stream().filter(ti -> ti.getTransactionId().equals(moneyTransfer.getMoneyTransferId())).findFirst();
+                    if (!txn.isPresent()) {
+                        fail(String.format("%s does not contain %s", transactionInfoList, moneyTransfer.getMoneyTransferId()));
+                    }
 
-        assertTrue(first.isPresent());
+                    AccountTransactionInfo ti = txn.get();
 
-        AccountTransactionInfo ti = first.get();
+                    assertEquals(fromAccountId, ti.getFromAccountId());
+                    assertEquals(toAccountId, ti.getToAccountId());
+                    assertEquals(toAccountId, ti.getToAccountId());
+                    assertEquals(fromAccountId, ti.getFromAccountId());
+                    assertEquals(toCents(amountToTransfer).longValue(), ti.getAmount());
+                }
+        );
 
-        assertEquals(fromAccountId, ti.getFromAccountId());
-        assertEquals(toAccountId, ti.getToAccountId());
-        assertEquals(toAccountId, ti.getToAccountId());
-        assertEquals(fromAccountId, ti.getFromAccountId());
-        assertEquals(toCents(amountToTransfer).longValue(), ti.getAmount());
 
     }
 

java-spring/testutil/src/main/java/net/chrisrichardson/eventstorestore/javaexamples/testutil/TestUtil.java

@@ -55,7 +55,7 @@ public class TestUtil {
   }
 
   public static <T> void eventually(final Producer<T> producer, final Verifier<T> verifier) {
-    final int n = 50;
+    final int n = 150;
     Object possibleException = Observable.timer(0, 200, TimeUnit.MILLISECONDS).flatMap(new Func1<Long, Observable<Outcome<T>>>() {
 
       @Override