1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
#!/bin/sh
# Reset rabbitmq
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
cat << EOF > new_task.py
#!/usr/bin/env python3
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(delivery_mode=2) # Set delivery mode to persistent (2)
)
print(f" [x] Sent {message}")
connection.close()
EOF
cat << EOF > worker.py
#!/usr/bin/env python3
import pika
import time
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(f" [x] Received {body.decode()}")
time.sleep(body.count(b'.'))
print(" [x] Done")
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()
EOF
# Create 5 queues messages
python3 new_task.py First message.
python3 new_task.py Second message..
python3 new_task.py Third message...
python3 new_task.py Fourth message....
python3 new_task.py Fifth message.....
# check the queued messages, we shoudl have 5 such as:
# # rabbitmqctl list_queues name messages_ready messages_unacknowledged
# Timeout: 60.0 seconds ...
# Listing queues for vhost / ...
# name messages_ready messages_unacknowledged
# hello 0 0
# task_queue 5 0
#
# Check that the task queue exists and that there is 5 messages
echo "Expected:"
echo "task_queue 5 0"
echo "Actual:"
tqueue=$(rabbitmqctl list_queues name messages_ready messages_unacknowledged | grep task_queue)
echo $tqueue
if echo $tqueue | awk '{print $2}' | grep -q "^5$"; then
echo "The 'task_queue' queue exists with 5 message."
else
echo "Error: error checking the task_queue"
exit 1
fi
# Now create 2 workers
python3 worker.py &
python3 worker.py &
# let the workers do their job
sleep 13
# Now there should be no messages_ready and no messages_unacknowledged such as:
# # rabbitmqctl list_queues name messages_ready messages_unacknowledged
# Timeout: 60.0 seconds ...
# Listing queues for vhost / ...
# name messages_ready messages_unacknowledged
# hello 0 0
# task_queue 0 0
echo "Expected:"
echo "task_queue 0 0"
echo "Actual:"
tqueue=$(rabbitmqctl list_queues name messages_ready messages_unacknowledged | grep task_queue)
echo $tqueue
if echo $tqueue | grep task_queue | awk '{print $2}' | grep -q "^0"; then
echo "The 'task_queue' queue exists with 0 messages_ready."
else
echo "Error: checking the task_queue messages_ready"
exit 1
fi
if echo $tqueue | awk '{print $3}' | grep -q "^0"; then
echo "The 'task_queue' queue exists with 0 messages_unacknowledged."
else
echo "Error: checking the task_queue messages_unacknowledged"
exit 1
fi
|
