Subscription.c 43 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156
  1. //-----------------------------------------------------------------------------
  2. // Subscription.c
  3. // Defines the routines for handling Oracle subscription information.
  4. //-----------------------------------------------------------------------------
  5. //-----------------------------------------------------------------------------
  6. // structures used for handling subscriptions
  7. //-----------------------------------------------------------------------------
  8. typedef struct {
  9. PyObject_HEAD
  10. OCISubscription *handle;
  11. udt_Connection *connection;
  12. PyObject *callback;
  13. ub4 namespace;
  14. ub4 protocol;
  15. ub4 port;
  16. ub4 timeout;
  17. ub4 operations;
  18. ub4 qos;
  19. ub4 cqqos;
  20. ub4 rowids;
  21. ub4 id;
  22. } udt_Subscription;
  23. typedef struct {
  24. PyObject_HEAD
  25. udt_Subscription *subscription;
  26. ub4 type;
  27. PyObject *dbname;
  28. PyObject *tables;
  29. PyObject *queries;
  30. } udt_Message;
  31. typedef struct {
  32. PyObject_HEAD
  33. PyObject *name;
  34. PyObject *rows;
  35. ub4 operation;
  36. } udt_MessageTable;
  37. typedef struct {
  38. PyObject_HEAD
  39. PyObject *rowid;
  40. ub4 operation;
  41. } udt_MessageRow;
  42. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  43. typedef struct {
  44. PyObject_HEAD
  45. ub8 id;
  46. ub4 operation;
  47. PyObject *tables;
  48. } udt_MessageQuery;
  49. #endif
  50. //-----------------------------------------------------------------------------
  51. // Declaration of subscription functions
  52. //-----------------------------------------------------------------------------
  53. static void Subscription_Free(udt_Subscription*);
  54. static PyObject *Subscription_Repr(udt_Subscription*);
  55. static PyObject *Subscription_RegisterQuery(udt_Subscription*, PyObject*);
  56. static void Message_Free(udt_Message*);
  57. static void MessageTable_Free(udt_MessageTable*);
  58. static void MessageRow_Free(udt_MessageRow*);
  59. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  60. static void MessageQuery_Free(udt_MessageQuery*);
  61. #endif
  62. //-----------------------------------------------------------------------------
  63. // declaration of members for Python types
  64. //-----------------------------------------------------------------------------
  65. static PyMemberDef g_SubscriptionTypeMembers[] = {
  66. { "callback", T_OBJECT, offsetof(udt_Subscription, callback), READONLY },
  67. { "connection", T_OBJECT, offsetof(udt_Subscription, connection),
  68. READONLY },
  69. { "namespace", T_INT, offsetof(udt_Subscription, namespace), READONLY },
  70. { "protocol", T_INT, offsetof(udt_Subscription, protocol), READONLY },
  71. { "port", T_INT, offsetof(udt_Subscription, port), READONLY },
  72. { "timeout", T_INT, offsetof(udt_Subscription, timeout), READONLY },
  73. { "operations", T_INT, offsetof(udt_Subscription, operations), READONLY },
  74. { "qos", T_INT, offsetof(udt_Subscription, qos), READONLY },
  75. { "cqqos", T_INT, offsetof(udt_Subscription, cqqos), READONLY },
  76. { "rowids", T_BOOL, offsetof(udt_Subscription, rowids), READONLY },
  77. { "id", T_INT, offsetof(udt_Subscription, id), READONLY },
  78. { NULL }
  79. };
  80. static PyMemberDef g_MessageTypeMembers[] = {
  81. { "subscription", T_OBJECT, offsetof(udt_Message, subscription),
  82. READONLY },
  83. { "type", T_INT, offsetof(udt_Message, type), READONLY },
  84. { "dbname", T_OBJECT, offsetof(udt_Message, dbname), READONLY },
  85. { "tables", T_OBJECT, offsetof(udt_Message, tables), READONLY },
  86. { "queries", T_OBJECT, offsetof(udt_Message, queries), READONLY },
  87. { NULL }
  88. };
  89. static PyMemberDef g_MessageTableTypeMembers[] = {
  90. { "name", T_OBJECT, offsetof(udt_MessageTable, name), READONLY },
  91. { "rows", T_OBJECT, offsetof(udt_MessageTable, rows), READONLY },
  92. { "operation", T_INT, offsetof(udt_MessageTable, operation), READONLY },
  93. { NULL }
  94. };
  95. static PyMemberDef g_MessageRowTypeMembers[] = {
  96. { "rowid", T_OBJECT, offsetof(udt_MessageRow, rowid), READONLY },
  97. { "operation", T_INT, offsetof(udt_MessageRow, operation), READONLY },
  98. { NULL }
  99. };
  100. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  101. static PyMemberDef g_MessageQueryTypeMembers[] = {
  102. { "id", T_INT, offsetof(udt_MessageQuery, id), READONLY },
  103. { "operation", T_INT, offsetof(udt_MessageQuery, operation), READONLY },
  104. { "tables", T_OBJECT, offsetof(udt_MessageQuery, tables), READONLY },
  105. { NULL }
  106. };
  107. #endif
  108. //-----------------------------------------------------------------------------
  109. // declaration of methods for Python types
  110. //-----------------------------------------------------------------------------
  111. static PyMethodDef g_SubscriptionTypeMethods[] = {
  112. { "registerquery", (PyCFunction) Subscription_RegisterQuery,
  113. METH_VARARGS },
  114. { NULL, NULL }
  115. };
  116. //-----------------------------------------------------------------------------
  117. // Python type declarations
  118. //-----------------------------------------------------------------------------
  119. static PyTypeObject g_SubscriptionType = {
  120. PyVarObject_HEAD_INIT(NULL, 0)
  121. "cx_Oracle.Subscription", // tp_name
  122. sizeof(udt_Subscription), // tp_basicsize
  123. 0, // tp_itemsize
  124. (destructor) Subscription_Free, // tp_dealloc
  125. 0, // tp_print
  126. 0, // tp_getattr
  127. 0, // tp_setattr
  128. 0, // tp_compare
  129. (reprfunc) Subscription_Repr, // tp_repr
  130. 0, // tp_as_number
  131. 0, // tp_as_sequence
  132. 0, // tp_as_mapping
  133. 0, // tp_hash
  134. 0, // tp_call
  135. 0, // tp_str
  136. 0, // tp_getattro
  137. 0, // tp_setattro
  138. 0, // tp_as_buffer
  139. Py_TPFLAGS_DEFAULT, // tp_flags
  140. 0, // tp_doc
  141. 0, // tp_traverse
  142. 0, // tp_clear
  143. 0, // tp_richcompare
  144. 0, // tp_weaklistoffset
  145. 0, // tp_iter
  146. 0, // tp_iternext
  147. g_SubscriptionTypeMethods, // tp_methods
  148. g_SubscriptionTypeMembers, // tp_members
  149. 0, // tp_getset
  150. 0, // tp_base
  151. 0, // tp_dict
  152. 0, // tp_descr_get
  153. 0, // tp_descr_set
  154. 0, // tp_dictoffset
  155. 0, // tp_init
  156. 0, // tp_alloc
  157. 0, // tp_new
  158. 0, // tp_free
  159. 0, // tp_is_gc
  160. 0 // tp_bases
  161. };
  162. static PyTypeObject g_MessageType = {
  163. PyVarObject_HEAD_INIT(NULL, 0)
  164. "cx_Oracle.Message", // tp_name
  165. sizeof(udt_Message), // tp_basicsize
  166. 0, // tp_itemsize
  167. (destructor) Message_Free, // tp_dealloc
  168. 0, // tp_print
  169. 0, // tp_getattr
  170. 0, // tp_setattr
  171. 0, // tp_compare
  172. 0, // tp_repr
  173. 0, // tp_as_number
  174. 0, // tp_as_sequence
  175. 0, // tp_as_mapping
  176. 0, // tp_hash
  177. 0, // tp_call
  178. 0, // tp_str
  179. 0, // tp_getattro
  180. 0, // tp_setattro
  181. 0, // tp_as_buffer
  182. Py_TPFLAGS_DEFAULT, // tp_flags
  183. 0, // tp_doc
  184. 0, // tp_traverse
  185. 0, // tp_clear
  186. 0, // tp_richcompare
  187. 0, // tp_weaklistoffset
  188. 0, // tp_iter
  189. 0, // tp_iternext
  190. 0, // tp_methods
  191. g_MessageTypeMembers, // tp_members
  192. 0, // tp_getset
  193. 0, // tp_base
  194. 0, // tp_dict
  195. 0, // tp_descr_get
  196. 0, // tp_descr_set
  197. 0, // tp_dictoffset
  198. 0, // tp_init
  199. 0, // tp_alloc
  200. 0, // tp_new
  201. 0, // tp_free
  202. 0, // tp_is_gc
  203. 0 // tp_bases
  204. };
  205. static PyTypeObject g_MessageTableType = {
  206. PyVarObject_HEAD_INIT(NULL, 0)
  207. "cx_Oracle.MessageTable", // tp_name
  208. sizeof(udt_MessageTable), // tp_basicsize
  209. 0, // tp_itemsize
  210. (destructor) MessageTable_Free, // tp_dealloc
  211. 0, // tp_print
  212. 0, // tp_getattr
  213. 0, // tp_setattr
  214. 0, // tp_compare
  215. 0, // tp_repr
  216. 0, // tp_as_number
  217. 0, // tp_as_sequence
  218. 0, // tp_as_mapping
  219. 0, // tp_hash
  220. 0, // tp_call
  221. 0, // tp_str
  222. 0, // tp_getattro
  223. 0, // tp_setattro
  224. 0, // tp_as_buffer
  225. Py_TPFLAGS_DEFAULT, // tp_flags
  226. 0, // tp_doc
  227. 0, // tp_traverse
  228. 0, // tp_clear
  229. 0, // tp_richcompare
  230. 0, // tp_weaklistoffset
  231. 0, // tp_iter
  232. 0, // tp_iternext
  233. 0, // tp_methods
  234. g_MessageTableTypeMembers, // tp_members
  235. 0, // tp_getset
  236. 0, // tp_base
  237. 0, // tp_dict
  238. 0, // tp_descr_get
  239. 0, // tp_descr_set
  240. 0, // tp_dictoffset
  241. 0, // tp_init
  242. 0, // tp_alloc
  243. 0, // tp_new
  244. 0, // tp_free
  245. 0, // tp_is_gc
  246. 0 // tp_bases
  247. };
  248. static PyTypeObject g_MessageRowType = {
  249. PyVarObject_HEAD_INIT(NULL, 0)
  250. "cx_Oracle.MessageRow", // tp_name
  251. sizeof(udt_MessageRow), // tp_basicsize
  252. 0, // tp_itemsize
  253. (destructor) MessageRow_Free, // tp_dealloc
  254. 0, // tp_print
  255. 0, // tp_getattr
  256. 0, // tp_setattr
  257. 0, // tp_compare
  258. 0, // tp_repr
  259. 0, // tp_as_number
  260. 0, // tp_as_sequence
  261. 0, // tp_as_mapping
  262. 0, // tp_hash
  263. 0, // tp_call
  264. 0, // tp_str
  265. 0, // tp_getattro
  266. 0, // tp_setattro
  267. 0, // tp_as_buffer
  268. Py_TPFLAGS_DEFAULT, // tp_flags
  269. 0, // tp_doc
  270. 0, // tp_traverse
  271. 0, // tp_clear
  272. 0, // tp_richcompare
  273. 0, // tp_weaklistoffset
  274. 0, // tp_iter
  275. 0, // tp_iternext
  276. 0, // tp_methods
  277. g_MessageRowTypeMembers, // tp_members
  278. 0, // tp_getset
  279. 0, // tp_base
  280. 0, // tp_dict
  281. 0, // tp_descr_get
  282. 0, // tp_descr_set
  283. 0, // tp_dictoffset
  284. 0, // tp_init
  285. 0, // tp_alloc
  286. 0, // tp_new
  287. 0, // tp_free
  288. 0, // tp_is_gc
  289. 0 // tp_bases
  290. };
  291. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  292. static PyTypeObject g_MessageQueryType = {
  293. PyVarObject_HEAD_INIT(NULL, 0)
  294. "cx_Oracle.MessageQuery", // tp_name
  295. sizeof(udt_MessageQuery), // tp_basicsize
  296. 0, // tp_itemsize
  297. (destructor) MessageQuery_Free, // tp_dealloc
  298. 0, // tp_print
  299. 0, // tp_getattr
  300. 0, // tp_setattr
  301. 0, // tp_compare
  302. 0, // tp_repr
  303. 0, // tp_as_number
  304. 0, // tp_as_sequence
  305. 0, // tp_as_mapping
  306. 0, // tp_hash
  307. 0, // tp_call
  308. 0, // tp_str
  309. 0, // tp_getattro
  310. 0, // tp_setattro
  311. 0, // tp_as_buffer
  312. Py_TPFLAGS_DEFAULT, // tp_flags
  313. 0, // tp_doc
  314. 0, // tp_traverse
  315. 0, // tp_clear
  316. 0, // tp_richcompare
  317. 0, // tp_weaklistoffset
  318. 0, // tp_iter
  319. 0, // tp_iternext
  320. 0, // tp_methods
  321. g_MessageQueryTypeMembers, // tp_members
  322. 0, // tp_getset
  323. 0, // tp_base
  324. 0, // tp_dict
  325. 0, // tp_descr_get
  326. 0, // tp_descr_set
  327. 0, // tp_dictoffset
  328. 0, // tp_init
  329. 0, // tp_alloc
  330. 0, // tp_new
  331. 0, // tp_free
  332. 0, // tp_is_gc
  333. 0 // tp_bases
  334. };
  335. #endif
  336. //-----------------------------------------------------------------------------
  337. // MessageRow_Initialize()
  338. // Initialize a new message row with the information from the descriptor.
  339. //-----------------------------------------------------------------------------
  340. static int MessageRow_Initialize(
  341. udt_MessageRow *self, // object to initialize
  342. udt_Environment *env, // environment to use
  343. dvoid *descriptor) // descriptor to get information from
  344. {
  345. ub4 rowidLength;
  346. sword status;
  347. char *rowid;
  348. // determine operation
  349. status = OCIAttrGet(descriptor, OCI_DTYPE_ROW_CHDES, &self->operation,
  350. NULL, OCI_ATTR_CHDES_ROW_OPFLAGS, env->errorHandle);
  351. if (Environment_CheckForError(env, status,
  352. "MessageRow_Initialize(): get operation") < 0)
  353. return -1;
  354. // determine table name
  355. status = OCIAttrGet(descriptor, OCI_DTYPE_ROW_CHDES, &rowid, &rowidLength,
  356. OCI_ATTR_CHDES_ROW_ROWID, env->errorHandle);
  357. if (Environment_CheckForError(env, status,
  358. "MessageRow_Initialize(): get rowid") < 0)
  359. return -1;
  360. self->rowid = cxString_FromEncodedString(rowid, rowidLength,
  361. env->encoding);
  362. if (!self->rowid)
  363. return -1;
  364. return 0;
  365. }
  366. //-----------------------------------------------------------------------------
  367. // MessageTable_Initialize()
  368. // Initialize a new message table with the information from the descriptor.
  369. //-----------------------------------------------------------------------------
  370. static int MessageTable_Initialize(
  371. udt_MessageTable *self, // object to initialize
  372. udt_Environment *env, // environment to use
  373. dvoid *descriptor) // descriptor to get information from
  374. {
  375. dvoid **rowDescriptor, *indicator;
  376. ub4 nameLength, i;
  377. udt_MessageRow *row;
  378. boolean exists;
  379. OCIColl *rows;
  380. sword status;
  381. sb4 numRows;
  382. char *name;
  383. // determine operation
  384. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &self->operation,
  385. NULL, OCI_ATTR_CHDES_TABLE_OPFLAGS, env->errorHandle);
  386. if (Environment_CheckForError(env, status,
  387. "MessageTable_Initialize(): get operation") < 0)
  388. return -1;
  389. // determine table name
  390. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &name, &nameLength,
  391. OCI_ATTR_CHDES_TABLE_NAME, env->errorHandle);
  392. if (Environment_CheckForError(env, status,
  393. "MessageTable_Initialize(): get table name") < 0)
  394. return -1;
  395. self->name = cxString_FromEncodedString(name, nameLength, env->encoding);
  396. if (!self->name)
  397. return -1;
  398. // if change invalidated all rows, nothing to do
  399. if (self->operation & OCI_OPCODE_ALLROWS)
  400. return 0;
  401. // determine rows collection
  402. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &rows, NULL,
  403. OCI_ATTR_CHDES_TABLE_ROW_CHANGES, env->errorHandle);
  404. if (Environment_CheckForError(env, status,
  405. "MessageTable_Initialize(): get rows collection") < 0)
  406. return -1;
  407. // determine number of rows in collection
  408. status = OCICollSize(env->handle, env->errorHandle, rows, &numRows);
  409. if (Environment_CheckForError(env, status,
  410. "MessageTable_Initialize(): get size of rows collection") < 0)
  411. return -1;
  412. // populate the rows attribute
  413. self->rows = PyList_New(numRows);
  414. if (!self->rows)
  415. return -1;
  416. for (i = 0; i < numRows; i++) {
  417. status = OCICollGetElem(env->handle, env->errorHandle, rows, i,
  418. &exists, (dvoid*) &rowDescriptor, &indicator);
  419. if (Environment_CheckForError(env, status,
  420. "MessageTable_Initialize(): get element from collection") < 0)
  421. return -1;
  422. row = (udt_MessageRow*)
  423. g_MessageRowType.tp_alloc(&g_MessageRowType, 0);
  424. if (!row)
  425. return -1;
  426. PyList_SET_ITEM(self->rows, i, (PyObject*) row);
  427. if (MessageRow_Initialize(row, env, *rowDescriptor) < 0)
  428. return -1;
  429. }
  430. return 0;
  431. }
  432. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  433. //-----------------------------------------------------------------------------
  434. // MessageQuery_Initialize()
  435. // Initialize a new message query with the information from the descriptor.
  436. //-----------------------------------------------------------------------------
  437. static int MessageQuery_Initialize(
  438. udt_MessageQuery *self, // object to initialize
  439. udt_Environment *env, // environment to use
  440. dvoid *descriptor) // descriptor to get information from
  441. {
  442. dvoid **tableDescriptor, *indicator;
  443. udt_MessageTable *table;
  444. ub4 i;
  445. OCIColl *tables;
  446. boolean exists;
  447. sb4 numTables;
  448. sword status;
  449. // determine query id
  450. status = OCIAttrGet(descriptor, OCI_DTYPE_CQDES, &self->id,
  451. NULL, OCI_ATTR_CQDES_QUERYID, env->errorHandle);
  452. if (Environment_CheckForError(env, status,
  453. "MessageQuery_Initialize(): get query id") < 0)
  454. return -1;
  455. // determine operation
  456. status = OCIAttrGet(descriptor, OCI_DTYPE_CQDES, &self->operation,
  457. NULL, OCI_ATTR_CQDES_OPERATION, env->errorHandle);
  458. if (Environment_CheckForError(env, status,
  459. "MessageQuery_Initialize(): get operation") < 0)
  460. return -1;
  461. // determine table collection
  462. status = OCIAttrGet(descriptor, OCI_DTYPE_CQDES, &tables, NULL,
  463. OCI_ATTR_CQDES_TABLE_CHANGES, env->errorHandle);
  464. if (Environment_CheckForError(env, status,
  465. "MessageQuery_Initialize(): get tables collection") < 0)
  466. return -1;
  467. // determine number of tables
  468. if (!tables)
  469. numTables = 0;
  470. else {
  471. status = OCICollSize(env->handle, env->errorHandle, tables,
  472. &numTables);
  473. if (Environment_CheckForError(env, status,
  474. "MessageQuery_Initialize(): get size of collection") < 0)
  475. return -1;
  476. }
  477. // create list to hold results
  478. self->tables = PyList_New(numTables);
  479. if (!self->tables)
  480. return -1;
  481. // populate each entry with a message table instance
  482. for (i = 0; i < numTables; i++) {
  483. status = OCICollGetElem(env->handle, env->errorHandle, tables, i,
  484. &exists, (dvoid*) &tableDescriptor, &indicator);
  485. if (Environment_CheckForError(env, status,
  486. "MessageQuery_Initialize(): get element from collection") < 0)
  487. return -1;
  488. table = (udt_MessageTable*)
  489. g_MessageTableType.tp_alloc(&g_MessageTableType, 0);
  490. if (!table)
  491. return -1;
  492. PyList_SET_ITEM(self->tables, i, (PyObject*) table);
  493. if (MessageTable_Initialize(table, env, *tableDescriptor) < 0)
  494. return -1;
  495. }
  496. return 0;
  497. }
  498. #endif
  499. //-----------------------------------------------------------------------------
  500. // Message_Initialize()
  501. // Initialize a new message with the information from the descriptor.
  502. //-----------------------------------------------------------------------------
  503. static int Message_Initialize(
  504. udt_Message *self, // object to initialize
  505. udt_Environment *env, // environment to use
  506. udt_Subscription *subscription, // associated subscription for message
  507. dvoid *descriptor) // descriptor to get information from
  508. {
  509. dvoid **tableDescriptor, *indicator;
  510. udt_MessageTable *table;
  511. ub4 dbnameLength, i;
  512. OCIColl *tables;
  513. boolean exists;
  514. sb4 numTables;
  515. char *dbname;
  516. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  517. dvoid **queryDescriptor;
  518. udt_MessageQuery *query;
  519. OCIColl *queries;
  520. sb4 numQueries;
  521. #endif
  522. sword status;
  523. // assign reference to associated subscription
  524. Py_INCREF(subscription);
  525. self->subscription = subscription;
  526. // determine type
  527. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &self->type, NULL,
  528. OCI_ATTR_CHDES_NFYTYPE, env->errorHandle);
  529. if (Environment_CheckForError(env, status,
  530. "Message_Initialize(): get type") < 0)
  531. return -1;
  532. // determine database name
  533. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &dbname, &dbnameLength,
  534. OCI_ATTR_CHDES_DBNAME, env->errorHandle);
  535. if (Environment_CheckForError(env, status,
  536. "Message_Initialize(): get database name") < 0)
  537. return -1;
  538. self->dbname = cxString_FromEncodedString(dbname, dbnameLength,
  539. env->encoding);
  540. if (!self->dbname)
  541. return -1;
  542. if (self->type == OCI_EVENT_OBJCHANGE) {
  543. // determine table collection
  544. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &tables, NULL,
  545. OCI_ATTR_CHDES_TABLE_CHANGES, env->errorHandle);
  546. if (Environment_CheckForError(env, status,
  547. "Message_Initialize(): get tables collection") < 0)
  548. return -1;
  549. // determine number of tables
  550. if (!tables)
  551. numTables = 0;
  552. else {
  553. status = OCICollSize(env->handle, env->errorHandle, tables,
  554. &numTables);
  555. if (Environment_CheckForError(env, status,
  556. "Message_Initialize(): get size of collection") < 0)
  557. return -1;
  558. }
  559. // create list to hold results
  560. self->tables = PyList_New(numTables);
  561. if (!self->tables)
  562. return -1;
  563. // populate each entry with a message table instance
  564. for (i = 0; i < numTables; i++) {
  565. status = OCICollGetElem(env->handle, env->errorHandle, tables, i,
  566. &exists, (dvoid*) &tableDescriptor, &indicator);
  567. if (Environment_CheckForError(env, status,
  568. "Message_Initialize(): get element from collection") < 0)
  569. return -1;
  570. table = (udt_MessageTable*)
  571. g_MessageTableType.tp_alloc(&g_MessageTableType, 0);
  572. if (!table)
  573. return -1;
  574. PyList_SET_ITEM(self->tables, i, (PyObject*) table);
  575. if (MessageTable_Initialize(table, env, *tableDescriptor) < 0)
  576. return -1;
  577. }
  578. }
  579. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  580. if (self->type == OCI_EVENT_QUERYCHANGE) {
  581. // determine query collection
  582. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &queries, NULL,
  583. OCI_ATTR_CHDES_QUERIES, env->errorHandle);
  584. if (Environment_CheckForError(env, status,
  585. "Message_Initialize(): get queries collection") < 0)
  586. return -1;
  587. // determine number of queries
  588. if (!queries)
  589. numQueries = 0;
  590. else {
  591. status = OCICollSize(env->handle, env->errorHandle, queries,
  592. &numQueries);
  593. if (Environment_CheckForError(env, status,
  594. "Message_Initialize(): get size of collection") < 0)
  595. return -1;
  596. }
  597. // create list to hold results
  598. self->queries = PyList_New(numQueries);
  599. if (!self->queries)
  600. return -1;
  601. // populate each entry with a message query instance
  602. for (i = 0; i < numQueries; i++) {
  603. status = OCICollGetElem(env->handle, env->errorHandle, queries, i,
  604. &exists, (dvoid*) &queryDescriptor, &indicator);
  605. if (Environment_CheckForError(env, status,
  606. "Message_Initialize(): get element from collection") < 0)
  607. return -1;
  608. query = (udt_MessageQuery*)
  609. g_MessageQueryType.tp_alloc(&g_MessageQueryType, 0);
  610. if (!query)
  611. return -1;
  612. PyList_SET_ITEM(self->queries, i, (PyObject*) query);
  613. if (MessageQuery_Initialize(query, env, *queryDescriptor) < 0)
  614. return -1;
  615. }
  616. }
  617. #endif
  618. return 0;
  619. }
  620. //-----------------------------------------------------------------------------
  621. // Subscription_CallbackHandler()
  622. // Routine that performs the actual call.
  623. //-----------------------------------------------------------------------------
  624. static int Subscription_CallbackHandler(
  625. udt_Subscription *self, // subscription object
  626. udt_Environment *env, // environment to use
  627. dvoid *descriptor) // descriptor to get information from
  628. {
  629. PyObject *result, *args;
  630. udt_Message *message;
  631. // create the message
  632. message = (udt_Message*) g_MessageType.tp_alloc(&g_MessageType, 0);
  633. if (!message)
  634. return -1;
  635. if (Message_Initialize(message, env, self, descriptor) < 0) {
  636. Py_DECREF(message);
  637. return -1;
  638. }
  639. // create the arguments for the call
  640. args = PyTuple_Pack(1, message);
  641. Py_DECREF(message);
  642. if (!args)
  643. return -1;
  644. // make the actual call
  645. result = PyObject_Call(self->callback, args, NULL);
  646. Py_DECREF(args);
  647. if (!result)
  648. return -1;
  649. Py_DECREF(result);
  650. return 0;
  651. }
  652. //-----------------------------------------------------------------------------
  653. // Subscription_Callback()
  654. // Routine that is called when a callback needs to be invoked.
  655. //-----------------------------------------------------------------------------
  656. static void Subscription_Callback(
  657. udt_Subscription *self, // subscription object
  658. OCISubscription *handle, // subscription handle
  659. dvoid *payload, // payload
  660. ub4 *payloadLength, // payload length
  661. dvoid *descriptor, // descriptor
  662. ub4 mode) // mode used
  663. {
  664. #ifdef WITH_THREAD
  665. PyGILState_STATE gstate = PyGILState_Ensure();
  666. #endif
  667. udt_Environment *env;
  668. // perform the call
  669. env = Environment_NewFromScratch(0, 0, NULL, NULL);
  670. if (!env)
  671. PyErr_Print();
  672. else {
  673. if (Subscription_CallbackHandler(self, env, descriptor) < 0)
  674. PyErr_Print();
  675. Py_DECREF(env);
  676. }
  677. // restore thread state, if necessary
  678. #ifdef WITH_THREAD
  679. PyGILState_Release(gstate);
  680. #endif
  681. }
  682. //-----------------------------------------------------------------------------
  683. // Subscription_Register()
  684. // Register the subscription.
  685. //-----------------------------------------------------------------------------
  686. static int Subscription_Register(
  687. udt_Subscription *self) // subscription to register
  688. {
  689. udt_Environment *env;
  690. sword status;
  691. // create the subscription handle
  692. env = self->connection->environment;
  693. status = OCIHandleAlloc(env->handle, (dvoid**) &self->handle,
  694. OCI_HTYPE_SUBSCRIPTION, 0, 0);
  695. if (Environment_CheckForError(env, status,
  696. "Subscription_Register(): allocate handle") < 0)
  697. return -1;
  698. // set the namespace
  699. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  700. (dvoid*) &self->namespace, sizeof(ub4), OCI_ATTR_SUBSCR_NAMESPACE,
  701. env->errorHandle);
  702. if (Environment_CheckForError(env, status,
  703. "Subscription_Register(): set namespace") < 0)
  704. return -1;
  705. // set the protocol
  706. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  707. (dvoid*) &self->protocol, sizeof(ub4), OCI_ATTR_SUBSCR_RECPTPROTO,
  708. env->errorHandle);
  709. if (Environment_CheckForError(env, status,
  710. "Subscription_Register(): set protocol") < 0)
  711. return -1;
  712. // set the timeout
  713. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  714. (dvoid*) &self->timeout, sizeof(ub4), OCI_ATTR_SUBSCR_TIMEOUT,
  715. env->errorHandle);
  716. if (Environment_CheckForError(env, status,
  717. "Subscription_Register(): set timeout") < 0)
  718. return -1;
  719. // set the TCP port used on client to listen for callback from DB server
  720. if (self->port > 0) {
  721. status = OCIAttrSet(env->handle, OCI_HTYPE_ENV,
  722. (dvoid*) &(self->port), (ub4) 0, OCI_ATTR_SUBSCR_PORTNO,
  723. env->errorHandle);
  724. if (Environment_CheckForError(env, status,
  725. "Subscription_Register(): set port") < 0)
  726. return -1;
  727. }
  728. // set the context for the callback
  729. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  730. (dvoid*) self, 0, OCI_ATTR_SUBSCR_CTX, env->errorHandle);
  731. if (Environment_CheckForError(env, status,
  732. "Subscription_Register(): set context") < 0)
  733. return -1;
  734. // set the callback, if applicable
  735. if (self->callback) {
  736. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  737. (dvoid*) Subscription_Callback, 0, OCI_ATTR_SUBSCR_CALLBACK,
  738. env->errorHandle);
  739. if (Environment_CheckForError(env, status,
  740. "Subscription_Register(): set callback") < 0)
  741. return -1;
  742. }
  743. // set suscription QOS
  744. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  745. (dvoid*) &self->qos, sizeof(ub4), OCI_ATTR_SUBSCR_QOSFLAGS,
  746. env->errorHandle);
  747. if (Environment_CheckForError(env, status,
  748. "Subscription_Register(): set qos flags") < 0)
  749. return -1;
  750. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  751. // set subscription change notification QOS flags
  752. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  753. (dvoid*) &self->cqqos, sizeof(ub4), OCI_ATTR_SUBSCR_CQ_QOSFLAGS,
  754. env->errorHandle);
  755. if (Environment_CheckForError(env, status,
  756. "Subscription_Register(): set cq qos flags") < 0)
  757. return -1;
  758. #endif
  759. // set whether or not rowids are desired
  760. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  761. (dvoid*) &self->rowids, sizeof(ub4), OCI_ATTR_CHNF_ROWIDS,
  762. env->errorHandle);
  763. if (Environment_CheckForError(env, status,
  764. "Subscription_Register(): set rowids") < 0)
  765. return -1;
  766. // set which operations are desired
  767. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  768. (dvoid*) &self->operations, sizeof(ub4), OCI_ATTR_CHNF_OPERATIONS,
  769. env->errorHandle);
  770. if (Environment_CheckForError(env, status,
  771. "Subscription_Register(): set operations") < 0)
  772. return -1;
  773. // register the subscription
  774. Py_BEGIN_ALLOW_THREADS
  775. status = OCISubscriptionRegister(self->connection->handle,
  776. &self->handle, 1, env->errorHandle, OCI_DEFAULT);
  777. Py_END_ALLOW_THREADS
  778. if (Environment_CheckForError(env, status,
  779. "Subscription_Register(): register") < 0)
  780. return -1;
  781. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  782. // get the registration id
  783. status = OCIAttrGet(self->handle, OCI_HTYPE_SUBSCRIPTION, &self->id,
  784. NULL, OCI_ATTR_SUBSCR_CQ_REGID, env->errorHandle);
  785. if (Environment_CheckForError(env, status,
  786. "Subscription_Register(): get registration id") < 0) {
  787. return -1;
  788. }
  789. #endif
  790. return 0;
  791. }
  792. //-----------------------------------------------------------------------------
  793. // Subscription_New()
  794. // Allocate a new subscription object.
  795. //-----------------------------------------------------------------------------
  796. static udt_Subscription *Subscription_New(
  797. udt_Connection *connection, // connection object
  798. ub4 namespace, // namespace to use
  799. ub4 protocol, // protocol to use
  800. ub4 port, // client port for callbacks
  801. PyObject *callback, // callback routine
  802. ub4 timeout, // timeout (in seconds)
  803. ub4 operations, // operations to notify
  804. ub4 qos, // QOS flags
  805. ub4 cqqos, // change notification QOS flags
  806. int rowids) // retrieve rowids?
  807. {
  808. udt_Subscription *self;
  809. self = (udt_Subscription*)
  810. g_SubscriptionType.tp_alloc(&g_SubscriptionType, 0);
  811. if (!self)
  812. return NULL;
  813. Py_INCREF(connection);
  814. self->connection = connection;
  815. Py_XINCREF(callback);
  816. self->callback = callback;
  817. self->namespace = namespace;
  818. self->protocol = protocol;
  819. self->port = port;
  820. self->timeout = timeout;
  821. self->rowids = rowids;
  822. self->operations = operations;
  823. self->qos = qos;
  824. self->cqqos = cqqos;
  825. self->handle = NULL;
  826. self->id = 0;
  827. if (Subscription_Register(self) < 0) {
  828. Py_DECREF(self);
  829. return NULL;
  830. }
  831. return self;
  832. }
  833. //-----------------------------------------------------------------------------
  834. // Subscription_Free()
  835. // Free the memory associated with a subscription.
  836. //-----------------------------------------------------------------------------
  837. static void Subscription_Free(
  838. udt_Subscription *self) // subscription to free
  839. {
  840. if (self->handle)
  841. OCISubscriptionUnRegister(self->connection->handle,
  842. self->handle, self->connection->environment->errorHandle,
  843. OCI_DEFAULT);
  844. Py_CLEAR(self->connection);
  845. Py_CLEAR(self->callback);
  846. Py_TYPE(self)->tp_free((PyObject*) self);
  847. }
  848. //-----------------------------------------------------------------------------
  849. // Subscription_Repr()
  850. // Return a string representation of the subscription.
  851. //-----------------------------------------------------------------------------
  852. static PyObject *Subscription_Repr(
  853. udt_Subscription *subscription) // subscription to repr
  854. {
  855. PyObject *connectionRepr, *module, *name, *result, *format, *formatArgs;
  856. format = cxString_FromAscii("<%s.%s on %s>");
  857. if (!format)
  858. return NULL;
  859. connectionRepr = PyObject_Repr((PyObject*) subscription->connection);
  860. if (!connectionRepr) {
  861. Py_DECREF(format);
  862. return NULL;
  863. }
  864. if (GetModuleAndName(Py_TYPE(subscription), &module, &name) < 0) {
  865. Py_DECREF(format);
  866. Py_DECREF(connectionRepr);
  867. return NULL;
  868. }
  869. formatArgs = PyTuple_Pack(3, module, name, connectionRepr);
  870. Py_DECREF(module);
  871. Py_DECREF(name);
  872. Py_DECREF(connectionRepr);
  873. if (!formatArgs) {
  874. Py_DECREF(format);
  875. return NULL;
  876. }
  877. result = cxString_Format(format, formatArgs);
  878. Py_DECREF(format);
  879. Py_DECREF(formatArgs);
  880. return result;
  881. }
  882. //-----------------------------------------------------------------------------
  883. // Subscription_RegisterQuery()
  884. // Register a query for database change notification.
  885. //-----------------------------------------------------------------------------
  886. static PyObject *Subscription_RegisterQuery(
  887. udt_Subscription *self, // subscription to use
  888. PyObject *args) // arguments
  889. {
  890. PyObject *statement, *executeArgs;
  891. udt_Buffer statementBuffer;
  892. udt_Environment *env;
  893. udt_Cursor *cursor;
  894. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  895. ub8 queryid;
  896. #endif
  897. sword status;
  898. // parse arguments
  899. executeArgs = NULL;
  900. if (!PyArg_ParseTuple(args, "O!|O", cxString_Type, &statement,
  901. &executeArgs))
  902. return NULL;
  903. if (executeArgs) {
  904. if (!PyDict_Check(executeArgs) && !PySequence_Check(executeArgs)) {
  905. PyErr_SetString(PyExc_TypeError,
  906. "expecting a dictionary or sequence");
  907. return NULL;
  908. }
  909. }
  910. // create cursor to perform query
  911. env = self->connection->environment;
  912. cursor = (udt_Cursor*) Connection_NewCursor(self->connection, NULL);
  913. if (!cursor)
  914. return NULL;
  915. // allocate the handle so the subscription handle can be set
  916. if (Cursor_AllocateHandle(cursor) < 0) {
  917. Py_DECREF(cursor);
  918. return NULL;
  919. }
  920. // prepare the statement for execution
  921. if (cxBuffer_FromObject(&statementBuffer, statement,
  922. env->encoding) < 0) {
  923. Py_DECREF(cursor);
  924. return NULL;
  925. }
  926. status = OCIStmtPrepare(cursor->handle, env->errorHandle,
  927. (text*) statementBuffer.ptr, statementBuffer.size, OCI_NTV_SYNTAX,
  928. OCI_DEFAULT);
  929. cxBuffer_Clear(&statementBuffer);
  930. if (Environment_CheckForError(env, status,
  931. "Subscription_RegisterQuery(): prepare statement") < 0) {
  932. Py_DECREF(cursor);
  933. return NULL;
  934. }
  935. // perform binds
  936. if (executeArgs && Cursor_SetBindVariables(cursor, executeArgs, 1, 0,
  937. 0) < 0) {
  938. Py_DECREF(cursor);
  939. return NULL;
  940. }
  941. if (Cursor_PerformBind(cursor) < 0) {
  942. Py_DECREF(cursor);
  943. return NULL;
  944. }
  945. // parse the query in order to get the defined variables
  946. Py_BEGIN_ALLOW_THREADS
  947. status = OCIStmtExecute(self->connection->handle, cursor->handle,
  948. env->errorHandle, 0, 0, 0, 0, OCI_DESCRIBE_ONLY);
  949. Py_END_ALLOW_THREADS
  950. if (Environment_CheckForError(env, status,
  951. "Subscription_RegisterQuery(): parse statement") < 0) {
  952. Py_DECREF(cursor);
  953. return NULL;
  954. }
  955. // perform define as needed
  956. if (Cursor_PerformDefine(cursor) < 0) {
  957. Py_DECREF(cursor);
  958. return NULL;
  959. }
  960. // set the subscription handle
  961. status = OCIAttrSet(cursor->handle, OCI_HTYPE_STMT, self->handle, 0,
  962. OCI_ATTR_CHNF_REGHANDLE, env->errorHandle);
  963. if (Environment_CheckForError(env, status,
  964. "Subscription_RegisterQuery(): set subscription handle") < 0) {
  965. Py_DECREF(cursor);
  966. return NULL;
  967. }
  968. // execute the query which registers it
  969. if (Cursor_InternalExecute(cursor, 0, 0) < 0) {
  970. Py_DECREF(cursor);
  971. return NULL;
  972. }
  973. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  974. if (self->cqqos & OCI_SUBSCR_CQ_QOS_QUERY) {
  975. // get the query id
  976. status = OCIAttrGet(cursor->handle, OCI_HTYPE_STMT, &queryid, NULL,
  977. OCI_ATTR_CQ_QUERYID, env->errorHandle);
  978. if (Environment_CheckForError(env, status,
  979. "Subscription_RegisterQuery(): get query id") < 0) {
  980. Py_DECREF(cursor);
  981. return NULL;
  982. }
  983. }
  984. #endif
  985. Py_DECREF(cursor);
  986. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  987. if (self->cqqos & OCI_SUBSCR_CQ_QOS_QUERY)
  988. return PyInt_FromLong(queryid);
  989. #endif
  990. Py_INCREF(Py_None);
  991. return Py_None;
  992. }
  993. //-----------------------------------------------------------------------------
  994. // Message_Free()
  995. // Free the memory associated with a message.
  996. //-----------------------------------------------------------------------------
  997. static void Message_Free(
  998. udt_Message *self) // object to free
  999. {
  1000. Py_CLEAR(self->subscription);
  1001. Py_CLEAR(self->dbname);
  1002. Py_CLEAR(self->tables);
  1003. Py_CLEAR(self->queries);
  1004. Py_TYPE(self)->tp_free((PyObject*) self);
  1005. }
  1006. //-----------------------------------------------------------------------------
  1007. // MessageTable_Free()
  1008. // Free the memory associated with a table in a message.
  1009. //-----------------------------------------------------------------------------
  1010. static void MessageTable_Free(
  1011. udt_MessageTable *self) // object to free
  1012. {
  1013. Py_CLEAR(self->name);
  1014. Py_CLEAR(self->rows);
  1015. Py_TYPE(self)->tp_free((PyObject*) self);
  1016. }
  1017. //-----------------------------------------------------------------------------
  1018. // MessageRow_Free()
  1019. // Free the memory associated with a row in a message.
  1020. //-----------------------------------------------------------------------------
  1021. static void MessageRow_Free(
  1022. udt_MessageRow *self) // object to free
  1023. {
  1024. Py_CLEAR(self->rowid);
  1025. Py_TYPE(self)->tp_free((PyObject*) self);
  1026. }
  1027. #if ORACLE_VERSION_HEX >= ORACLE_VERSION(11, 1)
  1028. //-----------------------------------------------------------------------------
  1029. // MessageQuery_Free()
  1030. // Free the memory associated with a query in a message.
  1031. //-----------------------------------------------------------------------------
  1032. static void MessageQuery_Free(
  1033. udt_MessageQuery *self) // object to free
  1034. {
  1035. Py_CLEAR(self->tables);
  1036. Py_TYPE(self)->tp_free((PyObject*) self);
  1037. }
  1038. #endif