Document stream support in consumer configuration

Author: acogoluegnes

src/docs/asciidoc/usage.adoc

@@ -102,6 +102,8 @@ include::{test-examples}/Api.java[tag=consumer-consume]
 <2> Process the message
 <3> Accept (acknowledge) the message
 
+==== Graceful Shutdown
+
 A consumer can accept, discard, or requeue a message.
 We say the consumer _settles_ a message.
 
@@ -138,6 +140,35 @@ include::{test-examples}/Api.java[tag=consumer-abrupt-shutdown]
 --------
 <1> Close the consumer with potential unsettled messages
 
+==== Support for Streams
+
+There is out-of-the-box support for https://www.rabbitmq.com/docs/streams[streams] in consumer configuration.
+
+It is possible to set where to attach to when https://www.rabbitmq.com/docs/streams#consuming[consuming] from a stream:
+
+.Attaching to the beginning of a stream
+[source,java,indent=0]
+--------
+include::{test-examples}/Api.java[tag=consumer-consume-stream]
+--------
+<1> Access to stream-specific configuration helper
+<2> Attach to the first offset in the stream
+<3> Go back to main consumer configuration
+
+There is also support for https://www.rabbitmq.com/docs/streams#filtering[stream filtering] configuration:
+
+.Configuring stream filtering
+[source,java,indent=0]
+--------
+include::{test-examples}/Api.java[tag=consumer-consume-stream-filtering]
+--------
+<1> Access to stream-specific configuration helper
+<2> Set one or several filter values
+<3> Get "unfiltered" messages as well
+<4> Go back to main consumer configuration
+
+Consider also using the https://www.rabbitmq.com/docs/stream[native stream protocol] with the RabbitMQ https://github.com/rabbitmq/rabbitmq-stream-java-client[stream Java client] when working with streams.
+
 === Resource Management
 
 The `Management` object is the entry point to deal with resources.

src/test/java/com/rabbitmq/client/amqp/docs/Api.java

@@ -160,6 +160,37 @@ void consuming() {
     // end::consumer-abrupt-shutdown[]
   }
 
+  void consumingStream() {
+    Connection connection = null;
+    // tag::consumer-consume-stream[]
+    Consumer consumer = connection.consumerBuilder()
+        .queue("some-stream")
+        .stream() // <1>
+          .offset(ConsumerBuilder.StreamOffsetSpecification.FIRST) // <2>
+        .builder() // <3>
+        .messageHandler((context, message) -> {
+          // message processing
+        })
+        .build();
+    // end::consumer-consume-stream[]
+  }
+
+  void consumingStreamFiltering() {
+    Connection connection = null;
+    // tag::consumer-consume-stream-filtering[]
+    Consumer consumer = connection.consumerBuilder()
+        .queue("some-stream")
+        .stream() // <1>
+          .filterValues("invoices", "orders") // <2>
+          .filterMatchUnfiltered(true) // <3>
+        .builder() // <4>
+        .messageHandler((context, message) -> {
+          // message processing
+        })
+        .build();
+    // end::consumer-consume-stream-filtering[]
+  }
+
   void management() {
     Connection connection = null;
     // tag::management[]