fix(polling-registry) Add more choreography in tests to fix race issues.
MODIFY
gateway/engine/core/src/main/java/io/apiman/gateway/engine/impl/InMemoryRegistry.java
gateway/engine/core/src/main/java/io/apiman/gateway/engine/impl/InMemoryRegistry.java
@@ -268,7 +268,7 @@ public class InMemoryRegistry implements IRegistry {
/**
* @return the map to use when storing stuff
*/
- protected Map<String, Object> getMap() {
+ public Map<String, Object> getMap() {
return map;
}

MODIFY
gateway/engine/ispn/src/main/java/io/apiman/gateway/engine/ispn/InfinispanRegistry.java
gateway/engine/ispn/src/main/java/io/apiman/gateway/engine/ispn/InfinispanRegistry.java
@@ -65,7 +65,7 @@ public class InfinispanRegistry extends InMemoryRegistry {
* @see io.apiman.gateway.engine.impl.InMemoryRegistry#getMap()
*/
@Override
- protected Map<String, Object> getMap() {
+ public Map<String, Object> getMap() {
if (cacheWrapper == null) {
cacheWrapper = new RegistryCacheMapWrapper(getCache());
}

MODIFY
gateway/engine/vertx-polling/src/main/java/io/apiman/gateway/engine/vertx/polling/URILoadingRegistry.java
gateway/engine/vertx-polling/src/main/java/io/apiman/gateway/engine/vertx/polling/URILoadingRegistry.java
@@ -150,6 +150,7 @@ public class URILoadingRegistry extends InMemoryRegistry implements AsyncInitial
awaiting.addAll(allRegistries);
apis.clear();
clients.clear();
+ failureHandlers.clear(); // Initial design assumption was that this would never be reloaded - so this is the future for boot-time and thus has already been called.
allRegistries.stream()
.map(URILoadingRegistry::getMap)
.forEach(Map::clear);

MODIFY
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/engine/VertxConfigDrivenEngineFactory.java
gateway/platforms/vertx3/vertx3/src/main/java/io/apiman/gateway/platforms/vertx3/engine/VertxConfigDrivenEngineFactory.java
@@ -27,6 +27,8 @@ import io.apiman.gateway.platforms.vertx3.common.AsyncInitialize;
import io.apiman.gateway.platforms.vertx3.common.config.VertxEngineConfig;
import io.apiman.gateway.platforms.vertx3.connector.ConnectorFactory;
import io.vertx.core.Vertx;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
import java.lang.reflect.Constructor;
import java.util.Map;
@@ -45,6 +47,7 @@ public class VertxConfigDrivenEngineFactory extends ConfigDrivenEngineFactory {
private int started = 0;
private static final int ELEMENTS_TO_START = 6;
private boolean failed = false;
+ private Logger log = LoggerFactory.getLogger(VertxConfigDrivenEngineFactory.class);
public VertxConfigDrivenEngineFactory(Vertx vertx, VertxEngineConfig config) {
@@ -70,8 +73,12 @@ public class VertxConfigDrivenEngineFactory extends ConfigDrivenEngineFactory {
if (instance instanceof AsyncInitialize) {
((AsyncInitialize) instance).initialize(initResult -> {
if (initResult.isError()) {
+ if (!failed) {
handler.handle(initResult);
failed = true;
+ } else {
+ log.error("Failure occurred, but error handler was already invoked.", initResult.getError().getCause()); //$NON-NLS-1$
+ }
}
});
}

MODIFY
gateway/test/src/test/java/io/apiman/gateway/test/junit/vertx3/ApiToFileRegistry.java
gateway/test/src/test/java/io/apiman/gateway/test/junit/vertx3/ApiToFileRegistry.java
@@ -85,7 +85,7 @@ public class ApiToFileRegistry extends InMemoryRegistry {
}
public void reset() {
- fileSystem.deleteBlocking(file.getAbsolutePath());
+ fileSystem.writeFileBlocking(file.getAbsolutePath(), Buffer.buffer(""));
super.getMap().clear();
apis.clear();
clients.clear();
@@ -93,36 +93,44 @@ public class ApiToFileRegistry extends InMemoryRegistry {
apiMap.clear();
root.clear();
linkRoot();
+ resetCompleteSignal();
}
private void rewrite() {
fileSystem.writeFileBlocking(file.getAbsolutePath(), Buffer.buffer(root.encodePrettily()));
+ rewriteCompleteSignal();
+ }
+
+ private void rewriteCompleteSignal() {
+ eb.publish("rewrite", null);
+ System.out.println("sent rewrite");
+ }
+
+ private void resetCompleteSignal() {
+ eb.publish("reset", null);
+ System.out.println("sent reset");
}
private void publish(Api api) {
JsonObject apiJson = new JsonObject(Json.encode(api));
apis.add(apiJson);
apiMap.put(api, apiJson);
- rewrite();
}
private void retire(Api api) {
JsonObject removeJson = apiMap.remove(api);
apis.remove(removeJson);
- rewrite();
}
private void register(Client client) {
JsonObject clientJson = new JsonObject(Json.encode(client));
clients.add(clientJson);
clientMap.put(client, clientJson);
- rewrite();
}
private void unregister(Client lookup) {
JsonObject removeJson = clientMap.remove(lookup);
clients.remove(removeJson);
- rewrite();
}
@Override
@@ -132,6 +140,7 @@ public class ApiToFileRegistry extends InMemoryRegistry {
publish(api);
}
handler.handle(result);
+ rewrite();
});
}
@@ -142,6 +151,7 @@ public class ApiToFileRegistry extends InMemoryRegistry {
retire(api);
}
handler.handle(result);
+ rewrite();
});
}
@@ -152,6 +162,7 @@ public class ApiToFileRegistry extends InMemoryRegistry {
register(client);
}
handler.handle(result);
+ rewrite();
});
}
@@ -162,6 +173,7 @@ public class ApiToFileRegistry extends InMemoryRegistry {
unregister(client);
}
handler.handle(result);
+ rewrite();
});
}

MODIFY
gateway/test/src/test/java/io/apiman/gateway/test/junit/vertx3/Vertx3GatewayFileRegistryServer.java
gateway/test/src/test/java/io/apiman/gateway/test/junit/vertx3/Vertx3GatewayFileRegistryServer.java
@@ -24,6 +24,7 @@ import io.vertx.core.DeploymentOptions;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.json.JsonObject;
import java.io.File;
@@ -54,6 +55,10 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
private JsonObject vertxConf;
private Vertx secondVx;
private JsonObject pushEmulatorConfig;
+ private MessageConsumer<Object> resetConsumer;
+ private MessageConsumer<Object> rewriteConsumer;
+ private CountDownLatch rewriteCdl = new CountDownLatch(1);
+ private CountDownLatch resetCdl = new CountDownLatch(1);
/**
* Constructor.
@@ -71,6 +76,9 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
secondVx.deployVerticle(ApiVerticle.class.getCanonicalName(),
new DeploymentOptions().setConfig(pushEmulatorConfig));
+
+ resetConsumer = secondVx.eventBus().consumer("reset").handler(reset -> resetCdl.countDown());
+ rewriteConsumer = secondVx.eventBus().consumer("rewrite").handler(rewritten -> rewriteCdl.countDown());
}
private JsonObject loadJsonObjectFromResources(JsonNode nodeConfig, String name) {
@@ -108,8 +116,8 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
echoServer.start();
startLatch = new CountDownLatch(1);
- DeploymentOptions options = new DeploymentOptions();
- options.setConfig(vertxConf);
+ DeploymentOptions options = new DeploymentOptions().
+ setConfig(vertxConf);
vertx.deployVerticle(InitVerticle.class.getCanonicalName(),
options, event -> {
@@ -131,6 +139,7 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
secondVx.eventBus().publish(ApiToFileRegistry.class.getCanonicalName(), null,
new DeliveryOptions().addHeader("action", "reset"));
+
URILoadingRegistry.reset();
vertx.close(result -> {
@@ -139,9 +148,12 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
try {
stopLatch.await();
+ resetCdl.await();
} catch (Exception e) {
throw new RuntimeException(e);
}
+
+ resetCdl = new CountDownLatch(1);
}
@Override
@@ -149,11 +161,17 @@ public class Vertx3GatewayFileRegistryServer implements IGatewayTestServer {
try {
if (URI.create(endpoint).getPort() == API_PORT) {
CountDownLatch cdl = new CountDownLatch(1);
+
+ System.out.println("awaiting rewrite");
+ rewriteCdl.await();
+
URILoadingRegistry.reloadData(done -> {
cdl.countDown();
});
+
cdl.await();
System.out.println("Next...");
+ rewriteCdl = new CountDownLatch(1);
}
} catch (Exception e) {
throw new RuntimeException(e);

fix(polling-registry) Add more choreography in tests to fix race issues.