feat: add full-stack signals base

packages/java/endpoint/src/main/java/com/vaadin/hilla/EndpointController.java

@@ -21,6 +21,7 @@
 import java.util.Optional;
 import java.util.TreeMap;
 
+import com.vaadin.experimental.FeatureFlags;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationContext;
@@ -262,6 +263,8 @@ private void registerEndpointsFromApiDefinition(
                                 .or(() -> Optional
                                         .ofNullable(tag.get("x-class-name"))
                                         .map(JsonNode::asText)
+                                        .filter(className -> !("com.vaadin.hilla.signals.core.SignalsHandler"
+                                                .equals(className)))
                                         .map(this::instantiateEndpointByClassName))
                                 .ifPresent(endpointRegistry::registerEndpoint);
                     });

packages/java/endpoint/src/main/java/com/vaadin/hilla/EndpointControllerConfiguration.java

@@ -18,13 +18,16 @@
 
 import java.lang.reflect.Method;
 
+import com.vaadin.hilla.signals.config.SignalsConfiguration;
+import com.vaadin.hilla.signals.core.SignalsRegistry;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.web.servlet.WebMvcRegistrations;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.web.servlet.mvc.method.RequestMappingInfo;
 import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping;
 import org.springframework.web.util.pattern.PathPatternParser;
@@ -44,6 +47,7 @@
 @Configuration
 public class EndpointControllerConfiguration {
     private final EndpointProperties endpointProperties;
+    private final SignalsRegistry signalsRegistry;
 
     /**
      * Initializes the endpoint configuration.
@@ -52,8 +56,10 @@ public class EndpointControllerConfiguration {
      *            Hilla endpoint properties
      */
     public EndpointControllerConfiguration(
-            EndpointProperties endpointProperties) {
+            EndpointProperties endpointProperties,
+            @Autowired(required = false) SignalsRegistry signalsRegistry) {
         this.endpointProperties = endpointProperties;
+        this.signalsRegistry = signalsRegistry;
     }
 
     /**
@@ -118,9 +124,11 @@ CsrfChecker csrfChecker(ServletContext servletContext) {
     EndpointInvoker endpointInvoker(ApplicationContext applicationContext,
             @Autowired(required = false) @Qualifier(EndpointController.ENDPOINT_MAPPER_FACTORY_BEAN_QUALIFIER) JacksonObjectMapperFactory endpointMapperFactory,
             ExplicitNullableTypeChecker explicitNullableTypeChecker,
-            ServletContext servletContext, EndpointRegistry endpointRegistry) {
+            ServletContext servletContext,
+            @Autowired(required = false) EndpointRegistry endpointRegistry) {
         return new EndpointInvoker(applicationContext, endpointMapperFactory,
-                explicitNullableTypeChecker, servletContext, endpointRegistry);
+                explicitNullableTypeChecker, servletContext, endpointRegistry,
+                signalsRegistry);
     }
 
     /**
@@ -237,9 +245,9 @@ private RequestMappingInfo prependEndpointPrefixUrl(
     }
 
     /**
-     * Can re-generate the TypeScipt code.
+     * Can re-generate the TypeScript code.
      *
-     * @param context
+     * @param servletContext
      *            the servlet context
      * @param endpointController
      *            the endpoint controller

packages/java/endpoint/src/main/java/com/vaadin/hilla/EndpointInvoker.java

@@ -20,6 +20,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.googlecode.gentyref.GenericTypeReflector;
+import com.vaadin.experimental.FeatureFlags;
+import com.vaadin.flow.server.VaadinService;
 import com.vaadin.flow.server.VaadinServletContext;
 import com.vaadin.hilla.EndpointInvocationException.EndpointAccessDeniedException;
 import com.vaadin.hilla.EndpointInvocationException.EndpointBadRequestException;
@@ -32,6 +34,8 @@
 import com.vaadin.hilla.exception.EndpointValidationException;
 import com.vaadin.hilla.exception.EndpointValidationException.ValidationErrorData;
 import com.vaadin.hilla.parser.jackson.JacksonObjectMapperFactory;
+import com.vaadin.hilla.signals.core.SignalQueue;
+import com.vaadin.hilla.signals.core.SignalsRegistry;
 import jakarta.servlet.ServletContext;
 import jakarta.validation.ConstraintViolation;
 import jakarta.validation.Validation;
@@ -80,6 +84,7 @@ public class EndpointInvoker {
     private final ExplicitNullableTypeChecker explicitNullableTypeChecker;
     private final ServletContext servletContext;
     private final Validator validator;
+    private final SignalsRegistry signalsRegistry;
 
     /**
      * Creates an instance of this bean.
@@ -103,7 +108,8 @@ public class EndpointInvoker {
     public EndpointInvoker(ApplicationContext applicationContext,
             JacksonObjectMapperFactory endpointMapperFactory,
             ExplicitNullableTypeChecker explicitNullableTypeChecker,
-            ServletContext servletContext, EndpointRegistry endpointRegistry) {
+            ServletContext servletContext, EndpointRegistry endpointRegistry,
+            SignalsRegistry signalsRegistry) {
         this.applicationContext = applicationContext;
         this.servletContext = servletContext;
         this.endpointMapper = endpointMapperFactory != null
@@ -115,6 +121,7 @@ public EndpointInvoker(ApplicationContext applicationContext,
         }
         this.explicitNullableTypeChecker = explicitNullableTypeChecker;
         this.endpointRegistry = endpointRegistry;
+        this.signalsRegistry = signalsRegistry;
 
         Validator validator = null;
         try {
@@ -481,6 +488,31 @@ private Object invokeVaadinEndpointMethod(String endpointName,
             throw new EndpointInternalException(errorMessage);
         }
 
+        if (returnValue instanceof SignalQueue<?>) {
+            if (signalsRegistry == null) {
+                throw new IllegalStateException(
+                        """
+                                Signals registry is not available, cannot register signal.
+                                Please make sure you have enabled the Full Stack Signals
+                                feature preview flag either through the Vaadin Copilot's
+                                Features panel, or by manually setting the
+                                'com.vaadin.experimental.fullstackSignals=true' in
+                                'src/main/resources/vaadin-featureflags.properties'.
+                                """
+                                .stripLeading());
+            }
+            if (signalsRegistry
+                    .contains(((SignalQueue<?>) returnValue).getId())) {
+                getLogger().debug(
+                        "Signal already registered before. Ignoring the registration. Endpoint: '{}', method: '{}'",
+                        endpointName, methodName);
+            } else {
+                signalsRegistry.register((SignalQueue<?>) returnValue);
+                getLogger().debug("Registered signal as a result of calling {}",
+                        methodName);
+            }
+        }
+
         return returnValue;
     }
 

packages/java/endpoint/src/main/java/com/vaadin/hilla/signals/NumberSignal.java

@@ -0,0 +1,12 @@
+package com.vaadin.hilla.signals;
+
+import com.fasterxml.jackson.databind.node.IntNode;
+import com.vaadin.hilla.signals.core.SignalQueue;
+
+public class NumberSignal extends SignalQueue<IntNode> {
+
+    public NumberSignal(int defaultValue) {
+        super(IntNode.valueOf(defaultValue));
+    }
+
+}

packages/java/endpoint/src/main/java/com/vaadin/hilla/signals/config/SignalsConfiguration.java

@@ -0,0 +1,39 @@
+package com.vaadin.hilla.signals.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.vaadin.hilla.ConditionalOnFeatureFlag;
+import com.vaadin.hilla.signals.core.SignalsHandler;
+import com.vaadin.hilla.signals.core.SignalsRegistry;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class SignalsConfiguration {
+
+    private SignalsRegistry signalsRegistry;
+    private SignalsHandler signalsHandler;
+    private final ObjectMapper objectMapper;
+
+    public SignalsConfiguration(ObjectMapper objectMapper) {
+        this.objectMapper = objectMapper;
+    }
+
+    @ConditionalOnFeatureFlag("fullstackSignals")
+    @Bean
+    public SignalsRegistry signalsRegistry() {
+        if (signalsRegistry == null) {
+            signalsRegistry = new SignalsRegistry();
+        }
+        return signalsRegistry;
+    }
+
+    @ConditionalOnFeatureFlag("fullstackSignals")
+    @Bean
+    public SignalsHandler signalsHandler() {
+        if (signalsHandler == null) {
+            signalsHandler = new SignalsHandler(signalsRegistry(),
+                    objectMapper);
+        }
+        return signalsHandler;
+    }
+}

packages/java/endpoint/src/main/java/com/vaadin/hilla/signals/core/EventQueue.java

@@ -0,0 +1,121 @@
+package com.vaadin.hilla.signals.core;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Sinks.Many;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.locks.ReentrantLock;
+
+public abstract class EventQueue<T extends StateEvent> {
+    private final ReentrantLock lock = new ReentrantLock();
+
+    public static final UUID ROOT = UUID
+            .fromString("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF");
+
+    private static class Entry<T extends StateEvent> {
+        private final T value;
+        private Entry<T> next;
+
+        private Entry(T value) {
+            this.value = value;
+        }
+    }
+
+    private Entry<T> head;
+    private Entry<T> tail;
+    private final Map<UUID, Entry<T>> idToEntry = new HashMap<>();
+
+    private final Set<Many<T>> subscribers = new HashSet<>();
+
+    public Flux<T> subscribe(UUID continueFrom) {
+        System.out.println("Continue from " + continueFrom);
+        Many<T> sink = Sinks.many().multicast().onBackpressureBuffer();
+
+        return sink.asFlux().doOnSubscribe(ignore -> {
+            System.out.println("New Flux subscription");
+
+            lock.lock();
+            try {
+                Entry<T> entry;
+                if (continueFrom != null
+                        && (entry = idToEntry.get(continueFrom)) != null) {
+                    entry = entry.next;
+                    // TODO maybe some heuristic to determine whether it would
+                    // be more efficient to restart from a snapshot instead of
+                    // replaying lots of events?
+                    while (entry != null) {
+                        System.out.println("Replay " + entry.value.getId());
+                        sink.tryEmitNext(entry.value);
+                        entry = entry.next;
+                    }
+                    ;
+                } else {
+                    T snapshot = createSnapshot();
+                    if (snapshot != null) {
+                        sink.tryEmitNext(snapshot);
+                    }
+                }
+
+                subscribers.add(sink);
+            } finally {
+                lock.unlock();
+            }
+        }).doFinally(ignore -> {
+            System.out.println("doFinally");
+            lock.lock();
+            try {
+                subscribers.remove(sink);
+            } finally {
+                lock.unlock();
+            }
+        });
+    }
+
+    public void submit(T event) {
+        // Thread.ofVirtual().start(() -> append(event));
+        append(event);
+    }
+
+    private void append(T event) {
+        lock.lock();
+        try {
+            processEvent(event);
+            Entry<T> entry = new Entry<>(event);
+
+            // Add to linked list
+            idToEntry.put(event.getId(), entry);
+            if (head == null) {
+                head = tail = entry;
+            } else {
+                tail.next = tail = entry;
+            }
+
+            // Truncate list
+            // TODO configurable or dynamic limit?
+            if (idToEntry.size() > 100) {
+                Entry<T> removed = idToEntry.remove(head.value.getId());
+                head = removed.next;
+            }
+
+            // Notify subscribers
+            subscribers.removeIf(sink -> {
+                boolean failure = sink.tryEmitNext(event).isFailure();
+                if (failure) {
+                    System.out.println("Failed push");
+                }
+                return failure;
+            });
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    protected abstract void processEvent(T newEvent);
+
+    protected abstract T createSnapshot();
+}

packages/java/endpoint/src/main/java/com/vaadin/hilla/signals/core/JsonEvent.java

@@ -0,0 +1,20 @@
+package com.vaadin.hilla.signals.core;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.UUID;
+
+public class JsonEvent extends StateEvent {
+
+    private final ObjectNode json;
+
+    public JsonEvent(UUID id, ObjectNode json) {
+        super(id);
+        this.json = json;
+    }
+
+    public ObjectNode getJson() {
+        return json;
+    }
+
+}

packages/java/endpoint/src/main/java/com/vaadin/hilla/signals/core/JsonEventMapper.java

@@ -0,0 +1,40 @@
+package com.vaadin.hilla.signals.core;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.util.Map;
+import java.util.UUID;
+
+public class JsonEventMapper {
+
+    private final ObjectMapper mapper;
+
+    public JsonEventMapper(ObjectMapper mapper) {
+        this.mapper = mapper;
+    }
+
+    public String toJson(JsonEvent jsonEvent) {
+        ObjectNode root = mapper.createObjectNode();
+        for (Map.Entry<String, JsonNode> entry : jsonEvent.getJson()
+                .properties()) {
+            root.set(entry.getKey(), entry.getValue());
+        }
+        UUID id = jsonEvent.getId();
+        root.put("id", id != null ? id.toString() : null);
+        return root.toString();
+    }
+
+    public JsonEvent fromJson(String json) {
+        try {
+            ObjectNode root = (ObjectNode) mapper.readTree(json);
+            UUID id = UUID.fromString(root.get("id").asText());
+            root.remove("id");
+            return new JsonEvent(id, root);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}