split BeanRegistry interface (#2488)

* split BeanRegistry interface

* fixes

* fixed thread-safety

Author: rrayst

annot/pom.xml

@@ -56,6 +56,11 @@
             <artifactId>json-schema-validator</artifactId>
             <version>2.0.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.github.spotbugs</groupId>
+            <artifactId>spotbugs-annotations</artifactId>
+            <version>4.9.8</version>
+        </dependency>
 
         <!-- Test dependencies -->
         <dependency>

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.*;

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<ChangeEvent> 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;
+                }
+            }
+        });
+    }
+
+}

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<BeanDefinition> 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();
+}

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;
+    }
+}

…/membrane/annot/yaml/BeanDefinition.java …e/annot/beanregistry/BeanDefinition.javaannot/src/main/java/com/predic8/membrane/annot/yaml/BeanDefinition.java renamed to annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanDefinition.java annot/src/main/java/com/predic8/membrane/annot/yaml/BeanDefinition.java renamed to 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;
-    }
 }

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 {
+}

…c8/membrane/annot/yaml/BeanRegistry.java …ane/annot/beanregistry/BeanRegistry.javaannot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistry.java renamed to annot/src/main/java/com/predic8/membrane/annot/beanregistry/BeanRegistry.java annot/src/main/java/com/predic8/membrane/annot/yaml/BeanRegistry.java renamed to 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<Object> getBeans();
 
-    void registerBeanDefinitions(List<BeanDefinition> beanDefinitions);
-
-    void start();
-
     Grammar getGrammar();
 
 }