diff --git a/annot/pom.xml b/annot/pom.xml index 6b9c52b104..6e5f702e6d 100644 --- a/annot/pom.xml +++ b/annot/pom.xml @@ -56,6 +56,11 @@ json-schema-validator 2.0.0 + + com.github.spotbugs + spotbugs-annotations + 4.9.8 + diff --git a/annot/src/main/java/com/predic8/membrane/annot/bean/BeanFactory.java b/annot/src/main/java/com/predic8/membrane/annot/bean/BeanFactory.java index 0a4d50e928..c356070cd9 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/bean/BeanFactory.java +++ b/annot/src/main/java/com/predic8/membrane/annot/bean/BeanFactory.java @@ -15,8 +15,8 @@ package com.predic8.membrane.annot.bean; import com.fasterxml.jackson.databind.*; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import com.predic8.membrane.annot.util.*; -import com.predic8.membrane.annot.yaml.*; import org.jetbrains.annotations.*; import java.lang.reflect.*; diff --git a/annot/src/main/java/com/predic8/membrane/annot/beanregistry/AsyncBeanCollector.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/AsyncBeanCollector.java new file mode 100644 index 0000000000..a61c3a8729 --- /dev/null +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/AsyncBeanCollector.java @@ -0,0 +1,43 @@ +package com.predic8.membrane.annot.beanregistry; + +import javax.annotation.concurrent.GuardedBy; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingDeque; + +/** + * Thread-safe, asynchronous wrapper for {@link BeanCollector}. + */ +public class AsyncBeanCollector implements BeanCollector { + + private final BlockingQueue changeEvents = new LinkedBlockingDeque<>(); + private final BeanCollector delegate; + + @GuardedBy("this") + Thread t; + + public AsyncBeanCollector(BeanCollector delegate) { + this.delegate = delegate; + } + + @Override + public void handle(ChangeEvent changeEvent, boolean isLast) { + changeEvents.add(changeEvent); + } + + @Override + public synchronized void start() { + if (t != null) + return; + t = Thread.ofVirtual().start(() -> { + while (true) { + try { + ChangeEvent changeEvent = changeEvents.take(); + delegate.handle(changeEvent, changeEvents.isEmpty()); + } catch (InterruptedException e) { + break; + } + } + }); + } + +} diff --git a/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanCollector.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanCollector.java new file mode 100644 index 0000000000..4c7d51a515 --- /dev/null +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanCollector.java @@ -0,0 +1,41 @@ +package com.predic8.membrane.annot.beanregistry; + +import com.predic8.membrane.annot.Grammar; +import com.predic8.membrane.annot.yaml.GenericYamlParser; +import com.predic8.membrane.annot.yaml.WatchAction; + +import java.io.IOException; +import java.io.InputStream; +import java.util.List; + +/** + * This is the definition side of a {@link BeanRegistryImplementation}. You can start the bean registry + * and send it a series of change events. + */ +public interface BeanCollector { + /** + * Utility method to ingest a stream of YAML objects as a static configuration and then + * start the bean registry. + * @param yamls stream of YAML objects + * @param grammar the grammar to use for parsing + */ + default void parseYamls(InputStream yamls, Grammar grammar) throws IOException { + List bds = GenericYamlParser.parseMembraneResources(yamls, grammar); + for (int i = 0; i < bds.size(); i++) { + handle(new BeanDefinitionChanged(WatchAction.ADDED, bds.get(i)), i == bds.size() - 1); + } + handle(new StaticConfigurationLoaded(), true); + start(); + } + + /** + * @param changeEvent the change event + * @param isLast indicates whether this is the last change event for this batch of changes + */ + void handle(ChangeEvent changeEvent, boolean isLast); + + /** + * Starts the bean registry. + */ + void start(); +} diff --git a/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanContainer.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanContainer.java new file mode 100644 index 0000000000..2611e6c249 --- /dev/null +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanContainer.java @@ -0,0 +1,28 @@ +package com.predic8.membrane.annot.beanregistry; + +import com.fasterxml.jackson.databind.JsonNode; + +public class BeanContainer { + private final BeanDefinition definition; + /** + * Constructed bean after initialization. + */ + private volatile Object singleton; + + public BeanContainer(BeanDefinition definition) { + this.definition = definition; + } + + + public Object getSingleton() { + return singleton; + } + + public void setSingleton(Object singleton) { + this.singleton = singleton; + } + + public BeanDefinition getDefinition() { + return definition; + } +} diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanDefinition.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinition.java similarity index 72% rename from annot/src/main/java/com/predic8/membrane/annot/yaml/BeanDefinition.java rename to annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinition.java index f1df1feae3..4530a030c5 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanDefinition.java +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinition.java @@ -11,13 +11,10 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ -package com.predic8.membrane.annot.yaml; +package com.predic8.membrane.annot.beanregistry; import com.fasterxml.jackson.databind.JsonNode; -import com.predic8.membrane.annot.*; -import com.predic8.membrane.annot.bean.*; - -import java.util.*; +import com.predic8.membrane.annot.yaml.WatchAction; import static com.predic8.membrane.annot.yaml.WatchAction.*; @@ -29,19 +26,12 @@ public class BeanDefinition { private final String namespace; private final String uid; private final JsonNode node; - private final WatchAction action; private final String kind; - /** - * Constructed bean after initialization. - */ - private Object bean; - /** * Only called from K8S. */ - private BeanDefinition(WatchAction action, JsonNode node) { - this.action = action; + private BeanDefinition(JsonNode node) { this.node = node; JsonNode metadata = node.get("metadata"); var kind2 = node.get("kind").asText(); @@ -55,8 +45,8 @@ private BeanDefinition(WatchAction action, JsonNode node) { uid = metadata.get("uid").asText(); } - public static BeanDefinition create4Kubernetes(WatchAction action, JsonNode node) { - return new BeanDefinition(action, node); + public static BeanDefinitionChanged create4Kubernetes(WatchAction action, JsonNode node) { + return new BeanDefinitionChanged(action, new BeanDefinition(node)); } public BeanDefinition(String kind, String name, String namespace, String uid, JsonNode node) { @@ -65,17 +55,12 @@ public BeanDefinition(String kind, String name, String namespace, String uid, Js this.namespace = namespace; this.uid = uid; this.node = node; - this.action = ADDED; } public JsonNode getNode() { return node; } - public WatchAction getAction() { - return action; - } - public String getNamespace() { return namespace; } @@ -92,15 +77,6 @@ public String getKind() { return kind; } - public Object getBean() { - return bean; - } - - // TODO: Rest is immutable - can we make this also? - public void setBean(Object bean) { - this.bean = bean; - } - public String getScope() { JsonNode meta = node.get("metadata"); if (meta == null) @@ -124,15 +100,4 @@ public boolean isPrototype() { return PROTOTYPE.equals(getScope()); } - public boolean isDeleted() { - return action == DELETED; - } - - public boolean isModified() { - return action == MODIFIED; - } - - public boolean isAdded() { - return action == ADDED; - } } diff --git a/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinitionChanged.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinitionChanged.java new file mode 100644 index 0000000000..adc2eb3a52 --- /dev/null +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinitionChanged.java @@ -0,0 +1,11 @@ +package com.predic8.membrane.annot.beanregistry; + +import com.predic8.membrane.annot.yaml.WatchAction; + +/** + * Signals that a BeanDefinition has changed (=was added, modified, or deleted). + */ +public record BeanDefinitionChanged( + WatchAction action, + BeanDefinition bd) implements ChangeEvent { +} diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistry.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistry.java similarity index 85% rename from annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistry.java rename to annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistry.java index 0fbae75dcb..9d67e71322 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistry.java +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistry.java @@ -11,7 +11,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ -package com.predic8.membrane.annot.yaml; +package com.predic8.membrane.annot.beanregistry; import com.predic8.membrane.annot.*; @@ -23,10 +23,6 @@ public interface BeanRegistry { List getBeans(); - void registerBeanDefinitions(List beanDefinitions); - - void start(); - Grammar getGrammar(); } diff --git a/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistryImplementation.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistryImplementation.java new file mode 100644 index 0000000000..066426540a --- /dev/null +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistryImplementation.java @@ -0,0 +1,165 @@ +/* Copyright 2022 predic8 GmbH, www.predic8.com + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. */ +package com.predic8.membrane.annot.beanregistry; + +import com.predic8.membrane.annot.Grammar; +import com.predic8.membrane.annot.bean.BeanFactory; +import com.predic8.membrane.annot.yaml.BeanCacheObserver; +import com.predic8.membrane.annot.yaml.GenericYamlParser; +import com.predic8.membrane.annot.yaml.WatchAction; +import org.jetbrains.annotations.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +public class BeanRegistryImplementation implements BeanRegistry, BeanCollector { + + private static final Logger log = LoggerFactory.getLogger(BeanRegistryImplementation.class); + + private final BeanCacheObserver observer; + private final Grammar grammar; + + // uid -> bean + private final ConcurrentHashMap singletonBeans = new ConcurrentHashMap<>(); // Order is here not critical + + // uid -> bean container + private final Map bcs = new ConcurrentHashMap<>(); // Order is not critical. Order is determined by uidsToActivate + private final Set uidsToActivate = Collections.synchronizedSet(new LinkedHashSet<>()); // keeps order + + record UidAction(String uid, WatchAction action) {} + + public BeanRegistryImplementation(BeanCacheObserver observer, Grammar grammar) { + this.observer = observer; + this.grammar = grammar; + } + + private Object define(BeanDefinition bd) { + log.debug("defining bean: {}", bd.getNode()); + try { + if ("bean".equals(bd.getKind())) { + return new BeanFactory(this).create(bd.getNode().path("bean")); + } + return GenericYamlParser.readMembraneObject(bd.getKind(), + grammar, + bd.getNode(), + this); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Override + public void start() { + } + + @Override + public void handle(ChangeEvent changeEvent, boolean isLast) { + if (changeEvent instanceof StaticConfigurationLoaded) { + activationRun(); + observer.handleAsynchronousInitializationResult(uidsToActivate.isEmpty()); + } + if (changeEvent instanceof BeanDefinitionChanged(WatchAction action, BeanDefinition bd)) { + // Keep the latest BeanDefinition for all actions so activationRun + // can see both metadata and the action (including DELETED). + bcs.put(bd.getUid(), new BeanContainer(bd)); + + if (!bd.isComponent() && observer.isActivatable(bd)) { + uidsToActivate.add(new UidAction(bd.getUid(), action)); + } + + if (isLast) + activationRun(); + } + } + + private void activationRun() { + Set uidsToRemove = new HashSet<>(); + for (UidAction uidAction : uidsToActivate) { + BeanContainer bc = bcs.get(uidAction.uid); + try { + Object bean = define(bc.getDefinition()); + bc.setSingleton(bean); + + // e.g. inform router about new proxy + observer.handleBeanEvent(new BeanDefinitionChanged(uidAction.action, bc.getDefinition()), bean, getOldBean(uidAction.action, bc.getDefinition())); + + if (uidAction.action.isAdded() || uidAction.action.isModified()) + singletonBeans.put(bc.getDefinition().getUid(), bean); + if (uidAction.action.isDeleted()) { + singletonBeans.remove(bc.getDefinition().getUid()); + bcs.remove(bc.getDefinition().getUid()); + } + uidsToRemove.add(uidAction); + } catch (Exception e) { + log.error("Could not handle {} {}/{}", uidAction.action, + bc.getDefinition().getNamespace(), bc.getDefinition().getName(), e); + throw new RuntimeException(e); + } + } + for (UidAction uidAction : uidsToRemove) + uidsToActivate.remove(uidAction); + } + + private @Nullable Object getOldBean(WatchAction action, BeanDefinition bd) { + Object oldBean = null; + if (action.isModified() || action.isDeleted()) + oldBean = singletonBeans.get(bd.getUid()); + return oldBean; + } + + @Override + public Object resolveReference(String url) { + BeanContainer bc = getFirstByName(url).orElseThrow(() -> new RuntimeException("Reference %s not found".formatted(url))); + + boolean prototype = isPrototypeScope(bc.getDefinition()); + + if (!prototype && bc.getSingleton() != null) + return bc.getSingleton(); + + Object instance = define(bc.getDefinition()); + + if (!prototype) + bc.setSingleton(instance); + + return instance; + } + + private @NotNull Optional getFirstByName(String url) { + return bcs.values().stream().filter(bd -> url.equals(bd.getDefinition().getName())).findFirst(); + } + + @Override + public List getBeans() { + return bcs.values().stream().filter(bd -> !bd.getDefinition().isComponent()) + .map(BeanContainer::getSingleton) + .filter(Objects::nonNull) + .toList(); + } + + @Override + public Grammar getGrammar() { + return grammar; + } + + private static boolean isPrototypeScope(BeanDefinition bd) { + if (!bd.isBean()) + return bd.isPrototype(); + + return "PROTOTYPE".equalsIgnoreCase( + bd.getNode().path("bean").path("scope").asText("SINGLETON") + ); + } +} diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/ChangeEvent.java b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/ChangeEvent.java similarity index 75% rename from annot/src/main/java/com/predic8/membrane/annot/yaml/ChangeEvent.java rename to annot/src/main/java/com/predic8/membrane/annot/beanregistry/ChangeEvent.java index 8b0abc403c..6244138905 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/ChangeEvent.java +++ b/annot/src/main/java/com/predic8/membrane/annot/beanregistry/ChangeEvent.java @@ -12,12 +12,13 @@ See the License for the specific language governing permissions and limitations under the License. */ -package com.predic8.membrane.annot.yaml; +package com.predic8.membrane.annot.beanregistry; -sealed interface ChangeEvent permits BeanDefinitionChanged, StaticConfigurationLoaded { -} +import com.predic8.membrane.annot.yaml.WatchAction; + +import static com.predic8.membrane.annot.yaml.WatchAction.*; -record BeanDefinitionChanged(BeanDefinition bd) implements ChangeEvent { +public sealed interface ChangeEvent permits BeanDefinitionChanged, StaticConfigurationLoaded { } /** diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanCacheObserver.java b/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanCacheObserver.java index 39175601bd..1fdb696581 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanCacheObserver.java +++ b/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanCacheObserver.java @@ -14,6 +14,10 @@ package com.predic8.membrane.annot.yaml; +import com.predic8.membrane.annot.beanregistry.BeanDefinition; +import com.predic8.membrane.annot.beanregistry.BeanDefinitionChanged; +import com.predic8.membrane.annot.beanregistry.BeanRegistryImplementation; + import java.io.IOException; /** @@ -34,16 +38,13 @@ public interface BeanCacheObserver { /** * Called for an add/modify/delete event of a bean. * - * @param bd the bean definition + * @param bd the bean definition changed event * @param bean the current instance (on ADD/MODIFY) or {@code null} (on DELETE) * @param oldBean the previous instance (on MODIFY) or {@code null} * @throws IOException if handling the event performs I/O and it fails * - * - * TODO: Make event visible: enum and add to signature? - * */ - void handleBeanEvent(BeanDefinition bd, Object bean, Object oldBean) throws IOException; + void handleBeanEvent(BeanDefinitionChanged bd, Object bean, Object oldBean) throws IOException; /** * Whether beans of the given definition should be considered activatable/usable diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistryImplementation.java b/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistryImplementation.java deleted file mode 100644 index 7a3edddade..0000000000 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistryImplementation.java +++ /dev/null @@ -1,204 +0,0 @@ -/* Copyright 2022 predic8 GmbH, www.predic8.com - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. */ -package com.predic8.membrane.annot.yaml; - -import com.fasterxml.jackson.databind.JsonNode; -import com.predic8.membrane.annot.Grammar; -import com.predic8.membrane.annot.bean.BeanFactory; -import org.jetbrains.annotations.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.*; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.LinkedBlockingDeque; - -import static com.predic8.membrane.annot.yaml.BeanDefinition.create4Kubernetes; -import static com.predic8.membrane.annot.yaml.WatchAction.*; - -public class BeanRegistryImplementation implements BeanRegistry { - - private static final Logger log = LoggerFactory.getLogger(BeanRegistryImplementation.class); - - private final BeanCacheObserver observer; - private final Grammar grammar; - - /** - * TODO Rename give meaningful name - */ - private final ConcurrentHashMap uuidMap = new ConcurrentHashMap<>(); // Order is here not critical - - private final BlockingQueue changeEvents = new LinkedBlockingDeque<>(); - - // uid -> bean definition - private final Map bds = new ConcurrentHashMap<>(); // Order is not critical. Order is determined by uidsToActivate - private final Set uidsToActivate = new LinkedHashSet<>(); // Provides order - - public BeanRegistryImplementation(BeanCacheObserver observer, Grammar grammar) { - this.observer = observer; - this.grammar = grammar; - } - - public void registerBeanDefinitions(List bds) { - bds.forEach(bd -> handle(ADDED, bd)); - fireConfigurationLoaded(); // Only put event in the queue - } - - public void start() { - while (!changeEvents.isEmpty()) { - try { - ChangeEvent changeEvent = changeEvents.take(); - if (changeEvent instanceof StaticConfigurationLoaded) { - activationRun(); - observer.handleAsynchronousInitializationResult(uidsToActivate.isEmpty()); - continue; - } - if (changeEvent instanceof BeanDefinitionChanged(BeanDefinition bd)) { - handle(bd); - } - } catch (InterruptedException e) { - break; - } - } - } - - private Object define(BeanDefinition bd) { - log.debug("defining bean: {}", bd.getNode()); - try { - if ("bean".equals(bd.getKind())) { - return new BeanFactory(this).create(bd.getNode().path("bean")); - } - return GenericYamlParser.readMembraneObject(bd.getKind(), - grammar, - bd.getNode(), - this); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - - /** - * May be called from multiple threads. - */ - public void handle(WatchAction action, JsonNode node) { - changeEvents.add(new BeanDefinitionChanged(create4Kubernetes(action, node))); - } - - /** - * May be called from multiple threads. - * TODO remove action? - */ - public void handle(WatchAction action, BeanDefinition bd) { - changeEvents.add(new BeanDefinitionChanged(bd)); - } - - /** - * Signals that all {@link ChangeEvent}s have been passed to {@link #handle(WatchAction, JsonNode)} which originate from - * static configuration (e.g. a file). - */ - public void fireConfigurationLoaded() { - changeEvents.add(new StaticConfigurationLoaded()); - } - - void handle(BeanDefinition bd) { - // Keep the latest BeanDefinition for all actions so activationRun - // can see both metadata and the action (including DELETED). - bds.put(bd.getUid(), bd); - - if (!bd.isComponent() && observer.isActivatable(bd)) { - uidsToActivate.add(bd.getUid()); - } - - if (changeEvents.isEmpty()) - activationRun(); - } - - private void activationRun() { - Set uidsToRemove = new HashSet<>(); - for (String uid1 : uidsToActivate) { - BeanDefinition bd = bds.get(uid1); - try { - Object bean = define(bd); - bd.setBean(bean); - - // e.g. inform router about new proxy - observer.handleBeanEvent(bd, bean, getOldBean(bd)); - - if (bd.isAdded() || bd.isModified()) - uuidMap.put(bd.getUid(), bean); - if (bd.isDeleted()) { - uuidMap.remove(bd.getUid()); - bds.remove(bd.getUid()); - } - uidsToRemove.add(bd.getUid()); - } catch (Exception e) { - log.error("Could not handle {} {}/{}", bd.getAction(), bd.getNamespace(), bd.getName(), e); - throw new RuntimeException(e); - } - } - for (String uid : uidsToRemove) - uidsToActivate.remove(uid); - } - - private @Nullable Object getOldBean(BeanDefinition bd) { - Object oldBean = null; - if (bd.isModified() || bd.isDeleted()) - oldBean = uuidMap.get(bd.getUid()); - return oldBean; - } - - @Override - public Object resolveReference(String url) { - BeanDefinition bd = getFirstByName(url).orElseThrow(() -> new RuntimeException("Reference %s not found".formatted(url))); - - boolean prototype = isPrototypeScope(bd); - - if (!prototype && bd.getBean() != null) - return bd.getBean(); - - Object instance = define(bd); - - if (!prototype) - bd.setBean(instance); - - return instance; - } - - private @NotNull Optional getFirstByName(String url) { - return bds.values().stream().filter(bd -> url.equals(bd.getName())).findFirst(); - } - - @Override - public List getBeans() { - return bds.values().stream().filter(bd -> !bd.isComponent()) - .map(BeanDefinition::getBean) - .filter(Objects::nonNull) - .toList(); - } - - @Override - public Grammar getGrammar() { - return grammar; - } - - private static boolean isPrototypeScope(BeanDefinition bd) { - if (!bd.isBean()) - return bd.isPrototype(); - - return "PROTOTYPE".equalsIgnoreCase( - bd.getNode().path("bean").path("scope").asText("SINGLETON") - ); - } -} diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/GenericYamlParser.java b/annot/src/main/java/com/predic8/membrane/annot/yaml/GenericYamlParser.java index a02006c744..571df26c3a 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/GenericYamlParser.java +++ b/annot/src/main/java/com/predic8/membrane/annot/yaml/GenericYamlParser.java @@ -20,6 +20,9 @@ import com.networknt.schema.*; import com.networknt.schema.Error; import com.predic8.membrane.annot.*; +import com.predic8.membrane.annot.beanregistry.BeanDefinition; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistryImplementation; import org.jetbrains.annotations.*; import org.slf4j.*; @@ -88,18 +91,16 @@ public GenericYamlParser(Grammar grammar, String yaml) throws IOException { } /** - * Entry point used by the runtime to consume a YAML stream and turn it into - * a {@link BeanRegistry} that the router can work with. + * Entry point used by the runtime to consume a YAML stream. *
    *
  • Reads the entire stream as UTF-8.
    • *
  • Splits multi-document YAML ("---" separators).
    • *
  • Validates each document against the JSON Schema provided by {@code grammar}.
    • *
  • Emits helpful line/column locations for malformed multi-document input.
    • *
- * The returned registry is fully populated and {@link BeanRegistryImplementation#fireConfigurationLoaded()} has been called. * @param resource the input stream to parse. The method takes care of closing the stream. * @param grammar the grammar to use for type resolution and schema location - * @return the bean registry + * @return list of parsed bean definitions */ public static List parseMembraneResources(@NotNull InputStream resource, Grammar grammar) throws IOException { try (resource) { diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/ParsingContext.java b/annot/src/main/java/com/predic8/membrane/annot/yaml/ParsingContext.java index b49441cc56..9dd2636173 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/ParsingContext.java +++ b/annot/src/main/java/com/predic8/membrane/annot/yaml/ParsingContext.java @@ -15,6 +15,7 @@ package com.predic8.membrane.annot.yaml; import com.predic8.membrane.annot.*; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; /** * Immutable parsing state passed down while traversing YAML. diff --git a/annot/src/main/java/com/predic8/membrane/annot/yaml/WatchAction.java b/annot/src/main/java/com/predic8/membrane/annot/yaml/WatchAction.java index 17abc39813..785e567ddb 100644 --- a/annot/src/main/java/com/predic8/membrane/annot/yaml/WatchAction.java +++ b/annot/src/main/java/com/predic8/membrane/annot/yaml/WatchAction.java @@ -14,5 +14,18 @@ package com.predic8.membrane.annot.yaml; public enum WatchAction { - ADDED, MODIFIED, DELETED + ADDED, MODIFIED, DELETED; + + public boolean isDeleted() { + return this == DELETED; + } + + public boolean isModified() { + return this == MODIFIED; + } + + public boolean isAdded() { + return this == ADDED; + } + } diff --git a/annot/src/test/java/com/predic8/membrane/annot/YAMLBeanParsingTest.java b/annot/src/test/java/com/predic8/membrane/annot/YAMLBeanParsingTest.java index 0f59db3fe0..8a77589ca7 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/YAMLBeanParsingTest.java +++ b/annot/src/test/java/com/predic8/membrane/annot/YAMLBeanParsingTest.java @@ -14,7 +14,7 @@ package com.predic8.membrane.annot; import com.predic8.membrane.annot.util.CompilerHelper; -import com.predic8.membrane.annot.yaml.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import com.predic8.membrane.annot.yaml.YamlSchemaValidationException; import org.junit.jupiter.api.Test; diff --git a/annot/src/test/java/com/predic8/membrane/annot/YAMLComponentsParsingTest.java b/annot/src/test/java/com/predic8/membrane/annot/YAMLComponentsParsingTest.java index 7c01c54bca..e01dd43e25 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/YAMLComponentsParsingTest.java +++ b/annot/src/test/java/com/predic8/membrane/annot/YAMLComponentsParsingTest.java @@ -15,7 +15,7 @@ package com.predic8.membrane.annot; import com.predic8.membrane.annot.util.CompilerHelper; -import com.predic8.membrane.annot.yaml.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import com.predic8.membrane.annot.yaml.YamlSchemaValidationException; import org.junit.jupiter.api.Test; diff --git a/annot/src/test/java/com/predic8/membrane/annot/util/CompilerHelper.java b/annot/src/test/java/com/predic8/membrane/annot/util/CompilerHelper.java index 4f417bbe34..224b8ffb2f 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/util/CompilerHelper.java +++ b/annot/src/test/java/com/predic8/membrane/annot/util/CompilerHelper.java @@ -13,7 +13,7 @@ limitations under the License. */ package com.predic8.membrane.annot.util; -import com.predic8.membrane.annot.yaml.*; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import org.hamcrest.*; import org.hamcrest.collection.*; import org.jetbrains.annotations.*; diff --git a/annot/src/test/java/com/predic8/membrane/annot/util/InMemoryClassLoader.java b/annot/src/test/java/com/predic8/membrane/annot/util/InMemoryClassLoader.java index c8029b8d0b..9543816bd1 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/util/InMemoryClassLoader.java +++ b/annot/src/test/java/com/predic8/membrane/annot/util/InMemoryClassLoader.java @@ -111,7 +111,7 @@ protected Class findClass(String name) throws ClassNotFoundException { private boolean delegateToRootClassLoader(String name) { return name.startsWith("java.") || name.startsWith("javax.") || name.startsWith("org.xml.sax") || name.startsWith("org.w3c.dom") - || name.equals("com.predic8.membrane.annot.yaml.BeanRegistry"); + || name.equals("com.predic8.membrane.annot.beanregistry.BeanRegistry"); } @Override diff --git a/annot/src/test/java/com/predic8/membrane/annot/util/StructureAssertionUtil.java b/annot/src/test/java/com/predic8/membrane/annot/util/StructureAssertionUtil.java index eaf451eaab..e31387d186 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/util/StructureAssertionUtil.java +++ b/annot/src/test/java/com/predic8/membrane/annot/util/StructureAssertionUtil.java @@ -14,10 +14,9 @@ package com.predic8.membrane.annot.util; -import com.predic8.membrane.annot.yaml.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import org.junit.jupiter.api.Assertions; -import java.lang.reflect.Method; import java.util.List; import static org.junit.jupiter.api.Assertions.*; diff --git a/annot/src/test/java/com/predic8/membrane/annot/util/YamlParser.java b/annot/src/test/java/com/predic8/membrane/annot/util/YamlParser.java index aa6a5e6629..72eda7c984 100644 --- a/annot/src/test/java/com/predic8/membrane/annot/util/YamlParser.java +++ b/annot/src/test/java/com/predic8/membrane/annot/util/YamlParser.java @@ -15,6 +15,7 @@ package com.predic8.membrane.annot.util; import com.predic8.membrane.annot.Grammar; +import com.predic8.membrane.annot.beanregistry.*; import com.predic8.membrane.annot.yaml.*; import org.jetbrains.annotations.*; @@ -47,10 +48,9 @@ public YamlParser(String resourceName) throws ClassNotFoundException, NoSuchMeth String normalized = resourceName.startsWith("/") ? resourceName.substring(1) : resourceName; - beanRegistry = new BeanRegistryImplementation(getLatchObserver(cdl),generator); - beanRegistry.registerBeanDefinitions(GenericYamlParser.parseMembraneResources( - requireNonNull(cl.getResourceAsStream(normalized)), generator)); - beanRegistry.start(); + BeanRegistryImplementation impl = new BeanRegistryImplementation(getLatchObserver(cdl), generator); + impl.parseYamls(requireNonNull(cl.getResourceAsStream(normalized)), generator); + beanRegistry = impl; cdl.await(); } @@ -82,7 +82,7 @@ public void handleAsynchronousInitializationResult(boolean empty) { } @Override - public void handleBeanEvent(BeanDefinition bd, Object bean, Object oldBean) { + public void handleBeanEvent(BeanDefinitionChanged bdc, Object bean, Object oldBean) { } diff --git a/core/src/main/java/com/predic8/membrane/core/Router.java b/core/src/main/java/com/predic8/membrane/core/Router.java index edff0f75e5..69ea9d0c19 100644 --- a/core/src/main/java/com/predic8/membrane/core/Router.java +++ b/core/src/main/java/com/predic8/membrane/core/Router.java @@ -15,6 +15,8 @@ package com.predic8.membrane.core; import com.predic8.membrane.annot.*; +import com.predic8.membrane.annot.beanregistry.BeanDefinition; +import com.predic8.membrane.annot.beanregistry.BeanDefinitionChanged; import com.predic8.membrane.annot.yaml.*; import com.predic8.membrane.core.RuleManager.*; import com.predic8.membrane.core.config.spring.*; @@ -649,13 +651,13 @@ public void handleAsynchronousInitializationResult(boolean success) { } @Override - public void handleBeanEvent(BeanDefinition bd, Object bean, Object oldBean) throws IOException { + public void handleBeanEvent(BeanDefinitionChanged bdc, Object bean, Object oldBean) throws IOException { if (!(bean instanceof Proxy newProxy)) { throw new IllegalArgumentException("Bean must be a Proxy instance, but got: " + bean.getClass().getName()); } if (newProxy.getName() == null) - newProxy.setName(bd.getName()); + newProxy.setName(bdc.bd().getName()); try { newProxy.init(this); @@ -666,11 +668,11 @@ public void handleBeanEvent(BeanDefinition bd, Object bean, Object oldBean) thro throw new RuntimeException("Could not init rule.", e); } - if (bd.getAction() == WatchAction.ADDED) + if (bdc.action().isAdded()) add(newProxy); - else if (bd.getAction() == WatchAction.DELETED) + else if (bdc.action().isDeleted()) getRuleManager().removeRule((Proxy) oldBean); - else if (bd.getAction() == WatchAction.MODIFIED) + else if (bdc.action().isModified()) getRuleManager().replaceRule((Proxy) oldBean, newProxy); } diff --git a/core/src/main/java/com/predic8/membrane/core/cli/RouterCLI.java b/core/src/main/java/com/predic8/membrane/core/cli/RouterCLI.java index 3034d27791..7f41263e94 100644 --- a/core/src/main/java/com/predic8/membrane/core/cli/RouterCLI.java +++ b/core/src/main/java/com/predic8/membrane/core/cli/RouterCLI.java @@ -14,9 +14,10 @@ package com.predic8.membrane.core.cli; -import com.predic8.membrane.annot.yaml.*; +import com.predic8.membrane.annot.beanregistry.BeanCollector; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistryImplementation; import com.predic8.membrane.core.*; -import com.predic8.membrane.core.cli.util.*; import com.predic8.membrane.core.config.spring.*; import com.predic8.membrane.core.exceptions.*; import com.predic8.membrane.core.openapi.serviceproxy.*; @@ -168,9 +169,8 @@ private static Router initRouterByYAML(String location) throws Exception { router.start(); GrammarAutoGenerated grammar = new GrammarAutoGenerated(); - BeanRegistry registry = new BeanRegistryImplementation(router, grammar); - registry.registerBeanDefinitions(parseMembraneResources(router.getResolverMap().resolve(location), grammar)); - registry.start(); + BeanCollector collector = new BeanRegistryImplementation(router, grammar); + collector.parseYamls(router.getResolverMap().resolve(location), grammar); return router; } diff --git a/core/src/main/java/com/predic8/membrane/core/config/spring/k8s/Envelope.java b/core/src/main/java/com/predic8/membrane/core/config/spring/k8s/Envelope.java index d6e9a4879c..6a4f34bdac 100644 --- a/core/src/main/java/com/predic8/membrane/core/config/spring/k8s/Envelope.java +++ b/core/src/main/java/com/predic8/membrane/core/config/spring/k8s/Envelope.java @@ -16,7 +16,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.predic8.membrane.annot.Grammar; import com.predic8.membrane.core.config.spring.GrammarAutoGenerated; -import com.predic8.membrane.annot.yaml.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import java.util.*; diff --git a/core/src/main/java/com/predic8/membrane/core/kubernetes/KubernetesWatcher.java b/core/src/main/java/com/predic8/membrane/core/kubernetes/KubernetesWatcher.java index 0d2171184c..ce453e845f 100644 --- a/core/src/main/java/com/predic8/membrane/core/kubernetes/KubernetesWatcher.java +++ b/core/src/main/java/com/predic8/membrane/core/kubernetes/KubernetesWatcher.java @@ -14,6 +14,10 @@ package com.predic8.membrane.core.kubernetes; import com.fasterxml.jackson.databind.*; +import com.predic8.membrane.annot.beanregistry.AsyncBeanCollector; +import com.predic8.membrane.annot.beanregistry.BeanCollector; +import com.predic8.membrane.annot.beanregistry.BeanDefinition; +import com.predic8.membrane.annot.beanregistry.BeanRegistryImplementation; import com.predic8.membrane.annot.yaml.*; import com.predic8.membrane.core.*; import com.predic8.membrane.core.config.spring.*; @@ -34,14 +38,17 @@ public class KubernetesWatcher { private static final Logger LOG = LoggerFactory.getLogger(KubernetesWatcher.class); private final Router router; - private final BeanRegistryImplementation beanRegistry; + private final BeanCollector beanRegistry; + private final GrammarAutoGenerated grammar; private KubernetesClient client; private ExecutorService executors; private final ConcurrentHashMap watches = new ConcurrentHashMap<>(); public KubernetesWatcher(Router router) { this.router = router; - this.beanRegistry = new BeanRegistryImplementation(router, new GrammarAutoGenerated()); + this.grammar = new GrammarAutoGenerated(); + var br = new BeanRegistryImplementation(router, grammar); + this.beanRegistry = new AsyncBeanCollector(br); } public void start() { @@ -54,7 +61,7 @@ public void start() { client = getClient(); - List crds = beanRegistry.getGrammar().getCrdSingularNames(); + List crds = grammar.getCrdSingularNames(); if (kvi.get().getResourcesList().size() > 0) crds = crds.stream().filter(s -> kvi.get().getResourcesList().contains(s)).toList(); if (crds.size() > 0) @@ -103,7 +110,7 @@ public void onEvent(WatchAction action, JsonNode node) { node.get("metadata").get("namespace").asText(), node.get("metadata").get("name").asText())); - beanRegistry.handle(action, node.get("spec")); + beanRegistry.handle(BeanDefinition.create4Kubernetes(action, node.get("spec")), true); } catch (Exception e) { e.printStackTrace(); } diff --git a/core/src/test/java/com/predic8/membrane/core/config/spring/k8s/EnvelopeTest.java b/core/src/test/java/com/predic8/membrane/core/config/spring/k8s/EnvelopeTest.java index d928776500..2f6891835e 100644 --- a/core/src/test/java/com/predic8/membrane/core/config/spring/k8s/EnvelopeTest.java +++ b/core/src/test/java/com/predic8/membrane/core/config/spring/k8s/EnvelopeTest.java @@ -24,7 +24,7 @@ import com.predic8.membrane.core.interceptor.rewrite.RewriteInterceptor; import com.predic8.membrane.core.interceptor.templating.TemplateInterceptor; import com.predic8.membrane.core.interceptor.flow.ReturnInterceptor; -import com.predic8.membrane.annot.yaml.BeanRegistry; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import com.predic8.membrane.core.openapi.serviceproxy.APIProxy; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; diff --git a/core/src/test/java/com/predic8/membrane/core/kubernetes/GenericYamlParserTest.java b/core/src/test/java/com/predic8/membrane/core/kubernetes/GenericYamlParserTest.java index a4fe5b72a6..b9553c466d 100644 --- a/core/src/test/java/com/predic8/membrane/core/kubernetes/GenericYamlParserTest.java +++ b/core/src/test/java/com/predic8/membrane/core/kubernetes/GenericYamlParserTest.java @@ -15,6 +15,8 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; import com.predic8.membrane.annot.Grammar; +import com.predic8.membrane.annot.beanregistry.BeanDefinition; +import com.predic8.membrane.annot.beanregistry.BeanRegistry; import com.predic8.membrane.annot.yaml.*; import com.predic8.membrane.core.config.spring.GrammarAutoGenerated; import com.predic8.membrane.core.interceptor.authentication.BasicAuthenticationInterceptor;