adding offsets to initialize and adding clear_queue script

clear_queue.py

@@ -0,0 +1,54 @@
+import sys, getopt, pika
+from pika import BlockingConnection
+
+DEBUG = False
+RUN_NUMBER = 1
+
+def usage():
+  print "python clear_queue.py -r <run> --debug"
+
+try:
+  opts, args = getopt.getopt(sys.argv[1:], "hr:dv", ["help", "run=", "debug", "verbose"])
+except getopt.GetoptError, err:
+  # print help information and exit:
+  print str(err) # will print something like "option -a not recognized"
+  usage()
+  sys.exit(2)
+
+for o, a in opts:
+  if o in ("-v", "--verbose"):
+    VERBOSE = True
+  elif o in ("-h", "--help"):
+    usage()
+    sys.exit()
+  elif o in ("-r", "--run"):
+    RUN_NUMBER = int(a)
+  elif o in ("-d", "--debug"):
+    DEBUG = True
+  else:
+    assert False, "unhandled option"
+
+# Connect to RabbitMQ
+TARGET_RMQ_SERVER = "ldr.myvnc.com" if DEBUG else "noddy.cs.berkeley.edu"
+parameters = pika.ConnectionParameters(TARGET_RMQ_SERVER)
+rmq_connection = BlockingConnection(parameters)
+rmq_channel = rmq_connection.channel()
+
+rmq_channel.queue_declare(queue="run%d" % RUN_NUMBER, durable=True,
+                          exclusive=False, auto_delete=False)
+
+num_cleared = 0
+def handle_delivery(channel, method_frame, header_frame, body):
+  # Receive the data in 3 frames from RabbitMQ
+  if VERBOSE:
+    pika.log.info("Basic.Deliver %s delivery-tag %i: %s",
+                  header_frame.content_type,
+                  method_frame.delivery_tag,
+                  body)
+  rmq_channel.basic_ack(delivery_tag=method_frame.delivery_tag)
+  num_cleared += 1
+  print "Cleared %d messages" % (num_cleared)
+
+# We're stuck looping here since this is a blocking adapter
+rmq_channel.basic_consume(handle_delivery, queue='run%d' % RUN_NUMBER)
+rmq_channel.start_consuming()

initialize_queue.py

@@ -12,12 +12,13 @@
 DEPTH = None
 NUM_SITES = None
 DEBUG = False
+OFFSET = 0
 
 def usage():
-  print "python initialize_queue.py -r <run> -n <num_sites> -f <fanout> -d <depth> --debug"
+  print "python initialize_queue.py -r <run> -n <num_sites> -f <fanout> -d <depth> -o <offset> --debug"
 
 try:
-  opts, args = getopt.getopt(sys.argv[1:], "hr:n:f:d:v", ["help", "run=", "num=", "fanout=", "depth=", "verbose", "debug"])
+  opts, args = getopt.getopt(sys.argv[1:], "hr:n:f:d:vo:", ["help", "run=", "num=", "fanout=", "depth=", "verbose", "debug", "offset="])
 except getopt.GetoptError, err:
   # print help information and exit:
   print str(err) # will print something like "option -a not recognized"
@@ -40,6 +41,8 @@ def usage():
     DEPTH = int(a)
   elif o in ("--debug"):
     DEBUG = True
+  elif o in ("-o", "--offset"):
+    OFFSET = int(a)
   else:
     assert False, "unhandled option"
 
@@ -67,6 +70,9 @@ def usage():
 
 n = 1
 for row in reader:
+    if n < OFFSET:
+      n += 1
+      continue
     url = "http://%s" % row[1]
     command_data = {
         'run': RUN_NUMBER,
@@ -83,7 +89,7 @@ def usage():
             delivery_mode=1))
     print "Delivered"
     n += 1
-    if n > NUM_SITES:
+    if n > NUM_SITES + OFFSET:
         break
 
 rmq_connection.close()