Reactive Messaging - @Stream qualifier support

build-parent/pom.xml

@@ -42,7 +42,7 @@
       <smallrye-opentracing.version>1.2.1</smallrye-opentracing.version>
       <smallrye-fault-tolerance.version>1.1.0</smallrye-fault-tolerance.version>
       <smallrye-reactive-streams-operators.version>1.0.0</smallrye-reactive-streams-operators.version>
-      <smallrye-reactive-messaging.version>0.0.3</smallrye-reactive-messaging.version>
+      <smallrye-reactive-messaging.version>0.0.4</smallrye-reactive-messaging.version>
       <javax.activation.version>1.1.1</javax.activation.version>
       <javax.inject.version>1</javax.inject.version>
       <javax.annotation-api.version>1.3.2</javax.annotation-api.version>

extensions/reactive-messaging/deployment/src/test/java/org/jboss/shamrock/reactivemessaging/SimpleTest.java

@@ -3,6 +3,10 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.util.List;
+
+import javax.inject.Inject;
+
 import org.jboss.shamrock.test.ShamrockUnitTest;
 import org.jboss.shrinkwrap.api.ShrinkWrap;
 import org.jboss.shrinkwrap.api.spec.JavaArchive;
@@ -14,7 +18,10 @@ public class SimpleTest {
     @RegisterExtension
     static final ShamrockUnitTest config = new ShamrockUnitTest()
             .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
-                    .addClasses(SimpleBean.class));
+                    .addClasses(SimpleBean.class, StreamConsumer.class));
+
+    @Inject
+    StreamConsumer streamConsumer;
 
     @Test
     public void testSimpleBean() {
@@ -24,4 +31,15 @@ public void testSimpleBean() {
         assertTrue(SimpleBean.RESULT.contains("REACTIVE"));
         assertTrue(SimpleBean.RESULT.contains("MESSAGE"));
     }
+
+    @Test
+    public void testStreamInject() {
+        List<String> consumed = streamConsumer.consume();
+        assertEquals(5, consumed.size());
+        assertEquals("hello", consumed.get(0));
+        assertEquals("with", consumed.get(1));
+        assertEquals("SmallRye", consumed.get(2));
+        assertEquals("reactive", consumed.get(3));
+        assertEquals("message", consumed.get(4));
+    }
 }

extensions/reactive-messaging/deployment/src/test/java/org/jboss/shamrock/reactivemessaging/StreamConsumer.java

@@ -0,0 +1,27 @@
+package org.jboss.shamrock.reactivemessaging;
+
+import java.util.List;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import org.eclipse.microprofile.reactive.messaging.Message;
+
+import io.reactivex.Flowable;
+import io.smallrye.reactive.messaging.annotations.Stream;
+
+@ApplicationScoped
+public class StreamConsumer {
+
+    @Inject
+    @Stream("source")
+    private Flowable<Message<String>> sourceStream;
+
+    public List<String> consume() {
+        return Flowable.fromPublisher(sourceStream)
+                .map(Message::getPayload)
+                .toList()
+                .blockingGet();
+    }
+
+}