Skip to content

Commit 180de8e

Browse files
mjd507mjd507
authored
GH-8108: Add DSL for Redis Store channel adapters
Related to: #8108 Signed-off-by: Jiandong Ma <jiandong.ma.cn@gmail.com> * Use Supplier instead of Function for `Redis.storeInboundChannelAdapterSpec`, since no `rootObject` for a MessageSource based expression. use same overload method name for both expression-String and Expression in `RedisStoreOutboundChannelAdapterSpec`. use `MapAssert.containsOnly` in map entry assertions. Docs: use `@InboundChannelAdapter` instead of manual define `SourcePollingChannelAdapterFactoryBean`. Docs: afterCommitChannel param should use a dedicated `@Bean QueueChannel`. * Rename method names to be ended with Function when the param is a function. Signed-off-by: Jiandong Ma <jiandong.ma.cn@gmail.com>
1 parent bdc0496 commit 180de8e

5 files changed

Lines changed: 634 additions & 3 deletions

File tree

‎spring-integration-redis/src/main/java/org/springframework/integration/redis/dsl/Redis.java‎

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,14 @@
1717
package org.springframework.integration.redis.dsl;
1818

1919
import java.util.function.Function;
20+
import java.util.function.Supplier;
2021

2122
import org.springframework.data.redis.connection.RedisConnectionFactory;
23+
import org.springframework.data.redis.core.RedisTemplate;
2224
import org.springframework.expression.Expression;
25+
import org.springframework.expression.common.LiteralExpression;
2326
import org.springframework.integration.expression.FunctionExpression;
27+
import org.springframework.integration.expression.SupplierExpression;
2428
import org.springframework.messaging.Message;
2529

2630
/**
@@ -98,6 +102,100 @@ public static RedisQueueOutboundChannelAdapterSpec queueOutboundChannelAdapter(
98102
return new RedisQueueOutboundChannelAdapterSpec(new FunctionExpression<>(queueFunction), connectionFactory);
99103
}
100104

105+
/**
106+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
107+
* @param connectionFactory the {@link RedisConnectionFactory} to build on
108+
* @param key The key of the Redis collection to build on
109+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
110+
*/
111+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
112+
RedisConnectionFactory connectionFactory, String key) {
113+
114+
return storeInboundChannelAdapterSpec(connectionFactory, new LiteralExpression(key));
115+
}
116+
117+
/**
118+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
119+
* @param connectionFactory the {@link RedisConnectionFactory} to build on
120+
* @param keyExpression The keyExpression of the Redis collection to build on
121+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
122+
*/
123+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
124+
RedisConnectionFactory connectionFactory, Expression keyExpression) {
125+
126+
return new RedisStoreInboundChannelAdapterSpec(connectionFactory, keyExpression);
127+
}
128+
129+
/**
130+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
131+
* @param connectionFactory the {@link RedisConnectionFactory} to build on
132+
* @param keySupplier The keySupplier of the Redis collection to build on
133+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
134+
*/
135+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
136+
RedisConnectionFactory connectionFactory, Supplier<Message<?>> keySupplier) {
137+
138+
return storeInboundChannelAdapterSpec(connectionFactory, new SupplierExpression<>(keySupplier));
139+
}
140+
141+
/**
142+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
143+
* @param redisTemplate the {@link RedisTemplate} to build on
144+
* @param key The key of the Redis collection to build on
145+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
146+
*/
147+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
148+
RedisTemplate<String, ?> redisTemplate, String key) {
149+
150+
return storeInboundChannelAdapterSpec(redisTemplate, new LiteralExpression(key));
151+
}
152+
153+
/**
154+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
155+
* @param redisTemplate the {@link RedisTemplate} to build on
156+
* @param keyExpression The keyExpression of the Redis collection to build on
157+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
158+
*/
159+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
160+
RedisTemplate<String, ?> redisTemplate, Expression keyExpression) {
161+
162+
return new RedisStoreInboundChannelAdapterSpec(redisTemplate, keyExpression);
163+
}
164+
165+
/**
166+
* The factory to produce a {@link RedisStoreInboundChannelAdapterSpec}.
167+
* @param redisTemplate the {@link RedisTemplate} to build on
168+
* @param keySupplier The keySupplier of the Redis collection to build on
169+
* @return the {@link RedisStoreInboundChannelAdapterSpec} instance
170+
*/
171+
public static RedisStoreInboundChannelAdapterSpec storeInboundChannelAdapterSpec(
172+
RedisTemplate<String, ?> redisTemplate, Supplier<Message<?>> keySupplier) {
173+
174+
return storeInboundChannelAdapterSpec(redisTemplate, new SupplierExpression<>(keySupplier));
175+
}
176+
177+
/**
178+
* The factory to produce a {@link RedisStoreOutboundChannelAdapterSpec}.
179+
* @param connectionFactory the {@link RedisConnectionFactory} to build on
180+
* @return the {@link RedisStoreOutboundChannelAdapterSpec} instance
181+
*/
182+
public static RedisStoreOutboundChannelAdapterSpec storeOutboundChannelAdapterSpec(
183+
RedisConnectionFactory connectionFactory) {
184+
185+
return new RedisStoreOutboundChannelAdapterSpec(connectionFactory);
186+
}
187+
188+
/**
189+
* The factory to produce a {@link RedisStoreOutboundChannelAdapterSpec}.
190+
* @param redisTemplate the {@link RedisTemplate} to build on
191+
* @return the {@link RedisStoreOutboundChannelAdapterSpec} instance
192+
*/
193+
public static RedisStoreOutboundChannelAdapterSpec storeOutboundChannelAdapterSpec(
194+
RedisTemplate<String, ?> redisTemplate) {
195+
196+
return new RedisStoreOutboundChannelAdapterSpec(redisTemplate);
197+
}
198+
101199
private Redis() {
102200
}
103201

‎spring-integration-redis/src/main/java/org/springframework/integration/redis/dsl/RedisStoreInboundChannelAdapterSpec.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.integration.redis.dsl;
18+
19+
import org.springframework.data.redis.connection.RedisConnectionFactory;
20+
import org.springframework.data.redis.core.RedisTemplate;
21+
import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
22+
import org.springframework.expression.Expression;
23+
import org.springframework.integration.dsl.MessageSourceSpec;
24+
import org.springframework.integration.redis.inbound.RedisStoreMessageSource;
25+
26+
/**
27+
* A {@link MessageSourceSpec} for a {@link RedisStoreMessageSource}.
28+
*
29+
* @author Jiandong Ma
30+
*
31+
* @since 7.1
32+
*/
33+
public class RedisStoreInboundChannelAdapterSpec extends
34+
MessageSourceSpec<RedisStoreInboundChannelAdapterSpec, RedisStoreMessageSource> {
35+
36+
protected RedisStoreInboundChannelAdapterSpec(RedisTemplate<String, ?> redisTemplate, Expression keyExpression) {
37+
this.target = new RedisStoreMessageSource(redisTemplate, keyExpression);
38+
}
39+
40+
protected RedisStoreInboundChannelAdapterSpec(RedisConnectionFactory connectionFactory, Expression keyExpression) {
41+
this.target = new RedisStoreMessageSource(connectionFactory, keyExpression);
42+
}
43+
44+
/**
45+
* Specify the collection type. supported collections are LIST, SET, ZSET, PROPERTIES, and MAP.
46+
* @param collectionType the collectionType
47+
* @return the spec
48+
* @see RedisStoreMessageSource#setCollectionType(CollectionType)
49+
*/
50+
public RedisStoreInboundChannelAdapterSpec collectionType(CollectionType collectionType) {
51+
this.target.setCollectionType(collectionType);
52+
return this;
53+
}
54+
55+
}

‎spring-integration-redis/src/main/java/org/springframework/integration/redis/dsl/RedisStoreOutboundChannelAdapterSpec.java‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.integration.redis.dsl;
18+
19+
import java.util.function.Function;
20+
21+
import org.springframework.data.redis.connection.RedisConnectionFactory;
22+
import org.springframework.data.redis.core.RedisTemplate;
23+
import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
24+
import org.springframework.expression.Expression;
25+
import org.springframework.integration.dsl.MessageHandlerSpec;
26+
import org.springframework.integration.expression.FunctionExpression;
27+
import org.springframework.integration.redis.outbound.RedisStoreWritingMessageHandler;
28+
import org.springframework.messaging.Message;
29+
30+
/**
31+
* A {@link MessageHandlerSpec} for a {@link RedisStoreWritingMessageHandler}.
32+
*
33+
* @author Jiandong Ma
34+
*
35+
* @since 7.1
36+
*/
37+
public class RedisStoreOutboundChannelAdapterSpec extends
38+
MessageHandlerSpec<RedisStoreOutboundChannelAdapterSpec, RedisStoreWritingMessageHandler> {
39+
40+
protected RedisStoreOutboundChannelAdapterSpec(RedisTemplate<String, ?> redisTemplate) {
41+
this.target = new RedisStoreWritingMessageHandler(redisTemplate);
42+
}
43+
44+
protected RedisStoreOutboundChannelAdapterSpec(RedisConnectionFactory connectionFactory) {
45+
this.target = new RedisStoreWritingMessageHandler(connectionFactory);
46+
}
47+
48+
/**
49+
* Specify the key for the Redis store.
50+
* @param key the key
51+
* @return the spec
52+
* @see RedisStoreWritingMessageHandler#setKey(String)
53+
*/
54+
public RedisStoreOutboundChannelAdapterSpec key(String key) {
55+
this.target.setKey(key);
56+
return this;
57+
}
58+
59+
/**
60+
* Specify a SpEL Expression to be used to determine the key for the Redis store.
61+
* @param keyExpression the keyExpression
62+
* @return the spec
63+
* @see RedisStoreWritingMessageHandler#setKeyExpressionString(String)
64+
*/
65+
public RedisStoreOutboundChannelAdapterSpec keyExpression(String keyExpression) {
66+
this.target.setKeyExpressionString(keyExpression);
67+
return this;
68+
}
69+
70+
/**
71+
* Specify an Expression to be used to determine the key for the Redis store.
72+
* @param keyExpression the keyExpression
73+
* @return the spec
74+
* @see RedisStoreWritingMessageHandler#setKeyExpression(Expression)
75+
*/
76+
public RedisStoreOutboundChannelAdapterSpec keyExpression(Expression keyExpression) {
77+
this.target.setKeyExpression(keyExpression);
78+
return this;
79+
}
80+
81+
/**
82+
* Specify a KeyFunction to be used to determine the key for the Redis store.
83+
* @param keyFunction the keyFunction
84+
* @return the spec
85+
* @see RedisStoreWritingMessageHandler#setKeyExpression(Expression)
86+
*/
87+
public RedisStoreOutboundChannelAdapterSpec keyFunction(Function<Message<?>, String> keyFunction) {
88+
this.target.setKeyExpression(new FunctionExpression<>(keyFunction));
89+
return this;
90+
}
91+
92+
/**
93+
* Specify the collection type. supported collections are LIST, SET, ZSET, PROPERTIES, and MAP.
94+
* @param collectionType the collectionType
95+
* @return the spec
96+
* @see RedisStoreWritingMessageHandler#setCollectionType(CollectionType)
97+
*/
98+
public RedisStoreOutboundChannelAdapterSpec collectionType(CollectionType collectionType) {
99+
this.target.setCollectionType(collectionType);
100+
return this;
101+
}
102+
103+
/**
104+
* Specify whether payload elements should be extracted.
105+
* @param extractPayloadElements the extractPayloadElements
106+
* @return the spec
107+
* @see RedisStoreWritingMessageHandler#setExtractPayloadElements(boolean)
108+
*/
109+
public RedisStoreOutboundChannelAdapterSpec extractPayloadElements(boolean extractPayloadElements) {
110+
this.target.setExtractPayloadElements(extractPayloadElements);
111+
return this;
112+
}
113+
114+
/**
115+
* Specify the SPEL expression used as the key for Map and Properties entries.
116+
* @param mapKeyExpression the mapKeyExpression
117+
* @return the spec
118+
* @see RedisStoreWritingMessageHandler#setMapKeyExpressionString(String)
119+
*/
120+
public RedisStoreOutboundChannelAdapterSpec mapKeyExpression(String mapKeyExpression) {
121+
this.target.setMapKeyExpressionString(mapKeyExpression);
122+
return this;
123+
}
124+
125+
/**
126+
* Specify the expression used as the key for Map and Properties entries.
127+
* @param mapKeyExpression the mapKeyExpression
128+
* @return the spec
129+
* @see RedisStoreWritingMessageHandler#setMapKeyExpression(Expression)
130+
*/
131+
public RedisStoreOutboundChannelAdapterSpec mapKeyExpression(Expression mapKeyExpression) {
132+
this.target.setMapKeyExpression(mapKeyExpression);
133+
return this;
134+
}
135+
136+
/**
137+
* Specify the function used as the key for Map and Properties entries.
138+
* @param mapKeyFunction the mapKeyFunction
139+
* @return the spec
140+
* @see RedisStoreWritingMessageHandler#setMapKeyExpression(Expression)
141+
*/
142+
public RedisStoreOutboundChannelAdapterSpec mapKeyFunction(Function<Message<?>, String> mapKeyFunction) {
143+
this.target.setMapKeyExpression(new FunctionExpression<>(mapKeyFunction));
144+
return this;
145+
}
146+
147+
/**
148+
* Specify the SPEL expression used as the INCR flag for the ZADD command in case of ZSet collection.
149+
* @param zsetIncrementScoreExpression the zsetIncrementScoreExpression
150+
* @return the spec
151+
* @see RedisStoreWritingMessageHandler#setZsetIncrementExpressionString(String)
152+
*/
153+
public RedisStoreOutboundChannelAdapterSpec zsetIncrementScoreExpression(String zsetIncrementScoreExpression) {
154+
this.target.setZsetIncrementExpressionString(zsetIncrementScoreExpression);
155+
return this;
156+
}
157+
158+
/**
159+
* Specify the expression used as the INCR flag for the ZADD command in case of ZSet collection.
160+
* @param zsetIncrementScoreExpression the zsetIncrementScoreExpression
161+
* @return the spec
162+
* @see RedisStoreWritingMessageHandler#setZsetIncrementExpression(Expression)
163+
*/
164+
public RedisStoreOutboundChannelAdapterSpec zsetIncrementScoreExpression(Expression zsetIncrementScoreExpression) {
165+
this.target.setZsetIncrementExpression(zsetIncrementScoreExpression);
166+
return this;
167+
}
168+
169+
/**
170+
* Specify the function used as the INCR flag for the ZADD command in case of ZSet collection.
171+
* @param zsetIncrementScoreFunction the zsetIncrementScoreFunction
172+
* @return the spec
173+
* @see RedisStoreWritingMessageHandler#setZsetIncrementExpression(Expression)
174+
*/
175+
public RedisStoreOutboundChannelAdapterSpec zsetIncrementScoreFunction(Function<Message<?>, Boolean> zsetIncrementScoreFunction) {
176+
this.target.setZsetIncrementExpression(new FunctionExpression<>(zsetIncrementScoreFunction));
177+
return this;
178+
}
179+
180+
}

0 commit comments

Comments
 (0)