Fixed some interfaces.

Author: s3rius

python/natsrpy/_natsrpy_rs/__init__.pyi

@@ -8,6 +8,7 @@ from natsrpy._natsrpy_rs.message import Message
 class IteratorSubscription:
     def __aiter__(self) -> IteratorSubscription: ...
     async def __anext__(self) -> Message: ...
+    async def next(self, timeout: float | timedelta | None = None) -> Message: ...
     async def unsubscribe(self, limit: int | None = None) -> None: ...
     async def drain(self) -> None: ...
 
@@ -41,7 +42,15 @@ class Nats:
         reply: str | None = None,
         err_on_disconnect: bool = False,
     ) -> None: ...
-    async def request(self, subject: str, payload: bytes) -> None: ...
+    async def request(
+        self,
+        subject: str,
+        payload: bytes | str | bytearray | memoryview,
+        *,
+        headers: dict[str, Any] | None = None,
+        reply: str | None = None,
+        err_on_disconnect: bool = False,
+    ) -> None: ...
     async def drain(self) -> None: ...
     async def flush(self) -> None: ...
     @overload

python/natsrpy/_natsrpy_rs/js/managers.pyi

@@ -15,14 +15,14 @@ class StreamsManager:
     async def create(self, config: StreamConfig) -> Stream: ...
     async def create_or_update(self, config: StreamConfig) -> Stream: ...
     async def get(self, bucket: str) -> Stream: ...
-    async def delete(self, bucket: str) -> None: ...
+    async def delete(self, bucket: str) -> bool: ...
     async def update(self, config: StreamConfig) -> Stream: ...
 
 class KVManager:
     async def create(self, config: KVConfig) -> KeyValue: ...
     async def create_or_update(self, config: KVConfig) -> KeyValue: ...
     async def get(self, bucket: str) -> KeyValue: ...
-    async def delete(self, bucket: str) -> None: ...
+    async def delete(self, bucket: str) -> bool: ...
     async def update(self, config: KVConfig) -> KeyValue: ...
 
 class ConsumersManager:

python/tests/test_configs.py

@@ -1,11 +1,9 @@
 from natsrpy.js import (
     AckPolicy,
-    Compression,
     DeliverPolicy,
     DiscardPolicy,
     KVConfig,
     ObjectStoreConfig,
-    PersistenceMode,
     PullConsumerConfig,
     PushConsumerConfig,
     ReplayPolicy,
@@ -16,55 +14,6 @@
 )
 
 
-async def test_storage_type_values() -> None:
-    assert StorageType.FILE is not None
-    assert StorageType.MEMORY is not None
-    assert StorageType.FILE != StorageType.MEMORY
-
-
-async def test_discard_policy_values() -> None:
-    assert DiscardPolicy.OLD is not None
-    assert DiscardPolicy.NEW is not None
-    assert DiscardPolicy.OLD != DiscardPolicy.NEW
-
-
-async def test_retention_policy_values() -> None:
-    assert RetentionPolicy.LIMITS is not None
-    assert RetentionPolicy.INTEREST is not None
-    assert RetentionPolicy.WORKQUEUE is not None
-
-
-async def test_compression_values() -> None:
-    assert Compression.S2 is not None
-    assert Compression.NONE is not None
-    assert Compression.S2 != Compression.NONE
-
-
-async def test_persistence_mode_values() -> None:
-    assert PersistenceMode.Default is not None
-    assert PersistenceMode.Async is not None
-
-
-async def test_deliver_policy_values() -> None:
-    assert DeliverPolicy.ALL is not None
-    assert DeliverPolicy.LAST is not None
-    assert DeliverPolicy.NEW is not None
-    assert DeliverPolicy.BY_START_SEQUENCE is not None
-    assert DeliverPolicy.BY_START_TIME is not None
-    assert DeliverPolicy.LAST_PER_SUBJECT is not None
-
-
-async def test_ack_policy_values() -> None:
-    assert AckPolicy.EXPLICIT is not None
-    assert AckPolicy.NONE is not None
-    assert AckPolicy.ALL is not None
-
-
-async def test_replay_policy_values() -> None:
-    assert ReplayPolicy.INSTANT is not None
-    assert ReplayPolicy.ORIGINAL is not None
-
-
 async def test_stream_config_defaults() -> None:
     config = StreamConfig(name="test", subjects=["test.>"])
     assert config.name == "test"

python/tests/test_consumers.py

@@ -21,50 +21,29 @@ async def js(nats: Nats) -> JetStream:
 
 
 async def test_pull_consumer_create(js: JetStream) -> None:
-    stream_name = f"test-pcreate-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pcreate-{uuid.uuid4()}"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
         )
         consumer = await stream.consumers.create(consumer_config)
         assert isinstance(consumer, PullConsumer)
     finally:
         await js.streams.delete(stream_name)
 
 
-async def test_pull_consumer_fetch(js: JetStream) -> None:
-    stream_name = f"test-pfetch-{uuid.uuid4().hex[:8]}"
-    subj = f"{stream_name}.data"
-    config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
-    stream = await js.streams.create(config)
-    try:
-        await js.publish(subj, b"fetch-msg-1")
-        await js.publish(subj, b"fetch-msg-2")
-
-        consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
-        )
-        consumer = await stream.consumers.create(consumer_config)
-        messages = await consumer.fetch(max_messages=2, timeout=5.0)
-        assert len(messages) == 2
-        assert messages[0].payload == b"fetch-msg-1"
-        assert messages[1].payload == b"fetch-msg-2"
-    finally:
-        await js.streams.delete(stream_name)
-
-
 async def test_pull_consumer_fetch_with_ack(js: JetStream) -> None:
-    stream_name = f"test-pack-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pack-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(subj, b"ack-msg")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             ack_policy=AckPolicy.EXPLICIT,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -76,15 +55,15 @@ async def test_pull_consumer_fetch_with_ack(js: JetStream) -> None:
 
 
 async def test_pull_consumer_nack(js: JetStream) -> None:
-    stream_name = f"test-pnack-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pnack-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(subj, b"nack-msg")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             ack_policy=AckPolicy.EXPLICIT,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -96,15 +75,15 @@ async def test_pull_consumer_nack(js: JetStream) -> None:
 
 
 async def test_pull_consumer_term(js: JetStream) -> None:
-    stream_name = f"test-pterm-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pterm-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(subj, b"term-msg")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             ack_policy=AckPolicy.EXPLICIT,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -116,15 +95,15 @@ async def test_pull_consumer_term(js: JetStream) -> None:
 
 
 async def test_pull_consumer_progress(js: JetStream) -> None:
-    stream_name = f"test-pprog-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pprog-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(subj, b"progress-msg")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             ack_policy=AckPolicy.EXPLICIT,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -137,14 +116,14 @@ async def test_pull_consumer_progress(js: JetStream) -> None:
 
 
 async def test_pull_consumer_message_properties(js: JetStream) -> None:
-    stream_name = f"test-pmsgprop-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pmsgprop-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(subj, b"prop-msg")
 
-        consumer_name = f"consumer-{uuid.uuid4().hex[:8]}"
+        consumer_name = f"consumer-{uuid.uuid4()}"
         consumer_config = PullConsumerConfig(name=consumer_name)
         consumer = await stream.consumers.create(consumer_config)
         messages = await consumer.fetch(max_messages=1, timeout=5.0)
@@ -166,15 +145,15 @@ async def test_pull_consumer_message_properties(js: JetStream) -> None:
 
 
 async def test_pull_consumer_with_filter_subject(js: JetStream) -> None:
-    stream_name = f"test-pfilter-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pfilter-{uuid.uuid4()}"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         await js.publish(f"{stream_name}.a", b"msg-a")
         await js.publish(f"{stream_name}.b", b"msg-b")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             filter_subject=f"{stream_name}.a",
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -187,7 +166,7 @@ async def test_pull_consumer_with_filter_subject(js: JetStream) -> None:
 
 
 async def test_pull_consumer_deliver_policy(js: JetStream) -> None:
-    stream_name = f"test-pdeliver-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pdeliver-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
@@ -196,7 +175,7 @@ async def test_pull_consumer_deliver_policy(js: JetStream) -> None:
         await js.publish(subj, b"new-msg")
 
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             deliver_policy=DeliverPolicy.LAST,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -208,12 +187,12 @@ async def test_pull_consumer_deliver_policy(js: JetStream) -> None:
 
 
 async def test_pull_consumer_replay_policy(js: JetStream) -> None:
-    stream_name = f"test-preplay-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-preplay-{uuid.uuid4()}"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         consumer_config = PullConsumerConfig(
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
             replay_policy=ReplayPolicy.INSTANT,
         )
         consumer = await stream.consumers.create(consumer_config)
@@ -223,12 +202,11 @@ async def test_pull_consumer_replay_policy(js: JetStream) -> None:
 
 
 async def test_pull_consumer_durable(js: JetStream) -> None:
-    stream_name = f"test-pdurable-{uuid.uuid4().hex[:8]}"
-    subj = f"{stream_name}.data"
+    stream_name = f"test-pdurable-{uuid.uuid4()}"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
-        durable_name = f"durable-{uuid.uuid4().hex[:8]}"
+        durable_name = f"durable-{uuid.uuid4()}"
         consumer_config = PullConsumerConfig(
             durable_name=durable_name,
         )
@@ -242,41 +220,59 @@ async def test_pull_consumer_durable(js: JetStream) -> None:
 
 
 async def test_push_consumer_create(js: JetStream) -> None:
-    stream_name = f"test-pushcreate-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pushcreate-{uuid.uuid4()}"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
     stream = await js.streams.create(config)
     try:
         deliver_subj = uuid.uuid4().hex
         consumer_config = PushConsumerConfig(
             deliver_subject=deliver_subj,
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
         )
         consumer = await stream.consumers.create(consumer_config)
         assert isinstance(consumer, PushConsumer)
     finally:
         await js.streams.delete(stream_name)
 
 
+async def test_pull_consumer_messages(js: JetStream) -> None:
+    stream_name = f"test-pushmsg-{uuid.uuid4()}"
+    subj = f"{stream_name}.data"
+    config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
+    messages = [uuid.uuid4().hex.encode(), uuid.uuid4().hex.encode()]
+    stream = await js.streams.create(config)
+    try:
+        for message in messages:
+            await js.publish(subj, message)
+        consumer_config = PullConsumerConfig(name=f"consumer-{uuid.uuid4()}")
+        consumer = await stream.consumers.create(consumer_config)
+        msgs_iter = await consumer.fetch(timeout=0.5)
+        for nats_msg, payload in zip(msgs_iter, messages, strict=True):
+            assert nats_msg.payload == payload
+    finally:
+        await js.streams.delete(stream_name)
+
+
 async def test_push_consumer_messages(js: JetStream) -> None:
-    stream_name = f"test-pushmsg-{uuid.uuid4().hex[:8]}"
+    stream_name = f"test-pushmsg-{uuid.uuid4()}"
     subj = f"{stream_name}.data"
     config = StreamConfig(name=stream_name, subjects=[f"{stream_name}.>"])
+    messages = [uuid.uuid4().hex.encode(), uuid.uuid4().hex.encode()]
     stream = await js.streams.create(config)
     try:
-        await js.publish(subj, b"push-msg-1")
-        await js.publish(subj, b"push-msg-2")
+        for message in messages:
+            await js.publish(subj, message)
 
         deliver_subj = uuid.uuid4().hex
         consumer_config = PushConsumerConfig(
             deliver_subject=deliver_subj,
-            name=f"consumer-{uuid.uuid4().hex[:8]}",
+            name=f"consumer-{uuid.uuid4()}",
         )
         consumer = await stream.consumers.create(consumer_config)
         msgs_iter = await consumer.messages()
-        msg1 = await msgs_iter.next(timeout=5.0)
-        msg2 = await msgs_iter.next(timeout=5.0)
-        assert msg1.payload == b"push-msg-1"
-        assert msg2.payload == b"push-msg-2"
+        for message in messages:
+            nats_msg = await msgs_iter.next(timeout=0.5)
+            assert message == nats_msg.payload
     finally:
         await js.streams.delete(stream_name)
 

python/tests/test_kv.py

@@ -4,8 +4,8 @@
 from natsrpy import Nats
 from natsrpy.js import (
     JetStream,
-    KVConfig,
     KeyValue,
+    KVConfig,
     StorageType,
 )
 

python/tests/test_request_reply.py

@@ -67,7 +67,7 @@ async def responder() -> None:
     task = asyncio.create_task(responder())
     await asyncio.sleep(0.1)
 
-    await nats.request(subj, None)
+    await nats.request(subj, b"")
     await task
 
     assert received_payload[0] == b""

python/tests/test_streams.py

@@ -3,7 +3,6 @@
 import pytest
 from natsrpy import Nats
 from natsrpy.js import (
-    Compression,
     DiscardPolicy,
     JetStream,
     RetentionPolicy,