Fix filter_subject overriding filter_subjects (#711)

Signed-off-by: Mark Jan van Kampen <[email protected]>

Author: nickolaj-jepsen

nats/src/nats/js/client.py

@@ -437,8 +437,9 @@ async def cb(msg):
                 deliver = self._nc.new_inbox()
                 config.deliver_subject = deliver
 
-            # Auto created consumers use the filter subject.
-            config.filter_subject = subject
+            # Auto created consumers use the filter subject, unless filter_subjects is set.
+            if not config.filter_subjects:
+                config.filter_subject = subject
 
             # Heartbeats / FlowControl
             config.flow_control = flow_control
@@ -593,9 +594,10 @@ async def main():
             if config is None:
                 config = api.ConsumerConfig()
 
-            # Auto created consumers use the filter subject.
-            # config.name = durable
-            config.filter_subject = subject
+            # Auto created consumers use the filter subject, unless filter_subjects is set.
+            if not config.filter_subjects:
+                config.filter_subject = subject
+
             if durable:
                 config.name = durable
                 config.durable_name = durable

nats/tests/test_js.py

@@ -1136,6 +1136,33 @@ async def test_fetch_heartbeats(self):
 
         await nc.close()
 
+    @async_long_test
+    async def test_subscribe_filter_subjects(self):
+        nc = NATS()
+        await nc.connect()
+
+        js = nc.jetstream()
+
+        await js.add_stream(name="events", subjects=["events.>"])
+
+        sub = await js.pull_subscribe(
+            "events.>",
+            "filter",
+            config=nats.js.api.ConsumerConfig(
+                filter_subjects=["events.1", "events.2"],
+            ),
+        )
+        for i in range(0, 15):
+            await js.publish("events.%d" % i, b"i:%d" % i)
+        msgs = await sub.fetch(20, timeout=5)
+        assert len(msgs) == 2
+        for msg in msgs:
+            await msg.ack_sync()
+        info = await js.consumer_info("events", "filter")
+        assert info.num_pending == 0
+
+        await nc.close()
+
 
 class JSMTest(SingleJetStreamServerTestCase):
     @async_test
@@ -2093,6 +2120,37 @@ async def cb_d(msg):
         await js.delete_stream("pconfig")
         await nc.close()
 
+    @async_long_test
+    async def test_subscribe_filter_subjects(self):
+        nc = NATS()
+        await nc.connect()
+
+        js = nc.jetstream()
+
+        await js.add_stream(name="events", subjects=["events.>"])
+        a = []
+
+        def cb(msg):
+            a.append(msg)
+
+        sub = await js.subscribe(
+            "events.>",
+            "filter",
+            cb=cb,
+            config=nats.js.api.ConsumerConfig(
+                filter_subjects=["events.1", "events.2"],
+            ),
+        )
+        for i in range(0, 15):
+            await js.publish("events.%d" % i, b"i:%d" % i)
+        await asyncio.sleep(1)
+        assert len(a) == 2
+
+        info = await sub.consumer_info()
+        assert info.num_pending == 0
+
+        await nc.close()
+
 
 class AckPolicyTest(SingleJetStreamServerTestCase):
     @async_test