QPID-4967 - Router code advances

  o Fixed handling of SASL on outbound connections
  o Added Send and Receive message paths in and out of Python modules
  o Overhauled the route-table data structures
    - Multicasting is now supported (multiple sender links with the same address)
    - Support has been added for message-based routing semantics as well as link-based
  o Two Dispatch processes connected to each other will now discover each other as neighbors


git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1502698 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 67 insertions, 3 deletions

diff --git a/qpid/extras/dispatch/src/python_embedded.c b/qpid/extras/dispatch/src/python_embedded.c
index 6b5250a34a..ffa5bda6b8 100644
--- a/qpid/extras/dispatch/src/python_embedded.c
+++ b/qpid/extras/dispatch/src/python_embedded.c
@@ -396,6 +396,7 @@ static PyTypeObject LogAdapterType = {
 typedef struct {
     PyObject_HEAD
     PyObject       *handler;
+    PyObject       *handler_rx_call;
     dx_dispatch_t  *dx;
     dx_address_t   *address;
 } IoAdapter;
@@ -403,9 +404,66 @@ typedef struct {
 
 static void dx_io_rx_handler(void *context, dx_message_t *msg)
 {
-    //IoAdapter *self = (IoAdapter*) context;
+    IoAdapter *self = (IoAdapter*) context;
+
+    //
+    // Parse the message through the body and exit if the message is not well formed.
+    //
+    if (!dx_message_check(msg, DX_DEPTH_BODY))
+        return;
+
+    //
+    // Get an iterator for the application-properties.  Exit if the message has none.
+    //
+    dx_field_iterator_t *ap = dx_message_field_iterator(msg, DX_FIELD_APPLICATION_PROPERTIES);
+    if (ap == 0)
+        return;
+
+    //
+    // Try to get a map-view of the application-properties.
+    //
+    dx_parsed_field_t *ap_map = dx_parse(ap);
+    if (ap_map == 0 || !dx_parse_ok(ap_map) || !dx_parse_is_map(ap_map)) {
+        dx_field_iterator_free(ap);
+        dx_parse_free(ap_map);
+        return;
+    }
+
+    //
+    // Get an iterator for the body.  Exit if the message has none.
+    //
+    dx_field_iterator_t *body = dx_message_field_iterator(msg, DX_FIELD_BODY);
+    if (body == 0) {
+        dx_field_iterator_free(ap);
+        dx_parse_free(ap_map);
+        return;
+    }
+
+    //
+    // Try to get a map-view of the body.
+    //
+    dx_parsed_field_t *body_map = dx_parse(body);
+    if (body_map == 0 || !dx_parse_ok(body_map) || !dx_parse_is_map(body_map)) {
+        printf("XXXX %s\n", dx_parse_error(body_map));
+        dx_field_iterator_free(ap);
+        dx_field_iterator_free(body);
+        dx_parse_free(ap_map);
+        dx_parse_free(body_map);
+        return;
+    }
 
-    // TODO - Parse the incoming message and send it to the python handler.
+    PyObject *pAP   = dx_field_to_py(ap_map);
+    PyObject *pBody = dx_field_to_py(body_map);
+
+    PyObject *pArgs = PyTuple_New(2);
+    PyTuple_SetItem(pArgs, 0, pAP);
+    PyTuple_SetItem(pArgs, 1, pBody);
+
+    PyObject *pValue = PyObject_CallObject(self->handler_rx_call, pArgs);
+    Py_DECREF(pArgs);
+    if (pValue) {
+        Py_DECREF(pValue);
+    }
 }
 
 
@@ -415,9 +473,14 @@ static int IoAdapter_init(IoAdapter *self, PyObject *args, PyObject *kwds)
     if (!PyArg_ParseTuple(args, "Os", &self->handler, &address))
         return -1;
 
+    self->handler_rx_call = PyObject_GetAttrString(self->handler, "receive");
+    if (!self->handler_rx_call || !PyCallable_Check(self->handler_rx_call))
+        return -1;
+
     Py_INCREF(self->handler);
+    Py_INCREF(self->handler_rx_call);
     self->dx = dispatch;
-    self->address = dx_router_register_address(self->dx, true, address, dx_io_rx_handler, self);
+    self->address = dx_router_register_address(self->dx, address, dx_io_rx_handler, self);
     return 0;
 }
 
@@ -426,6 +489,7 @@ static void IoAdapter_dealloc(IoAdapter* self)
 {
     dx_router_unregister_address(self->address);
     Py_DECREF(self->handler);
+    Py_DECREF(self->handler_rx_call);
     self->ob_type->tp_free((PyObject*)self);
 }