Allow threaded operation, including a handler that places the message…

…s on the queue and an event to stop the thread.

gcn/cmdline.py

@@ -17,6 +17,7 @@
 """
 Utilities for command-line interface.
 """
+from __future__ import print_function
 import argparse
 import collections
 import logging
@@ -147,6 +148,67 @@ def inthandler(signum, frame):
     print('\nFinishing')
     thread.join()
 
+def threaded_listen_main(args=None):
+    """Example VOEvent listener that demonstrates threaded operation"""
+
+    # Command line interface
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('addr', default='68.169.57.253:8099',
+                        action=HostPortAction,
+                        help='Server host and port (default: %(default)s)')
+    parser.add_argument('--version', action='version',
+                        version='pygcn ' + __version__)
+    parser.add_argument('--maxtime', default=None,
+                        help='Time to process until returning (s)')
+    args = parser.parse_args(args)
+
+    if args.maxtime is not None:
+        args.maxtime = datetime.timedelta(seconds=float(args.maxtime))
+
+    # Set up logger
+    logging.basicConfig(level=logging.INFO)
+
+    # Listen for GCN notices (until interrupted, killed, or maxtime reached)
+    # in a second thread, while counting up seconds in the main thread.
+    messagequeue = queue.Queue()
+    stopevent = threading.Event()
+
+    def inthandler(signum, frame):
+        stopevent.set()
+
+    signal.signal(signal.SIGINT, handler=inthandler)  # Keyboard etc interrupt
+    try:
+        listenargs = dict(host=args.addr.host, port=args.addr.port,
+                          handler=handlers.queuehandlerfor(messagequeue),
+                          stopevent=stopevent)
+        thread = threading.Thread(target=listen, kwargs=listenargs)
+        starttime = datetime.datetime.utcnow()
+        lasttime = starttime
+        thread.start()
+
+        while thread.is_alive():
+            try:
+                payload, root = messagequeue.get(timeout=1)
+                print('\r{} {}'
+                      .format(datetime.datetime.utcnow().strftime("%H:%M:%S"),
+                              root.attrib['ivorn']))
+                lasttime = datetime.datetime.utcnow()
+            except queue.Empty:
+                dt = (datetime.datetime.utcnow() - lasttime).total_seconds()
+                print('\r{:.0f}'.format(dt), end='\r')
+            if args.maxtime is not None:
+                if (datetime.datetime.utcnow() - starttime) > args.maxtime:
+                    stopevent.set()
+                    break
+    except Exception as e:
+        stopevent.set()
+        print(e, file=sys.stderr)
+        thread.join()
+        raise
+    print('\nFinishing')
+    thread.join()
+
+
 def serve_main(args=None):
     """Rudimentary GCN server, for testing purposes. Serves just one connection
     at a time, and repeats the same payloads in order, repeating, for each

gcn/handlers.py

@@ -87,14 +87,16 @@ def archive(payload, root):
         f.write(payload)
     logging.getLogger('gcn.handlers.archive').info("archived %s", ivorn)
 
+
 def _queuehandler(payload, root, queue=None):
     """ Place (payload, root) on queue for threaded operation.
     This can be used in the following manner:
         gcn.listen(handler = functools.partial(partialize_queue, queue=a_queue))
     """
     if queue is None:
         raise TypeError("The queue must be set (use queuehandlerfor())")
-    queue.put( (payload,root) )
+    queue.put((payload, root))
+
 
 def queuehandlerfor(queue):
     """Create a handler that places (payload, root) on the given queue

gcn/tests/test_handlers.py

@@ -7,6 +7,10 @@
 from .. import handlers
 from .. import notice_types
 
+try:
+    import queue
+except:
+    import Queue as queue
 
 payloads = [pkg_resources.resource_string(__name__, 'data/gbm_flt_pos.xml'),
             pkg_resources.resource_string(__name__, 'data/kill_socket.xml')]
@@ -52,3 +56,24 @@ def test_archive(tmpdir):
             assert (tmpdir / filename).exists()
     finally:
         os.chdir(old_dir)
+
+
+def test_queuehandler():
+    queue_ = queue.Queue()
+    queuehandler = handlers.queuehandlerfor(queue_)
+    assert queue_.empty()
+
+    for payload in payloads:
+        queuehandler(payload, fromstring(payload))
+        qpayload, qtree = queue_.get()
+        assert qpayload == payload
+
+    assert queue_.empty()
+
+    for payload in payloads:
+        queuehandler(payload, fromstring(payload))
+    for payload in payloads:
+        qpayload, qtree = queue_.get()
+        assert qpayload == payload
+
+    assert queue_.empty()