CAMEL-23239: Add camel-state-store component with pluggable key-value store

bom/camel-bom/pom.xml

@@ -2257,6 +2257,26 @@
         <artifactId>camel-ssh</artifactId>
         <version>4.19.0-SNAPSHOT</version>
       </dependency>
+      <dependency>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>camel-state-store</artifactId>
+        <version>4.19.0-SNAPSHOT</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>camel-state-store-caffeine</artifactId>
+        <version>4.19.0-SNAPSHOT</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>camel-state-store-infinispan</artifactId>
+        <version>4.19.0-SNAPSHOT</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>camel-state-store-redis</artifactId>
+        <version>4.19.0-SNAPSHOT</version>
+      </dependency>
       <dependency>
         <groupId>org.apache.camel</groupId>
         <artifactId>camel-stax</artifactId>

catalog/camel-allcomponents/pom.xml

@@ -2037,6 +2037,26 @@
             <artifactId>camel-ssh</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-state-store</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-state-store-caffeine</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-state-store-infinispan</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-state-store-redis</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-stax</artifactId>

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components.properties

@@ -351,6 +351,7 @@ spring-ws
 sql
 sql-stored
 ssh
+state-store
 stax
 stitch
 stomp

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/state-store.json

@@ -0,0 +1,42 @@
+{
+  "component": {
+    "kind": "component",
+    "name": "state-store",
+    "title": "State Store",
+    "description": "Perform key-value operations against a pluggable state store backend.",
+    "deprecated": false,
+    "firstVersion": "4.19.0",
+    "label": "cache",
+    "javaType": "org.apache.camel.component.statestore.StateStoreComponent",
+    "supportLevel": "Preview",
+    "groupId": "org.apache.camel",
+    "artifactId": "camel-state-store",
+    "version": "4.19.0-SNAPSHOT",
+    "scheme": "state-store",
+    "extendsScheme": "",
+    "syntax": "state-store:storeName",
+    "async": false,
+    "api": false,
+    "consumerOnly": false,
+    "producerOnly": true,
+    "lenientProperties": false,
+    "browsable": false,
+    "remote": false
+  },
+  "componentProperties": {
+    "lazyStartProducer": { "index": 0, "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." },
+    "autowiredEnabled": { "index": 1, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc." }
+  },
+  "headers": {
+    "CamelStateStoreOperation": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "org.apache.camel.component.statestore.StateStoreOperations", "enum": [ "put", "putIfAbsent", "get", "delete", "contains", "keys", "size", "clear" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The operation to perform", "constantName": "org.apache.camel.component.statestore.StateStoreConstants#OPERATION" },
+    "CamelStateStoreKey": { "index": 1, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The key to use for the operation", "constantName": "org.apache.camel.component.statestore.StateStoreConstants#KEY" },
+    "CamelStateStoreTtl": { "index": 2, "kind": "header", "displayName": "", "group": "producer", "label": "producer", "required": false, "javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Per-message TTL override in milliseconds. Takes precedence over the endpoint ttl option.", "constantName": "org.apache.camel.component.statestore.StateStoreConstants#TTL" }
+  },
+  "properties": {
+    "storeName": { "index": 0, "kind": "path", "displayName": "Store Name", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The name of the state store" },
+    "operation": { "index": 1, "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.statestore.StateStoreOperations", "enum": [ "put", "putIfAbsent", "get", "delete", "contains", "keys", "size", "clear" ], "deprecated": false, "autowired": false, "secret": false, "description": "The default operation to perform" },
+    "ttl": { "index": 2, "kind": "parameter", "displayName": "Ttl", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Time-to-live in milliseconds for entries. 0 means no expiry." },
+    "lazyStartProducer": { "index": 3, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and starting the producer may take a little time and prolong the total processing time of the processing." },
+    "backend": { "index": 4, "kind": "parameter", "displayName": "Backend", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.component.statestore.StateStoreBackend", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "memory", "description": "The backend to use. Default is an in-memory store. Set to a bean reference (e.g. #myBackend) for custom backends." }
+  }
+}

components/camel-state-store/camel-state-store-caffeine/pom.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.camel</groupId>
+        <artifactId>camel-state-store-parent</artifactId>
+        <version>4.19.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>camel-state-store-caffeine</artifactId>
+    <packaging>jar</packaging>
+
+    <name>Camel :: State Store :: Caffeine</name>
+    <description>Camel State Store backend using Caffeine cache</description>
+
+    <properties>
+        <firstVersion>4.19.0</firstVersion>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-state-store</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+            <version>${caffeine-version}</version>
+        </dependency>
+
+        <!-- testing -->
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-test-junit5</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>

components/camel-state-store/camel-state-store-caffeine/src/generated/resources/META-INF/services/org/apache/camel/other.properties

@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+name=state-store-caffeine
+groupId=org.apache.camel
+artifactId=camel-state-store-caffeine
+version=4.19.0-SNAPSHOT
+projectName=Camel :: State Store :: Caffeine
+projectDescription=Camel State Store backend using Caffeine cache

components/camel-state-store/camel-state-store-caffeine/src/generated/resources/state-store-caffeine.json

@@ -0,0 +1,14 @@
+{
+  "other": {
+    "kind": "other",
+    "name": "state-store-caffeine",
+    "title": "State Store Caffeine",
+    "description": "Camel State Store backend using Caffeine cache",
+    "deprecated": false,
+    "firstVersion": "4.19.0",
+    "supportLevel": "Preview",
+    "groupId": "org.apache.camel",
+    "artifactId": "camel-state-store-caffeine",
+    "version": "4.19.0-SNAPSHOT"
+  }
+}

components/camel-state-store/camel-state-store-caffeine/src/main/java/org/apache/camel/component/statestore/caffeine/CaffeineStateStoreBackend.java

@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.component.statestore.caffeine;
+
+import java.time.Duration;
+import java.util.Set;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import org.apache.camel.component.statestore.StateStoreBackend;
+
+/**
+ * A {@link StateStoreBackend} implementation backed by Caffeine cache. Supports per-entry TTL using Caffeine's variable
+ * expiration.
+ */
+public class CaffeineStateStoreBackend implements StateStoreBackend {
+
+    private Cache<String, TimedValue> cache;
+    private int maximumSize = 10_000;
+
+    @Override
+    public Object put(String key, Object value, long ttlMillis) {
+        TimedValue previous = cache.getIfPresent(key);
+        cache.put(key, new TimedValue(value, ttlMillis));
+        return previous != null ? previous.value() : null;
+    }
+
+    @Override
+    public Object get(String key) {
+        TimedValue entry = cache.getIfPresent(key);
+        return entry != null ? entry.value() : null;
+    }
+
+    @Override
+    public Object delete(String key) {
+        TimedValue previous = cache.getIfPresent(key);
+        cache.invalidate(key);
+        return previous != null ? previous.value() : null;
+    }
+
+    @Override
+    public boolean contains(String key) {
+        return cache.getIfPresent(key) != null;
+    }
+
+    @Override
+    public Object putIfAbsent(String key, Object value, long ttlMillis) {
+        TimedValue existing = cache.getIfPresent(key);
+        if (existing != null) {
+            return existing.value();
+        }
+        cache.put(key, new TimedValue(value, ttlMillis));
+        return null;
+    }
+
+    @Override
+    public int size() {
+        cache.cleanUp();
+        return (int) cache.estimatedSize();
+    }
+
+    @Override
+    public Set<String> keys() {
+        return Set.copyOf(cache.asMap().keySet());
+    }
+
+    @Override
+    public void clear() {
+        cache.invalidateAll();
+    }
+
+    @Override
+    public void start() {
+        cache = Caffeine.newBuilder()
+                .maximumSize(maximumSize)
+                .expireAfter(new Expiry<String, TimedValue>() {
+                    @Override
+                    public long expireAfterCreate(String key, TimedValue value, long currentTime) {
+                        return value.ttlMillis() > 0
+                                ? Duration.ofMillis(value.ttlMillis()).toNanos()
+                                : Long.MAX_VALUE;
+                    }
+
+                    @Override
+                    public long expireAfterUpdate(String key, TimedValue value, long currentTime, long currentDuration) {
+                        return value.ttlMillis() > 0
+                                ? Duration.ofMillis(value.ttlMillis()).toNanos()
+                                : Long.MAX_VALUE;
+                    }
+
+                    @Override
+                    public long expireAfterRead(String key, TimedValue value, long currentTime, long currentDuration) {
+                        return currentDuration;
+                    }
+                })
+                .build();
+    }
+
+    @Override
+    public void stop() {
+        if (cache != null) {
+            cache.invalidateAll();
+            cache.cleanUp();
+            cache = null;
+        }
+    }
+
+    public int getMaximumSize() {
+        return maximumSize;
+    }
+
+    public void setMaximumSize(int maximumSize) {
+        this.maximumSize = maximumSize;
+    }
+
+    private record TimedValue(Object value, long ttlMillis) {
+    }
+}