/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ #include #include #include #include #include #include //=============================================================================== // Control Functions //=============================================================================== static qd_dispatch_t *dispatch = 0; static uint32_t ref_count = 0; static sys_mutex_t *ilock = 0; static qd_log_source_t *log_source = 0; static PyObject *dispatch_module = 0; static PyObject *dispatch_python_pkgdir = 0; static qd_address_semantics_t py_semantics = QD_FANOUT_MULTIPLE | QD_BIAS_NONE | QD_CONGESTION_DROP | QD_DROP_FOR_SLOW_CONSUMERS; static void qd_python_setup(void); void qd_python_initialize(qd_dispatch_t *qd, const char *python_pkgdir) { log_source = qd_log_source("PYTHON"); dispatch = qd; ilock = sys_mutex(); if (python_pkgdir) dispatch_python_pkgdir = PyString_FromString(python_pkgdir); } void qd_python_finalize(void) { assert(ref_count == 0); sys_mutex_free(ilock); } void qd_python_start(void) { sys_mutex_lock(ilock); if (ref_count == 0) { Py_Initialize(); qd_python_setup(); qd_log(log_source, QD_LOG_TRACE, "Embedded Python Interpreter Initialized"); } ref_count++; sys_mutex_unlock(ilock); } void qd_python_stop(void) { sys_mutex_lock(ilock); ref_count--; if (ref_count == 0) { Py_DECREF(dispatch_module); dispatch_module = 0; Py_Finalize(); qd_log(log_source, QD_LOG_TRACE, "Embedded Python Interpreter Shut Down"); } sys_mutex_unlock(ilock); } PyObject *qd_python_module(void) { assert(dispatch_module); return dispatch_module; } //=============================================================================== // Data Conversion Functions //=============================================================================== static PyObject *parsed_to_py_string(qd_parsed_field_t *field) { switch (qd_parse_tag(field)) { case QD_AMQP_VBIN8: case QD_AMQP_VBIN32: case QD_AMQP_STR8_UTF8: case QD_AMQP_STR32_UTF8: case QD_AMQP_SYM8: case QD_AMQP_SYM32: break; default: return Py_None; } #define SHORT_BUF 1024 uint8_t short_buf[SHORT_BUF]; PyObject *result; qd_field_iterator_t *raw = qd_parse_raw(field); qd_field_iterator_reset(raw); uint32_t length = qd_field_iterator_remaining(raw); uint8_t *buffer = short_buf; uint8_t *ptr; int alloc = 0; if (length > SHORT_BUF) { alloc = 1; buffer = (uint8_t*) malloc(length); } ptr = buffer; while (!qd_field_iterator_end(raw)) *(ptr++) = qd_field_iterator_octet(raw); result = PyString_FromStringAndSize((char*) buffer, ptr - buffer); if (alloc) free(buffer); return result; } void qd_py_to_composed(PyObject *value, qd_composed_field_t *field) { if (PyBool_Check(value)) qd_compose_insert_bool(field, PyInt_AS_LONG(value) ? 1 : 0); //else if (PyFloat_Check(value)) // qd_compose_insert_double(field, PyFloat_AS_DOUBLE(value)); else if (PyInt_Check(value)) qd_compose_insert_long(field, (int64_t) PyInt_AS_LONG(value)); else if (PyLong_Check(value)) qd_compose_insert_long(field, (int64_t) PyLong_AsLongLong(value)); else if (PyString_Check(value)) qd_compose_insert_string(field, PyString_AS_STRING(value)); else if (PyDict_Check(value)) { Py_ssize_t iter = 0; PyObject *key; PyObject *val; qd_compose_start_map(field); while (PyDict_Next(value, &iter, &key, &val)) { qd_py_to_composed(key, field); qd_py_to_composed(val, field); } qd_compose_end_map(field); } else if (PyList_Check(value)) { Py_ssize_t count = PyList_Size(value); qd_compose_start_list(field); for (Py_ssize_t idx = 0; idx < count; idx++) { PyObject *item = PyList_GetItem(value, idx); qd_py_to_composed(item, field); } qd_compose_end_list(field); } else if (PyTuple_Check(value)) { Py_ssize_t count = PyTuple_Size(value); qd_compose_start_list(field); for (Py_ssize_t idx = 0; idx < count; idx++) { PyObject *item = PyTuple_GetItem(value, idx); qd_py_to_composed(item, field); } qd_compose_end_list(field); } } PyObject *qd_field_to_py(qd_parsed_field_t *field) { PyObject *result = Py_None; uint8_t tag = qd_parse_tag(field); switch (tag) { case QD_AMQP_NULL: result = Py_None; break; case QD_AMQP_BOOLEAN: case QD_AMQP_TRUE: case QD_AMQP_FALSE: result = qd_parse_as_uint(field) ? Py_True : Py_False; break; case QD_AMQP_UBYTE: case QD_AMQP_USHORT: case QD_AMQP_UINT: case QD_AMQP_SMALLUINT: case QD_AMQP_UINT0: result = PyInt_FromLong((long) qd_parse_as_uint(field)); break; case QD_AMQP_ULONG: case QD_AMQP_SMALLULONG: case QD_AMQP_ULONG0: case QD_AMQP_TIMESTAMP: result = PyLong_FromUnsignedLongLong((unsigned PY_LONG_LONG) qd_parse_as_ulong(field)); break; case QD_AMQP_BYTE: case QD_AMQP_SHORT: case QD_AMQP_INT: case QD_AMQP_SMALLINT: result = PyInt_FromLong((long) qd_parse_as_int(field)); break; case QD_AMQP_LONG: case QD_AMQP_SMALLLONG: result = PyLong_FromUnsignedLongLong((unsigned PY_LONG_LONG) qd_parse_as_long(field)); break; case QD_AMQP_FLOAT: case QD_AMQP_DOUBLE: case QD_AMQP_DECIMAL32: case QD_AMQP_DECIMAL64: case QD_AMQP_DECIMAL128: case QD_AMQP_UTF32: case QD_AMQP_UUID: break; case QD_AMQP_VBIN8: case QD_AMQP_VBIN32: case QD_AMQP_STR8_UTF8: case QD_AMQP_STR32_UTF8: case QD_AMQP_SYM8: case QD_AMQP_SYM32: result = parsed_to_py_string(field); break; case QD_AMQP_LIST0: case QD_AMQP_LIST8: case QD_AMQP_LIST32: { uint32_t count = qd_parse_sub_count(field); result = PyList_New(count); for (uint32_t idx = 0; idx < count; idx++) { qd_parsed_field_t *sub = qd_parse_sub_value(field, idx); PyObject *pysub = qd_field_to_py(sub); if (pysub == 0) return 0; PyList_SetItem(result, idx, pysub); } break; } case QD_AMQP_MAP8: case QD_AMQP_MAP32: { uint32_t count = qd_parse_sub_count(field); result = PyDict_New(); for (uint32_t idx = 0; idx < count; idx++) { qd_parsed_field_t *key = qd_parse_sub_key(field, idx); qd_parsed_field_t *val = qd_parse_sub_value(field, idx); PyObject *pykey = parsed_to_py_string(key); PyObject *pyval = qd_field_to_py(val); if (pyval == 0) return 0; PyDict_SetItem(result, pykey, pyval); Py_DECREF(pykey); Py_DECREF(pyval); } break; } case QD_AMQP_ARRAY8: case QD_AMQP_ARRAY32: break; } return result; } //=============================================================================== // Logging Object //=============================================================================== typedef struct { PyObject_HEAD PyObject *module_name; qd_log_source_t *log_source; } LogAdapter; static int LogAdapter_init(LogAdapter *self, PyObject *args, PyObject *kwds) { const char *text; if (!PyArg_ParseTuple(args, "s", &text)) return -1; self->module_name = PyString_FromString(text); self->log_source = qd_log_source(text); return 0; } static void LogAdapter_dealloc(LogAdapter* self) { Py_XDECREF(self->module_name); self->ob_type->tp_free((PyObject*)self); } static PyObject* qd_python_log(PyObject *self, PyObject *args) { int level; const char* text; if (!PyArg_ParseTuple(args, "is", &level, &text)) return 0; LogAdapter *self_ptr = (LogAdapter*) self; //char *logmod = PyString_AS_STRING(self_ptr->module_name); qd_log(self_ptr->log_source, level, text); Py_INCREF(Py_None); return Py_None; } static PyMethodDef LogAdapter_methods[] = { {"log", qd_python_log, METH_VARARGS, "Emit a Log Line"}, {0, 0, 0, 0} }; static PyMethodDef empty_methods[] = { {0, 0, 0, 0} }; static PyTypeObject LogAdapterType = { PyObject_HEAD_INIT(0) 0, /* ob_size*/ "dispatch.LogAdapter", /* tp_name*/ sizeof(LogAdapter), /* tp_basicsize*/ 0, /* tp_itemsize*/ (destructor)LogAdapter_dealloc, /* tp_dealloc*/ 0, /* tp_print*/ 0, /* tp_getattr*/ 0, /* tp_setattr*/ 0, /* tp_compare*/ 0, /* tp_repr*/ 0, /* tp_as_number*/ 0, /* tp_as_sequence*/ 0, /* tp_as_mapping*/ 0, /* tp_hash */ 0, /* tp_call*/ 0, /* tp_str*/ 0, /* tp_getattro*/ 0, /* tp_setattro*/ 0, /* tp_as_buffer*/ Py_TPFLAGS_DEFAULT, /* tp_flags*/ "Dispatch Log Adapter", /* tp_doc */ 0, /* tp_traverse */ 0, /* tp_clear */ 0, /* tp_richcompare */ 0, /* tp_weaklistoffset */ 0, /* tp_iter */ 0, /* tp_iternext */ LogAdapter_methods, /* tp_methods */ 0, /* tp_members */ 0, /* tp_getset */ 0, /* tp_base */ 0, /* tp_dict */ 0, /* tp_descr_get */ 0, /* tp_descr_set */ 0, /* tp_dictoffset */ (initproc)LogAdapter_init, /* tp_init */ 0, /* tp_alloc */ 0, /* tp_new */ 0, /* tp_free */ 0, /* tp_is_gc */ 0, /* tp_bases */ 0, /* tp_mro */ 0, /* tp_cache */ 0, /* tp_subclasses */ 0, /* tp_weaklist */ 0, /* tp_del */ 0 /* tp_version_tag */ }; //=============================================================================== // Message IO Object //=============================================================================== typedef struct { PyObject_HEAD PyObject *handler; PyObject *handler_rx_call; qd_dispatch_t *qd; Py_ssize_t addr_count; qd_address_t **addrs; } IoAdapter; static void qd_io_rx_handler(void *context, qd_message_t *msg, int link_id) { IoAdapter *self = (IoAdapter*) context; // // Parse the message through the body and exit if the message is not well formed. // if (!qd_message_check(msg, QD_DEPTH_BODY)) return; // // Get an iterator for the application-properties. Exit if the message has none. // qd_field_iterator_t *ap = qd_message_field_iterator(msg, QD_FIELD_APPLICATION_PROPERTIES); if (ap == 0) return; // // Try to get a map-view of the application-properties. // qd_parsed_field_t *ap_map = qd_parse(ap); if (ap_map == 0 || !qd_parse_ok(ap_map) || !qd_parse_is_map(ap_map)) { qd_field_iterator_free(ap); qd_parse_free(ap_map); return; } // // Get an iterator for the body. Exit if the message has none. // qd_field_iterator_t *body = qd_message_field_iterator(msg, QD_FIELD_BODY); if (body == 0) { qd_field_iterator_free(ap); qd_parse_free(ap_map); return; } // // Try to get a map-view of the body. // qd_parsed_field_t *body_map = qd_parse(body); if (body_map == 0 || !qd_parse_ok(body_map) || !qd_parse_is_map(body_map)) { qd_field_iterator_free(ap); qd_field_iterator_free(body); qd_parse_free(ap_map); qd_parse_free(body_map); return; } sys_mutex_lock(ilock); PyObject *pAP = qd_field_to_py(ap_map); PyObject *pBody = qd_field_to_py(body_map); PyObject *pArgs = PyTuple_New(3); PyTuple_SetItem(pArgs, 0, pAP); PyTuple_SetItem(pArgs, 1, pBody); PyTuple_SetItem(pArgs, 2, PyInt_FromLong((long) link_id)); PyObject *pValue = PyObject_CallObject(self->handler_rx_call, pArgs); Py_DECREF(pArgs); if (pValue) { Py_DECREF(pValue); } sys_mutex_unlock(ilock); qd_field_iterator_free(ap); qd_field_iterator_free(body); qd_parse_free(ap_map); qd_parse_free(body_map); } static int IoAdapter_init(IoAdapter *self, PyObject *args, PyObject *kwds) { PyObject *addrs; if (!PyArg_ParseTuple(args, "OO", &self->handler, &addrs)) return -1; self->handler_rx_call = PyObject_GetAttrString(self->handler, "receive"); if (!self->handler_rx_call || !PyCallable_Check(self->handler_rx_call)) return -1; if (!PyTuple_Check(addrs)) return -1; Py_INCREF(self->handler); Py_INCREF(self->handler_rx_call); self->qd = dispatch; self->addr_count = PyTuple_Size(addrs); self->addrs = NEW_PTR_ARRAY(qd_address_t, self->addr_count); for (Py_ssize_t idx = 0; idx < self->addr_count; idx++) self->addrs[idx] = qd_router_register_address(self->qd, PyString_AS_STRING(PyTuple_GetItem(addrs, idx)), qd_io_rx_handler, py_semantics, self); return 0; } static void IoAdapter_dealloc(IoAdapter* self) { for (Py_ssize_t idx = 0; idx < self->addr_count; idx++) qd_router_unregister_address(self->addrs[idx]); free(self->addrs); Py_DECREF(self->handler); Py_DECREF(self->handler_rx_call); self->ob_type->tp_free((PyObject*)self); } static PyObject* qd_python_send(PyObject *self, PyObject *args) { IoAdapter *ioa = (IoAdapter*) self; qd_composed_field_t *field = 0; const char *address; PyObject *app_properties; PyObject *body; if (!PyArg_ParseTuple(args, "sOO", &address, &app_properties, &body)) return 0; field = qd_compose(QD_PERFORMATIVE_DELIVERY_ANNOTATIONS, field); qd_compose_start_map(field); qd_compose_insert_string(field, QD_DA_INGRESS); qd_compose_insert_string(field, qd_router_id(ioa->qd)); qd_compose_insert_string(field, QD_DA_TRACE); qd_compose_start_list(field); qd_compose_insert_string(field, qd_router_id(ioa->qd)); qd_compose_end_list(field); qd_compose_end_map(field); field = qd_compose(QD_PERFORMATIVE_PROPERTIES, field); qd_compose_start_list(field); qd_compose_insert_null(field); // message-id qd_compose_insert_null(field); // user-id qd_compose_insert_string(field, address); // to qd_compose_end_list(field); field = qd_compose(QD_PERFORMATIVE_APPLICATION_PROPERTIES, field); qd_py_to_composed(app_properties, field); field = qd_compose(QD_PERFORMATIVE_BODY_AMQP_VALUE, field); qd_py_to_composed(body, field); qd_message_t *msg = qd_message(); qd_message_compose_2(msg, field); qd_router_send2(ioa->qd, address, msg); qd_message_free(msg); qd_compose_free(field); Py_INCREF(Py_None); return Py_None; } static PyMethodDef IoAdapter_methods[] = { {"send", qd_python_send, METH_VARARGS, "Send a Message"}, {0, 0, 0, 0} }; static PyTypeObject IoAdapterType = { PyObject_HEAD_INIT(0) 0, /* ob_size*/ "dispatch.IoAdapter", /* tp_name*/ sizeof(IoAdapter), /* tp_basicsize*/ 0, /* tp_itemsize*/ (destructor)IoAdapter_dealloc, /* tp_dealloc*/ 0, /* tp_print*/ 0, /* tp_getattr*/ 0, /* tp_setattr*/ 0, /* tp_compare*/ 0, /* tp_repr*/ 0, /* tp_as_number*/ 0, /* tp_as_sequence*/ 0, /* tp_as_mapping*/ 0, /* tp_hash */ 0, /* tp_call*/ 0, /* tp_str*/ 0, /* tp_getattro*/ 0, /* tp_setattro*/ 0, /* tp_as_buffer*/ Py_TPFLAGS_DEFAULT, /* tp_flags*/ "Dispatch IO Adapter", /* tp_doc */ 0, /* tp_traverse */ 0, /* tp_clear */ 0, /* tp_richcompare */ 0, /* tp_weaklistoffset */ 0, /* tp_iter */ 0, /* tp_iternext */ IoAdapter_methods, /* tp_methods */ 0, /* tp_members */ 0, /* tp_getset */ 0, /* tp_base */ 0, /* tp_dict */ 0, /* tp_descr_get */ 0, /* tp_descr_set */ 0, /* tp_dictoffset */ (initproc)IoAdapter_init, /* tp_init */ 0, /* tp_alloc */ 0, /* tp_new */ 0, /* tp_free */ 0, /* tp_is_gc */ 0, /* tp_bases */ 0, /* tp_mro */ 0, /* tp_cache */ 0, /* tp_subclasses */ 0, /* tp_weaklist */ 0, /* tp_del */ 0 /* tp_version_tag */ }; //=============================================================================== // Initialization of Modules and Types //=============================================================================== static void qd_register_log_constant(PyObject *module, const char *name, uint32_t value) { PyObject *const_object = PyInt_FromLong((long) value); Py_INCREF(const_object); PyModule_AddObject(module, name, const_object); } static void qd_python_setup(void) { LogAdapterType.tp_new = PyType_GenericNew; IoAdapterType.tp_new = PyType_GenericNew; if ((PyType_Ready(&LogAdapterType) < 0) || (PyType_Ready(&IoAdapterType) < 0)) { PyErr_Print(); qd_log(log_source, QD_LOG_ERROR, "Unable to initialize Adapters"); assert(0); } else { PyObject *m = Py_InitModule3("dispatch", empty_methods, "Dispatch Adapter Module"); // // Append sys.path to include location of Dispatch libraries // if (dispatch_python_pkgdir) { PyObject *sys_path = PySys_GetObject("path"); PyList_Append(sys_path, dispatch_python_pkgdir); } // // Add LogAdapter // PyTypeObject *laType = &LogAdapterType; Py_INCREF(laType); PyModule_AddObject(m, "LogAdapter", (PyObject*) &LogAdapterType); qd_register_log_constant(m, "LOG_TRACE", QD_LOG_TRACE); qd_register_log_constant(m, "LOG_DEBUG", QD_LOG_DEBUG); qd_register_log_constant(m, "LOG_INFO", QD_LOG_INFO); qd_register_log_constant(m, "LOG_NOTICE", QD_LOG_NOTICE); qd_register_log_constant(m, "LOG_WARNING", QD_LOG_WARNING); qd_register_log_constant(m, "LOG_ERROR", QD_LOG_ERROR); qd_register_log_constant(m, "LOG_CRITICAL", QD_LOG_CRITICAL); // PyTypeObject *ioaType = &IoAdapterType; Py_INCREF(ioaType); PyModule_AddObject(m, "IoAdapter", (PyObject*) &IoAdapterType); Py_INCREF(m); dispatch_module = m; } } void qd_python_lock(void) { sys_mutex_lock(ilock); } void qd_python_unlock(void) { sys_mutex_unlock(ilock); }