class _PrivateConnection(object._PrivateObject, _Connection):
def create_notification_iterator(self, notification_types=None):
- if notification_types is None:
- notif_types = None
- else:
- for notif_cls in notification_types:
- if notif_cls not in bt2.notification._NOTIF_TYPE_TO_CLS.values():
- raise ValueError("'{}' is not a notification class".format(notif_cls))
-
- notif_types = [notif_cls._TYPE for notif_cls in notification_types]
-
- status, notif_iter_ptr = native_bt.py3_create_notif_iter(int(self._ptr),
- notif_types)
+ notif_types = bt2.notification._notif_types_from_notif_classes(notification_types)
+ status, notif_iter_ptr = native_bt.py3_create_priv_conn_notif_iter(int(self._ptr),
+ notif_types)
_handle_status(status, 'cannot create notification iterator object')
assert(notif_iter_ptr)
return bt2.notification_iterator._PrivateConnectionNotificationIterator._create_from_ptr(notif_iter_ptr)
$result = $1;
}
+%{
+static enum bt_notification_type *bt_py3_notif_types_from_py_list(
+ PyObject *py_notif_types)
+{
+ enum bt_notification_type *notification_types = NULL;
+ size_t i;
+
+ assert(!PyErr_Occurred());
+
+ if (py_notif_types == Py_None) {
+ goto end;
+ }
+
+ assert(PyList_Check(py_notif_types));
+ notification_types = g_new0(enum bt_notification_type,
+ PyList_Size(py_notif_types) + 1);
+ assert(notification_types);
+ notification_types[PyList_Size(py_notif_types)] =
+ BT_NOTIFICATION_TYPE_SENTINEL;
+
+ for (i = 0; i < PyList_Size(py_notif_types); i++) {
+ PyObject *item = PyList_GetItem(py_notif_types, i);
+ long value;
+ int overflow;
+
+ assert(item);
+ assert(PyLong_Check(item));
+ value = PyLong_AsLongAndOverflow(item, &overflow);
+ assert(overflow == 0);
+ notification_types[i] = value;
+ }
+
+end:
+ return notification_types;
+}
+%}
+
/* Per-module interface files */
%include "native_btccpriomap.i"
%include "native_btclockclass.i"
struct bt_notification_iterator *notif_iter;
};
-static struct bt_py3_create_notif_iter_ret bt_py3_create_notif_iter(
+static struct bt_py3_create_notif_iter_ret bt_py3_create_priv_conn_notif_iter(
unsigned long long priv_conn_addr, PyObject *py_notif_types)
{
struct bt_private_connection *priv_conn;
assert(!PyErr_Occurred());
assert(priv_conn);
- if (py_notif_types != Py_None) {
- size_t i;
-
- assert(PyList_Check(py_notif_types));
- notification_types = g_new0(enum bt_notification_type,
- PyList_Size(py_notif_types) + 1);
- assert(notification_types);
- notification_types[PyList_Size(py_notif_types)] =
- BT_NOTIFICATION_TYPE_SENTINEL;
-
- for (i = 0; i < PyList_Size(py_notif_types); i++) {
- PyObject *item = PyList_GetItem(py_notif_types, i);
- long value;
- int overflow;
-
- assert(item);
- assert(PyLong_Check(item));
- value = PyLong_AsLongAndOverflow(item, &overflow);
- assert(overflow == 0);
- notification_types[i] = value;
- }
- }
-
+ notification_types = bt_py3_notif_types_from_py_list(py_notif_types);
ret.status = bt_private_connection_create_notification_iterator(
priv_conn, notification_types, &ret.notif_iter);
-
- if (notification_types) {
- g_free(notification_types);
- }
+ g_free(notification_types);
return ret;
}
%}
-struct bt_py3_create_notif_iter_ret bt_py3_create_notif_iter(
+struct bt_py3_create_notif_iter_ret bt_py3_create_priv_conn_notif_iter(
unsigned long long priv_conn_addr, PyObject *notif_types);
struct bt_component *bt_private_connection_notification_iterator_get_component(
struct bt_notification_iterator *iterator);
+/* Functions (output port) */
+struct bt_notification_iterator *bt_output_port_notification_iterator_create(
+ struct bt_port *port, const char *colander_component_name,
+ const enum bt_notification_type *notification_types);
+
/* Helper functions for Python */
%{
static PyObject *bt_py3_get_user_component_from_user_notif_iter(
struct bt_private_port *private_port);
void *bt_private_port_get_user_data(
struct bt_private_port *private_port);
+
+%{
+static struct bt_notification_iterator *bt_py3_create_output_port_notif_iter(
+ unsigned long long port_addr, const char *colander_name,
+ PyObject *py_notif_types)
+{
+ struct bt_notification_iterator *notif_iter;
+ struct bt_port *output_port;
+ enum bt_notification_type *notification_types;
+
+ output_port = (void *) port_addr;
+ assert(!PyErr_Occurred());
+ assert(output_port);
+
+ notification_types = bt_py3_notif_types_from_py_list(py_notif_types);
+ notif_iter = bt_output_port_notification_iterator_create(output_port,
+ colander_name, notification_types);
+ g_free(notification_types);
+ return notif_iter;
+}
+%}
+
+struct bt_notification_iterator *bt_py3_create_output_port_notif_iter(
+ unsigned long long port_addr, const char *colander_name,
+ PyObject *py_notif_types);
return _NOTIF_TYPE_TO_CLS[notif_type]._create_from_ptr(ptr)
+def _notif_types_from_notif_classes(notification_types):
+ if notification_types is None:
+ notif_types = None
+ else:
+ for notif_cls in notification_types:
+ if notif_cls not in _NOTIF_TYPE_TO_CLS.values():
+ raise ValueError("'{}' is not a notification class".format(notif_cls))
+
+ notif_types = [notif_cls._TYPE for notif_cls in notification_types]
+
+ return notif_types
+
+
class _Notification(object._Object):
pass
return bt2.component._create_generic_component_from_ptr(comp_ptr)
+class _OutputPortNotificationIterator(_GenericNotificationIterator):
+ pass
+
+
class _UserNotificationIterator(_NotificationIterator):
def __new__(cls, ptr):
# User iterator objects are always created by the native side,
import collections.abc
import bt2.component
import bt2.connection
+import bt2.notification_iterator
+import bt2.notification
import copy
import bt2
class _OutputPort(_Port):
- pass
+ def create_notification_iterator(self, notification_types=None,
+ colander_component_name=None):
+ notif_types = bt2.notification._notif_types_from_notif_classes(notification_types)
+
+ if colander_component_name is not None:
+ utils._check_str(colander_component_name)
+
+ notif_iter_ptr = native_bt.py3_create_output_port_notif_iter(int(self._ptr),
+ colander_component_name,
+ notif_types)
+
+ if notif_iter_ptr is None:
+ raise bt2.CreationError('cannot create output port notification iterator')
+
+ return bt2.notification_iterator._OutputPortNotificationIterator._create_from_ptr(notif_iter_ptr)
class _PrivatePort(object._PrivateObject, _Port):