Start unwinding unnecessary uses of asynch I/O.

svn path=/branches/tk705/; revision=5920

1 files changed, 18 insertions, 23 deletions

diff --git a/rpki/pubd.py b/rpki/pubd.py
index b6788ebd..a2f996b6 100644
--- a/rpki/pubd.py
+++ b/rpki/pubd.py
@@ -28,11 +28,11 @@ import time
 import socket
 import logging
 import argparse
+
 import rpki.resource_set
 import rpki.up_down
 import rpki.x509
 import rpki.sql
-import rpki.http
 import rpki.config
 import rpki.exceptions
 import rpki.relaxng
@@ -40,6 +40,7 @@ import rpki.log
 import rpki.publication
 import rpki.publication_control
 import rpki.daemonize
+import rpki.http_simple
 
 from lxml.etree import Element, SubElement, tostring as ElementToString
 
@@ -50,6 +51,8 @@ rrdp_xmlns   = rpki.relaxng.rrdp.xmlns
 rrdp_nsmap   = rpki.relaxng.rrdp.nsmap
 rrdp_version = "1"
 
+rpki_content_type = "application/x-rpki"
+
 
 def DERSubElement(elt, name, der, attrib = None, **kwargs):
   """
@@ -133,7 +136,7 @@ class main(object):
 
     self.session = session_obj.fetch(self)
 
-    rpki.http.server(
+    rpki.http_simple.server(
       host     = self.http_server_host,
       port     = self.http_server_port,
       handlers = (("/control", self.control_handler),
@@ -144,44 +147,41 @@ class main(object):
     return "%s/%s" % (self.rrdp_uri_base.rstrip("/"), fn)
 
 
-  def control_handler(self, query, path, cb):
+  def control_handler(self, request, q_der):
     """
     Process one PDU from the IRBE.
     """
 
     def done(r_msg):
       self.sql.sweep()
-      cb(code = 200,
-         body = rpki.publication_control.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert))
+      request.send_cms_response(rpki.publication_control.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert))
 
     try:
-      q_cms = rpki.publication_control.cms_msg(DER = query)
+      q_cms = rpki.publication_control.cms_msg(DER = q_der)
       q_msg = q_cms.unwrap((self.bpki_ta, self.irbe_cert))
       self.irbe_cms_timestamp = q_cms.check_replay(self.irbe_cms_timestamp, "control")
       q_msg.serve_top_level(self, done)
-    except (rpki.async.ExitNow, SystemExit):
-      raise
     except Exception, e:
-      logger.exception("Unhandled exception processing control query, path %r", path)
-      cb(code = 500, reason = "Unhandled exception %s: %s" % (e.__class__.__name__, e))
+      logger.exception("Unhandled exception processing control query, path %r", request.path)
+      request.send_error(500, "Unhandled exception %s: %s" % (e.__class__.__name__, e))
 
 
   client_url_regexp = re.compile("/client/([-A-Z0-9_/]+)$", re.I)
 
-  def client_handler(self, query, path, cb):
+  def client_handler(self, request, q_der):
     """
     Process one PDU from a client.
     """
 
     try:
-      match = self.client_url_regexp.search(path)
+      match = self.client_url_regexp.search(request.path)
       if match is None:
-        raise rpki.exceptions.BadContactURL("Bad path: %s" % path)
+        raise rpki.exceptions.BadContactURL("Bad path: %s" % request.path)
       client_handle = match.group(1)
       client = rpki.publication_control.client_elt.sql_fetch_where1(self, "client_handle = %s", (client_handle,))
       if client is None:
         raise rpki.exceptions.ClientNotFound("Could not find client %s" % client_handle)
-      q_cms = rpki.publication.cms_msg(DER = query)
+      q_cms = rpki.publication.cms_msg(DER = q_der)
       q_msg = q_cms.unwrap((self.bpki_ta, client.bpki_cert, client.bpki_glue))
       q_cms.check_replay_sql(client, client.client_handle)
       if not q_msg.is_query():
@@ -209,8 +209,6 @@ class main(object):
             r_pdu.tag = q_pdu.tag
             r_pdu.uri = q_pdu.uri
             r_msg.append(r_pdu)
-        except (rpki.async.ExitNow, SystemExit):
-          raise
         except Exception, e:
           if not isinstance(e, rpki.exceptions.NotFound):
             logger.exception("Exception processing PDU %r", q_pdu)
@@ -246,14 +244,11 @@ class main(object):
         # database of record.  This may require doing the filesystem
         # updates from the delta, but that should be straightforward.
 
-      cb(code = 200,
-         body = rpki.publication.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert, self.pubd_crl))
-    except (rpki.async.ExitNow, SystemExit):
-      raise
+      request.send_cms_response(rpki.publication.cms_msg().wrap(r_msg, self.pubd_key, self.pubd_cert, self.pubd_crl))
+
     except Exception, e:
-      logger.exception("Unhandled exception processing client query, path %r", path)
-      cb(code = 500,
-         reason = "Could not process PDU: %s" % e)
+      logger.exception("Unhandled exception processing client query, path %r", request.path)
+      request.send_error(500, "Could not process PDU: %s" % e)
 
   def uri_to_filename(self, uri):
     """