GH-2540: Add SuperStreamBuilder

Resolves #2540

Usability improvements: New SuperStream builder

builder provide a way to configure:
- maxAge
- maxLength
- maxSegmentSize

Usability improvements: New SuperStream builder

License

Usability improvements: New SuperStream builder

Fix style tests and add a new one for the super stream builder

Usability improvements: New SuperStream builder

Covered x-initial-cluster-size. Fixes after review

Author: kurenchuksergey

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/SuperStream.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.BiFunction;
@@ -36,8 +37,8 @@
  * Create Super Stream Topology {@link Declarable}s.
  *
  * @author Gary Russell
+ * @author Sergei Kurenchuk
  * @since 3.0
- *
  */
 public class SuperStream extends Declarables {
 
@@ -47,9 +48,22 @@ public class SuperStream extends Declarables {
 	 * @param partitions the number of partitions.
 	 */
 	public SuperStream(String name, int partitions) {
+		this(name, partitions, Map.of());
+	}
+
+	/**
+	 * Create a Super Stream with the provided parameters.
+	 * @param name the stream name.
+	 * @param partitions the number of partitions.
+	 * @param arguments the stream arguments
+	 * @since 3.1
+	 */
+	public SuperStream(String name, int partitions, Map<String, Object> arguments) {
 		this(name, partitions, (q, i) -> IntStream.range(0, i)
 				.mapToObj(String::valueOf)
-				.collect(Collectors.toList()));
+				.collect(Collectors.toList()),
+				arguments
+		);
 	}
 
 	/**
@@ -61,19 +75,37 @@ public SuperStream(String name, int partitions) {
 	 * partitions, the returned list must have a size equal to the partitions.
 	 */
 	public SuperStream(String name, int partitions, BiFunction<String, Integer, List<String>> routingKeyStrategy) {
-		super(declarables(name, partitions, routingKeyStrategy));
+		this(name, partitions, routingKeyStrategy, Map.of());
+	}
+
+	/**
+	 * Create a Super Stream with the provided parameters.
+	 * @param name the stream name.
+	 * @param partitions the number of partitions.
+	 * @param routingKeyStrategy a strategy to determine routing keys to use for the
+	 * partitions. The first parameter is the queue name, the second the number of
+	 * partitions, the returned list must have a size equal to the partitions.
+	 * @param arguments the stream arguments
+	 * @since 3.1
+	 */
+	public SuperStream(String name, int partitions, BiFunction<String, Integer, List<String>> routingKeyStrategy, Map<String, Object> arguments) {
+		super(declarables(name, partitions, routingKeyStrategy, arguments));
 	}
 
 	private static Collection<Declarable> declarables(String name, int partitions,
-			BiFunction<String, Integer, List<String>> routingKeyStrategy) {
+													  BiFunction<String, Integer, List<String>> routingKeyStrategy,
+													  Map<String, Object> arguments) {
 
 		List<Declarable> declarables = new ArrayList<>();
 		List<String> rks = routingKeyStrategy.apply(name, partitions);
 		Assert.state(rks.size() == partitions, () -> "Expected " + partitions + " routing keys, not " + rks.size());
 		declarables.add(new DirectExchange(name, true, false, Map.of("x-super-stream", true)));
+
+		Map<String, Object> argumentsCopy = new HashMap<>(arguments);
+		argumentsCopy.put("x-queue-type", "stream");
 		for (int i = 0; i < partitions; i++) {
 			String rk = rks.get(i);
-			Queue q = new Queue(name + "-" + i, true, false, false, Map.of("x-queue-type", "stream"));
+			Queue q = new Queue(name + "-" + i, true, false, false, argumentsCopy);
 			declarables.add(q);
 			declarables.add(new Binding(q.getName(), DestinationType.QUEUE, name, rk,
 					Map.of("x-stream-partition-order", i)));

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/SuperStreamBuilder.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2021-2023 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.springframework.rabbit.stream.config;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.springframework.util.StringUtils;
+
+/**
+ * Builds a Spring AMQP Super Stream using a fluent API.
+ * Based on <a href="https://www.rabbitmq.com/streams.html">Streams documentation</a>
+ *
+ * @author Sergei Kurenchuk
+ * @since 3.1
+ */
+public class SuperStreamBuilder {
+	private final Map<String, Object> arguments = new HashMap<>();
+	private String name;
+	private int partitions = -1;
+
+	private BiFunction<String, Integer, List<String>> routingKeyStrategy;
+
+	/**
+	 * Creates a builder for Super Stream.
+	 * @param name stream name
+	 * @return the builder
+	 */
+	public static SuperStreamBuilder superStream(String name) {
+		SuperStreamBuilder builder = new SuperStreamBuilder();
+		builder.name(name);
+		return builder;
+	}
+
+	/**
+	 * Creates a builder for Super Stream.
+	 * @param name stream name
+	 * @param partitions partitions number
+	 * @return the builder
+	 */
+	public static SuperStreamBuilder superStream(String name, int partitions) {
+		return superStream(name).partitions(partitions);
+	}
+
+	/**
+	 * Set the maximum age retention per stream, which will remove the oldest data.
+	 * @param maxAge valid units: Y, M, D, h, m, s. For example: "7D" for a week
+	 * @return the builder
+	 */
+	public SuperStreamBuilder maxAge(String maxAge) {
+		return withArgument("x-max-age", maxAge);
+	}
+
+	/**
+	 * Set the maximum log size as the retention configuration for each stream,
+	 * which will truncate the log based on the data size.
+	 * @param bytes the max total size in bytes
+	 * @return the builder
+	 */
+	public SuperStreamBuilder maxLength(int bytes) {
+		return withArgument("max-length-bytes", bytes);
+	}
+
+	/**
+	 * Set the maximum size limit for segment file.
+	 * @param bytes the max segments size in bytes
+	 * @return the builder
+	 */
+	public SuperStreamBuilder maxSegmentSize(int bytes) {
+		return withArgument("x-stream-max-segment-size-bytes", bytes);
+	}
+
+	/**
+	 * Set initial replication factor for each partition.
+	 * @param count number of nodes per partition
+	 * @return the builder
+	 */
+	public SuperStreamBuilder initialClusterSize(int count) {
+		return withArgument("x-initial-cluster-size", count);
+	}
+
+	/**
+	 * Set extra argument which is not covered by builder's methods.
+	 * @param key argument name
+	 * @param value argument value
+	 * @return the builder
+	 */
+	public SuperStreamBuilder withArgument(String key, Object value) {
+		if ("x-queue-type".equals(key) && !"stream".equals(value)) {
+			throw new IllegalArgumentException("Changing x-queue-type argument is not permitted");
+		}
+		this.arguments.put(key, value);
+		return this;
+	}
+
+	/**
+	 * Set the stream name.
+	 * @param name the stream name.
+	 * @return the builder
+	 */
+	public SuperStreamBuilder name(String name) {
+		this.name = name;
+		return this;
+	}
+
+	/**
+	 * Set the partitions number.
+	 * @param partitions the partitions number
+	 * @return the builder
+	 */
+	public SuperStreamBuilder partitions(int partitions) {
+		this.partitions = partitions;
+		return this;
+	}
+
+	/**
+	 * Set a strategy to determine routing keys to use for the
+	 * partitions. The first parameter is the queue name, the second the number of
+	 * partitions, the returned list must have a size equal to the partitions.
+	 * @param routingKeyStrategy the strategy
+	 * @return the builder
+	 */
+	public SuperStreamBuilder routingKeyStrategy(BiFunction<String, Integer, List<String>> routingKeyStrategy) {
+		this.routingKeyStrategy = routingKeyStrategy;
+		return this;
+	}
+
+	/**
+	 * Builds a final Super Stream.
+	 * @return the Super Stream instance
+	 */
+	public SuperStream build() {
+		if (!StringUtils.hasText(this.name)) {
+			throw new IllegalArgumentException("Stream name can't be empty");
+		}
+
+		if (this.partitions <= 0) {
+			throw new IllegalArgumentException(
+					String.format("Partitions number should be great then zero. Current value; %d", this.partitions)
+			);
+		}
+
+		if (this.routingKeyStrategy == null) {
+			return new SuperStream(this.name, this.partitions, this.arguments);
+		}
+
+		return new SuperStream(this.name, this.partitions, this.routingKeyStrategy, this.arguments);
+	}
+}