EventuateRESTClient implements AggregateCrud { @Override public CompletableFuture<EntityIdVersionAndEventIds> update(EntityIdAndType aggregateIdAndType, Int128 entityVersion, List<EventTypeAndData> events, Optional<AggregateCrudUpdateOptions> updateOptions) { return withRetry(() -> { if (logger.isDebugEnabled()) logger.debug("update: " + aggregateIdAndType.getEntityType() + ", " + aggregateIdAndType.getEntityId() + ", " + ", events" + events + ", " + updateOptions); CompletableFuture<EntityIdVersionAndEventIds> cf = new CompletableFuture<>(); UpdateEntityRequest request = new UpdateEntityRequest(events, entityVersion, updateOptions.flatMap(AggregateCrudUpdateOptions::getTriggeringEvent).map(EventContext::getEventToken).orElse(null)); String json = JSonMapper.toJson(request); context.runOnContext(ignore -> { httpClient.post(makePath() + "/" + aggregateIdAndType.getEntityType() + "/" + aggregateIdAndType.getEntityId()) .handler(response -> { if (logger.isDebugEnabled()) logger.debug("HTTP response=" + response.statusCode()); if (response.statusCode() == 200) response.bodyHandler(body -> { try { UpdateEntityResponse r = JSonMapper.fromJson(body.toString(), UpdateEntityResponse.class); cf.complete(new EntityIdVersionAndEventIds(r.getEntityId(), r.getEntityVersion(), r.getEventIds())); } catch (Throwable e) { cf.completeExceptionally(e); } }); else { handleErrorResponse(cf, response, aggregateIdAndType, entityVersion); } }) .putHeader("authorization", authorizationHeader) .putHeader("content-type", "application/json") .exceptionHandler(cf::completeExceptionally) .end(Buffer.buffer(json)); }); return cf; }); } EventuateRESTClient(Vertx vertx, EventuateCredentials credentials, URI url); @Override CompletableFuture<EntityIdVersionAndEventIds> save(String aggregateType, List<EventTypeAndData> events, Optional<AggregateCrudSaveOptions> options); @Override CompletableFuture<LoadedEvents> find(String aggregateType, String entityId, Optional<AggregateCrudFindOptions> findOptions); @Override CompletableFuture<EntityIdVersionAndEventIds> update(EntityIdAndType aggregateIdAndType, Int128 entityVersion, List<EventTypeAndData> events, Optional<AggregateCrudUpdateOptions> updateOptions);  }
@Test public void shouldHandleErrorAfterServiceUnavailable() throws InterruptedException, ExecutionException, TimeoutException { mockHttpServer.expect().respondingWith(503); mockHttpServer.expect().respondingWith(409, RequestResponseJsonObjects.makeOptimisticLockingErrorResponse()); CompletableFuture<EntityIdVersionAndEventIds> f = client.update(new EntityIdAndType(RequestResponseJsonObjects.ENTITY_ID, RequestResponseJsonObjects.aggregateType), new Int128(5, 6), Collections.singletonList(new EventTypeAndData(RequestResponseJsonObjects.debitedEvent, RequestResponseJsonObjects.eventData, Optional.empty())), Optional.empty()); try { f.get(4, TimeUnit.SECONDS); fail(); } catch (ExecutionException e) { if (!(e.getCause() instanceof OptimisticLockingException)) throw e; } mockHttpServer.assertSatisfied(); }