Skip to content

Commit

Permalink
Split up topic names between tests
Browse files Browse the repository at this point in the history
Signed-off-by: Elena Kolevska <elena@kolevska.com>
  • Loading branch information
elena-kolevska committed Oct 21, 2024
1 parent fc0c5fa commit 09576c8
Show file tree
Hide file tree
Showing 4 changed files with 41 additions and 18 deletions.
28 changes: 14 additions & 14 deletions examples/pubsub-streaming-async/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,11 @@ Run the following command in a terminal/command prompt:
<!-- STEP
name: Run subscriber
expected_stdout_lines:
- "== APP == Processing message: {'id': 1, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 2, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 1, 'message': 'hello world'} from TOPIC_B1..."
- "== APP == Processing message: {'id': 2, 'message': 'hello world'} from TOPIC_B1..."
- "== APP == Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_B1..."
- "== APP == Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_B1..."
- "== APP == Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_B1..."
- "== APP == Closing subscription..."
output_match_mode: substring
background: true
Expand All @@ -41,7 +41,7 @@ sleep: 3

```bash
# 1. Start Subscriber
dapr run --app-id python-subscriber --app-protocol grpc python3 subscriber.py
dapr run --app-id python-subscriber --app-protocol grpc -- python3 subscriber.py --topic=TOPIC_B1
```

<!-- END_STEP -->
Expand All @@ -63,7 +63,7 @@ sleep: 15

```bash
# 2. Start Publisher
dapr run --app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 --enable-app-health-check python3 publisher.py
dapr run --app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 --enable-app-health-check -- python3 publisher.py --topic=TOPIC_B1
```

<!-- END_STEP -->
Expand All @@ -75,11 +75,11 @@ Run the following command in a terminal/command prompt:
<!-- STEP
name: Run subscriber
expected_stdout_lines:
- "== APP == Processing message: {'id': 1, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 2, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_A..."
- "== APP == Processing message: {'id': 1, 'message': 'hello world'} from TOPIC_B2..."
- "== APP == Processing message: {'id': 2, 'message': 'hello world'} from TOPIC_B2..."
- "== APP == Processing message: {'id': 3, 'message': 'hello world'} from TOPIC_B2..."
- "== APP == Processing message: {'id': 4, 'message': 'hello world'} from TOPIC_B2..."
- "== APP == Processing message: {'id': 5, 'message': 'hello world'} from TOPIC_B2..."
- "== APP == Closing subscription..."
output_match_mode: substring
background: true
Expand All @@ -89,7 +89,7 @@ sleep: 3

```bash
# 1. Start Subscriber
dapr run --app-id python-subscriber --app-protocol grpc python3 subscriber-handler.py
dapr run --app-id python-subscriber --app-protocol grpc -- python3 subscriber-handler.py --topic=TOPIC_B2
```

<!-- END_STEP -->
Expand All @@ -111,7 +111,7 @@ sleep: 15

```bash
# 2. Start Publisher
dapr run --app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 --enable-app-health-check python3 publisher.py
dapr run --app-id python-publisher --app-protocol grpc --dapr-grpc-port=3500 --enable-app-health-check -- python3 publisher.py --topic=TOPIC_B2
```

<!-- END_STEP -->
Expand Down
9 changes: 8 additions & 1 deletion examples/pubsub-streaming-async/publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,18 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# ------------------------------------------------------------
import argparse
import asyncio
import json

from dapr.aio.clients import DaprClient

parser = argparse.ArgumentParser(description='Publish events to a Dapr pub/sub topic.')
parser.add_argument('--topic', type=str, required=True, help='The topic name to publish to.')
args = parser.parse_args()

topic_name = args.topic


async def publish_events():
"""
Expand All @@ -30,7 +37,7 @@ async def publish_events():
# Create a typed message with content type and body
await d.publish_event(
pubsub_name='pubsub',
topic_name='TOPIC_A',
topic_name=topic_name,
data=json.dumps(req_data),
data_content_type='application/json',
publish_metadata={'ttlInSeconds': '100', 'rawPayload': 'false'},
Expand Down
12 changes: 10 additions & 2 deletions examples/pubsub-streaming-async/subscriber-handler.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,15 @@
import argparse
import asyncio
from dapr.aio.clients import DaprClient
from dapr.clients.grpc._response import TopicEventResponse

parser = argparse.ArgumentParser(description='Publish events to a Dapr pub/sub topic.')
parser.add_argument('--topic', type=str, required=True, help='The topic name to publish to.')
args = parser.parse_args()

topic_name = args.topic
dlq_topic_name = topic_name + '_DEAD'

counter = 0


Expand All @@ -24,9 +32,9 @@ async def main():
# Subscribe to the pubsub topic with the message handler
close_fn = await client.subscribe_with_handler(
pubsub_name='pubsub',
topic='TOPIC_A',
topic=topic_name,
handler_fn=process_message,
dead_letter_topic='TOPIC_A_DEAD',
dead_letter_topic=dlq_topic_name,
)

# Wait until 5 messages are processed
Expand Down
10 changes: 9 additions & 1 deletion examples/pubsub-streaming-async/subscriber.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,16 @@
import argparse
import asyncio

from dapr.aio.clients import DaprClient
from dapr.clients.grpc.subscription import StreamInactiveError

parser = argparse.ArgumentParser(description='Publish events to a Dapr pub/sub topic.')
parser.add_argument('--topic', type=str, required=True, help='The topic name to publish to.')
args = parser.parse_args()

topic_name = args.topic
dlq_topic_name = topic_name + '_DEAD'

counter = 0


Expand All @@ -18,7 +26,7 @@ async def main():
async with DaprClient() as client:
global counter
subscription = await client.subscribe(
pubsub_name='pubsub', topic='TOPIC_A', dead_letter_topic='TOPIC_A_DEAD'
pubsub_name='pubsub', topic=topic_name, dead_letter_topic=dlq_topic_name
)

try:
Expand Down

0 comments on commit 09576c8

Please sign in to comment.