|
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import argparse
|
|
import pika
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--host', default='localhost', action='store', dest='host', help='server host')
|
|
parser.add_argument('--port', default=5672, type=int, action='store', dest='port', help='server port')
|
|
parser.add_argument('--ssl', default=False, action='store_true', dest='ssl', help='use ssl')
|
|
parser.add_argument('--user', default='myuser', action='store', dest='user', help='username')
|
|
parser.add_argument('--password', default='mypass', action='store', dest='password', help='password')
|
|
parser.add_argument('--name', default='myq', action='store', dest='name', help='queue name')
|
|
parser.add_argument('--file', required=True, action='store', dest='file', help='import filename')
|
|
parser.add_argument('--keep', default=False, action='store_true', dest='keep', help='keep msg in queue, copy only')
|
|
args = parser.parse_args()
|
|
|
|
parameters = pika.ConnectionParameters(host=args.host,
|
|
port=args.port,
|
|
virtual_host='/',
|
|
credentials=pika.PlainCredentials(args.user, args.password),
|
|
ssl=args.ssl)
|
|
|
|
connection = pika.BlockingConnection(parameters)
|
|
if connection.is_open:
|
|
channel = connection.channel()
|
|
channel.queue_declare(queue=args.name, durable=True)
|
|
print(' [*] Waiting for messages, timeout 2 sec')
|
|
with open(args.file, 'w') as ptr:
|
|
CNT = 1
|
|
for data in channel.consume(args.name, inactivity_timeout=2):
|
|
if data:
|
|
ptr.write("%s\n" % data[2])
|
|
if args.keep:
|
|
print(" [x] copy from %s cnt(%s)" % (args.name, CNT))
|
|
else:
|
|
channel.basic_ack(data[0].delivery_tag)
|
|
print(" [x] move from %s cnt(%s)" % (args.name, CNT))
|
|
CNT += 1
|
|
else:
|
|
print(' [*] empty, break')
|
|
break
|
|
ptr.close()
|
|
|
|
connection.close()
|