Make RoutingStratey a public API

The metadata of the super stream are now provided in the
route method. This abstraction hides the infrastructure
needed to find out about the super stream topology.

The user can now have full control over the routing strategy,
even though the provided implementations ("hash" and "key")
should be enough most of the time.

The configuration of the routing strategy has been also refactored
to reflect the change and to be simpler.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/ConsumerBuilder.java

@@ -27,7 +27,9 @@ public interface ConsumerBuilder {
   ConsumerBuilder stream(String stream);
 
   /**
-   * Set the consumer to consume from a super stream (partitioned stream). Experimental!
+   * Set the consumer to consume from a super stream (partitioned stream).
+   *
+   * <p>This is an experimental API, subject to change.
    *
    * @param superStream
    * @return this builder instance

src/main/java/com/rabbitmq/stream/ProducerBuilder.java

@@ -108,37 +108,81 @@ public interface ProducerBuilder {
   ProducerBuilder enqueueTimeout(Duration timeout);
 
   /**
-   * Routing strategy for super streams. Experimental!
+   * Create the {@link Producer} instance.
    *
-   * @param routingKeyExtractor
-   * @param routingType
-   * @return this builder instance
+   * @return the configured producer
    */
-  ProducerBuilder routing(Function<Message, String> routingKeyExtractor, RoutingType routingType);
+  Producer build();
 
   /**
-   * Routing strategy for super streams. Experimental!
+   * Configure the routing for super streams (partitioned streams).
    *
-   * @param routingKeyExtractor
-   * @param routingType
-   * @param hash
-   * @return this builder instance
+   * <p>This is an experimental API, subject to change.
+   *
+   * <p>The to-be-created producer will be a composite producer when this method is called. It will
+   * use the routing configuration to find out where a message should be routed. The application
+   * developer must provide the logic to extract a "routing key" from a message, which will decide
+   * the destination(s) of the message.
+   *
+   * <p>The default routing strategy hashes the routing key to choose the stream (partition) to send
+   * the message to.
+   *
+   * @param routingKeyExtractor the logic to extract a routing key from a message
+   * @return the routing configuration instance
+   * @see RoutingConfiguration
    */
-  ProducerBuilder routing(
-      Function<Message, String> routingKeyExtractor,
-      RoutingType routingType,
-      ToIntFunction<String> hash);
+  RoutingConfiguration routing(Function<Message, String> routingKeyExtractor);
 
   /**
-   * Create the {@link Producer} instance.
+   * Routing configuration for super streams (partitioned streams).
    *
-   * @return the configured producer
+   * <p>This is an experimental API, subject to change.
    */
-  Producer build();
+  interface RoutingConfiguration {
+
+    /**
+     * Enable the "hash" routing strategy (the default).
+     *
+     * <p>The default hash algorithm is 32-bit MurmurHash3.
+     *
+     * @return the routing configuration instance
+     */
+    RoutingConfiguration hash();
+
+    /**
+     * Enable the "hash" routing strategy with a specific hash algorithm.
+     *
+     * @param hash
+     * @return
+     */
+    RoutingConfiguration hash(ToIntFunction<String> hash);
+
+    /**
+     * Enable the "key" routing strategy.
+     *
+     * <p>It consists in using the "route" command of the RabbitMQ Stream protocol to determine the
+     * streams to send a message to.
+     *
+     * @return the routing configuration instance
+     */
+    RoutingConfiguration key();
+
+    /**
+     * Set the routing strategy to use.
+     *
+     * <p>Providing the routing strategy provides control over the streams a message is routed to
+     * (routing key extraction logic if relevant and destination(s) decision).
+     *
+     * @param routingStrategy
+     * @return the routing configuration instance
+     */
+    RoutingConfiguration strategy(RoutingStrategy routingStrategy);
 
-  /** Routing type when using super streams. Experimental! */
-  enum RoutingType {
-    HASH,
-    KEY
+    /**
+     * Go back to the producer builder.
+     *
+     * @return the producer builder
+     */
+    ProducerBuilder producerBuilder();
   }
 }

src/main/java/com/rabbitmq/stream/impl/RoutingStrategy.java renamed to src/main/java/com/rabbitmq/stream/RoutingStrategy.java

@@ -11,12 +11,36 @@
 //
 // If you have any questions regarding licensing, please contact us at
 // [email protected].
-package com.rabbitmq.stream.impl;
+package com.rabbitmq.stream;
 
-import com.rabbitmq.stream.Message;
 import java.util.List;
+import java.util.function.Function;
 
-interface RoutingStrategy {
+/**
+ * Strategy to route outbound messages to appropriate streams.
+ *
+ * <p>This is an experimental API, subject to change.
+ *
+ * <p>Used for super streams (partitioned stream).
+ *
+ * @see ProducerBuilder#routing(Function)
+ */
+public interface RoutingStrategy {
 
-  List<String> route(Message message);
+  /**
+   * Where to route a message.
+   *
+   * @param message
+   * @param metadata
+   * @return
+   */
+  List<String> route(Message message, Metadata metadata);
+
+  /** Metadata on the super stream. */
+  interface Metadata {
+
+    List<String> partitions();
+
+    List<String> route(String routingKey);
+  }
 }

src/main/java/com/rabbitmq/stream/impl/HashRoutingStrategy.java

@@ -14,41 +14,28 @@
 package com.rabbitmq.stream.impl;
 
 import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.RoutingStrategy;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
 import java.util.function.ToIntFunction;
-import java.util.stream.Collectors;
 
 class HashRoutingStrategy implements RoutingStrategy {
 
   private final Function<Message, String> routingKeyExtractor;
 
-  private final StreamEnvironment env;
-
-  private final List<List<String>> partitions;
-
   private final ToIntFunction<String> hash;
 
-  HashRoutingStrategy(
-      String superStream,
-      Function<Message, String> routingKeyExtractor,
-      StreamEnvironment env,
-      ToIntFunction<String> hash) {
+  HashRoutingStrategy(Function<Message, String> routingKeyExtractor, ToIntFunction<String> hash) {
     this.routingKeyExtractor = routingKeyExtractor;
-    this.env = env;
-    List<String> ps = this.env.locatorOperation(c -> c.partitions(superStream));
-    this.partitions =
-        new CopyOnWriteArrayList<>(
-            ps.stream().map(Collections::singletonList).collect(Collectors.toList()));
     this.hash = hash;
   }
 
   @Override
-  public List<String> route(Message message) {
+  public List<String> route(Message message, Metadata metadata) {
     String routingKey = routingKeyExtractor.apply(message);
     int hashValue = hash.applyAsInt(routingKey);
-    return this.partitions.get((hashValue & 0x7FFFFFFF) % this.partitions.size());
+    List<String> partitions = metadata.partitions();
+    return Collections.singletonList(partitions.get((hashValue & 0x7FFFFFFF) % partitions.size()));
   }
 }

src/main/java/com/rabbitmq/stream/impl/RoutingKeyRoutingStrategy.java

@@ -14,6 +14,7 @@
 package com.rabbitmq.stream.impl;
 
 import com.rabbitmq.stream.Message;
+import com.rabbitmq.stream.RoutingStrategy;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -25,27 +26,16 @@ class RoutingKeyRoutingStrategy implements RoutingStrategy {
 
   private final Map<String, List<String>> routingKeysToStreams = new ConcurrentHashMap<>();
 
-  private final StreamEnvironment env;
-
-  private final String superStream;
-
-  RoutingKeyRoutingStrategy(
-      String superStream, Function<Message, String> routingKeyExtractor, StreamEnvironment env) {
+  RoutingKeyRoutingStrategy(Function<Message, String> routingKeyExtractor) {
     this.routingKeyExtractor = routingKeyExtractor;
-    this.env = env;
-    this.superStream = superStream;
   }
 
   @Override
-  public List<String> route(Message message) {
+  public List<String> route(Message message, Metadata metadata) {
     String routingKey = this.routingKeyExtractor.apply(message);
     List<String> streams =
         routingKeysToStreams.computeIfAbsent(
-            routingKey,
-            routingKey1 -> {
-              // TODO retry on locator lookup
-              return env.locatorOperation(c -> c.route(routingKey1, superStream));
-            });
+            routingKey, routingKey1 -> metadata.route(routingKey1));
     return streams;
   }
 }

src/main/java/com/rabbitmq/stream/impl/StreamProducerBuilder.java

@@ -16,6 +16,7 @@
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.Producer;
 import com.rabbitmq.stream.ProducerBuilder;
+import com.rabbitmq.stream.RoutingStrategy;
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.compression.Compression;
 import java.lang.reflect.Field;
@@ -45,11 +46,7 @@ class StreamProducerBuilder implements ProducerBuilder {
 
   private Duration enqueueTimeout = Duration.ofSeconds(10);
 
-  private Function<Message, String> routingKeyExtractor;
-
-  private RoutingType routingType;
-
-  private ToIntFunction<String> hash = HashUtils.MURMUR3;
+  private DefaultRoutingConfiguration routingConfiguration;
 
   StreamProducerBuilder(StreamEnvironment environment) {
     this.environment = environment;
@@ -126,32 +123,14 @@ public ProducerBuilder enqueueTimeout(Duration timeout) {
   }
 
   @Override
-  public ProducerBuilder routing(
-      Function<Message, String> routingKeyExtractor, RoutingType routingType) {
-    return this.routing(routingKeyExtractor, routingType, HashUtils.MURMUR3);
-  }
-
-  @Override
-  public ProducerBuilder routing(
-      Function<Message, String> routingKeyExtractor,
-      RoutingType routingType,
-      ToIntFunction<String> hash) {
-    if (routingKeyExtractor == null || routingType == null) {
-      throw new IllegalArgumentException(
-          "both routing key extractor and routing type must be non-null");
-    }
-    this.routingKeyExtractor = routingKeyExtractor;
-    this.routingType = routingType;
-    if (hash != null) {
-      this.hash = hash;
-    }
-    return this;
+  public RoutingConfiguration routing(Function<Message, String> routingKeyExtractor) {
+    this.routingConfiguration = new DefaultRoutingConfiguration(this);
+    this.routingConfiguration.routingKeyExtractor = routingKeyExtractor;
+    return this.routingConfiguration;
   }
 
   void resetRouting() {
-    this.routingKeyExtractor = null;
-    this.routingType = null;
-    this.hash = null;
+    this.routingConfiguration = null;
   }
 
   public Producer build() {
@@ -164,7 +143,7 @@ public Producer build() {
     }
     this.environment.maybeInitializeLocator();
     Producer producer;
-    if (this.routingKeyExtractor == null) {
+    if (this.routingConfiguration == null) {
       producer =
           new StreamProducer(
               name,
@@ -179,14 +158,17 @@ public Producer build() {
               environment);
       this.environment.addProducer((StreamProducer) producer);
     } else {
-      // FIXME propagate compression to super stream producer
-      ToIntFunction<String> hashFunction = this.hash == null ? HashUtils.MURMUR3 : this.hash;
-      RoutingStrategy routingStrategy =
-          this.routingType == RoutingType.HASH
-              ? new HashRoutingStrategy(
-                  this.stream, this.routingKeyExtractor, this.environment, hashFunction)
-              : new RoutingKeyRoutingStrategy(
-                  this.stream, this.routingKeyExtractor, this.environment);
+      RoutingStrategy routingStrategy = this.routingConfiguration.routingStrategy;
+      if (routingStrategy == null) {
+        if (this.routingConfiguration.hash == null) {
+          routingStrategy =
+              new RoutingKeyRoutingStrategy(this.routingConfiguration.routingKeyExtractor);
+        } else {
+          routingStrategy =
+              new HashRoutingStrategy(
+                  this.routingConfiguration.routingKeyExtractor, this.routingConfiguration.hash);
+        }
+      }
       producer =
           new SuperStreamProducer(this, this.name, this.stream, routingStrategy, this.environment);
     }
@@ -205,4 +187,53 @@ StreamProducerBuilder duplicate() {
     }
     return duplicate;
   }
+
+  static final class DefaultRoutingConfiguration implements RoutingConfiguration {
+
+    private final StreamProducerBuilder producerBuilder;
+
+    private Function<Message, String> routingKeyExtractor;
+
+    private RoutingStrategy routingStrategy;
+
+    private ToIntFunction<String> hash = HashUtils.MURMUR3;
+
+    DefaultRoutingConfiguration(StreamProducerBuilder producerBuilder) {
+      this.producerBuilder = producerBuilder;
+    }
+
+    @Override
+    public RoutingConfiguration hash() {
+      if (this.hash == null) {
+        this.hash = HashUtils.MURMUR3;
+      }
+      this.routingStrategy = null;
+      return this;
+    }
+
+    @Override
+    public RoutingConfiguration hash(ToIntFunction<String> hash) {
+      this.hash = hash;
+      this.routingStrategy = null;
+      return this;
+    }
+
+    @Override
+    public RoutingConfiguration key() {
+      this.hash = null;
+      this.routingStrategy = null;
+      return this;
+    }
+
+    @Override
+    public RoutingConfiguration strategy(RoutingStrategy routingStrategy) {
+      this.routingStrategy = routingStrategy;
+      return this;
+    }
+
+    @Override
+    public ProducerBuilder producerBuilder() {
+      return this.producerBuilder;
+    }
+  }
 }