Subscription.c 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873
  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 rowids;
  19. } udt_Subscription;
  20. typedef struct {
  21. PyObject_HEAD
  22. ub4 type;
  23. PyObject *dbname;
  24. PyObject *tables;
  25. } udt_Message;
  26. typedef struct {
  27. PyObject_HEAD
  28. PyObject *name;
  29. PyObject *rows;
  30. ub4 operation;
  31. } udt_MessageTable;
  32. typedef struct {
  33. PyObject_HEAD
  34. PyObject *rowid;
  35. ub4 operation;
  36. } udt_MessageRow;
  37. //-----------------------------------------------------------------------------
  38. // Declaration of subscription functions
  39. //-----------------------------------------------------------------------------
  40. static void Subscription_Free(udt_Subscription*);
  41. static PyObject *Subscription_Repr(udt_Subscription*);
  42. static PyObject *Subscription_RegisterQuery(udt_Subscription*, PyObject*);
  43. static void Message_Free(udt_Message*);
  44. static void MessageTable_Free(udt_MessageTable*);
  45. static void MessageRow_Free(udt_MessageRow*);
  46. //-----------------------------------------------------------------------------
  47. // declaration of members for Python types
  48. //-----------------------------------------------------------------------------
  49. static PyMemberDef g_SubscriptionTypeMembers[] = {
  50. { "callback", T_OBJECT, offsetof(udt_Subscription, callback), READONLY },
  51. { "connection", T_OBJECT, offsetof(udt_Subscription, connection),
  52. READONLY },
  53. { "namespace", T_INT, offsetof(udt_Subscription, namespace), READONLY },
  54. { "protocol", T_INT, offsetof(udt_Subscription, protocol), READONLY },
  55. { "port", T_INT, offsetof(udt_Subscription, port), READONLY },
  56. { "timeout", T_INT, offsetof(udt_Subscription, timeout), READONLY },
  57. { "operations", T_INT, offsetof(udt_Subscription, operations), READONLY },
  58. { "rowids", T_BOOL, offsetof(udt_Subscription, rowids), READONLY },
  59. { NULL }
  60. };
  61. static PyMemberDef g_MessageTypeMembers[] = {
  62. { "type", T_INT, offsetof(udt_Message, type), READONLY },
  63. { "dbname", T_OBJECT, offsetof(udt_Message, dbname), READONLY },
  64. { "tables", T_OBJECT, offsetof(udt_Message, tables), READONLY },
  65. { NULL }
  66. };
  67. static PyMemberDef g_MessageTableTypeMembers[] = {
  68. { "name", T_OBJECT, offsetof(udt_MessageTable, name), READONLY },
  69. { "rows", T_OBJECT, offsetof(udt_MessageTable, rows), READONLY },
  70. { "operation", T_INT, offsetof(udt_MessageTable, operation), READONLY },
  71. { NULL }
  72. };
  73. static PyMemberDef g_MessageRowTypeMembers[] = {
  74. { "rowid", T_OBJECT, offsetof(udt_MessageRow, rowid), READONLY },
  75. { "operation", T_INT, offsetof(udt_MessageRow, operation), READONLY },
  76. { NULL }
  77. };
  78. //-----------------------------------------------------------------------------
  79. // declaration of methods for Python types
  80. //-----------------------------------------------------------------------------
  81. static PyMethodDef g_SubscriptionTypeMethods[] = {
  82. { "registerquery", (PyCFunction) Subscription_RegisterQuery,
  83. METH_VARARGS },
  84. { NULL, NULL }
  85. };
  86. //-----------------------------------------------------------------------------
  87. // Python type declarations
  88. //-----------------------------------------------------------------------------
  89. static PyTypeObject g_SubscriptionType = {
  90. PyVarObject_HEAD_INIT(NULL, 0)
  91. "cx_Oracle.Subscription", // tp_name
  92. sizeof(udt_Subscription), // tp_basicsize
  93. 0, // tp_itemsize
  94. (destructor) Subscription_Free, // tp_dealloc
  95. 0, // tp_print
  96. 0, // tp_getattr
  97. 0, // tp_setattr
  98. 0, // tp_compare
  99. (reprfunc) Subscription_Repr, // tp_repr
  100. 0, // tp_as_number
  101. 0, // tp_as_sequence
  102. 0, // tp_as_mapping
  103. 0, // tp_hash
  104. 0, // tp_call
  105. 0, // tp_str
  106. 0, // tp_getattro
  107. 0, // tp_setattro
  108. 0, // tp_as_buffer
  109. Py_TPFLAGS_DEFAULT, // tp_flags
  110. 0, // tp_doc
  111. 0, // tp_traverse
  112. 0, // tp_clear
  113. 0, // tp_richcompare
  114. 0, // tp_weaklistoffset
  115. 0, // tp_iter
  116. 0, // tp_iternext
  117. g_SubscriptionTypeMethods, // tp_methods
  118. g_SubscriptionTypeMembers, // tp_members
  119. 0, // tp_getset
  120. 0, // tp_base
  121. 0, // tp_dict
  122. 0, // tp_descr_get
  123. 0, // tp_descr_set
  124. 0, // tp_dictoffset
  125. 0, // tp_init
  126. 0, // tp_alloc
  127. 0, // tp_new
  128. 0, // tp_free
  129. 0, // tp_is_gc
  130. 0 // tp_bases
  131. };
  132. static PyTypeObject g_MessageType = {
  133. PyVarObject_HEAD_INIT(NULL, 0)
  134. "cx_Oracle.Message", // tp_name
  135. sizeof(udt_Message), // tp_basicsize
  136. 0, // tp_itemsize
  137. (destructor) Message_Free, // tp_dealloc
  138. 0, // tp_print
  139. 0, // tp_getattr
  140. 0, // tp_setattr
  141. 0, // tp_compare
  142. 0, // tp_repr
  143. 0, // tp_as_number
  144. 0, // tp_as_sequence
  145. 0, // tp_as_mapping
  146. 0, // tp_hash
  147. 0, // tp_call
  148. 0, // tp_str
  149. 0, // tp_getattro
  150. 0, // tp_setattro
  151. 0, // tp_as_buffer
  152. Py_TPFLAGS_DEFAULT, // tp_flags
  153. 0, // tp_doc
  154. 0, // tp_traverse
  155. 0, // tp_clear
  156. 0, // tp_richcompare
  157. 0, // tp_weaklistoffset
  158. 0, // tp_iter
  159. 0, // tp_iternext
  160. 0, // tp_methods
  161. g_MessageTypeMembers, // tp_members
  162. 0, // tp_getset
  163. 0, // tp_base
  164. 0, // tp_dict
  165. 0, // tp_descr_get
  166. 0, // tp_descr_set
  167. 0, // tp_dictoffset
  168. 0, // tp_init
  169. 0, // tp_alloc
  170. 0, // tp_new
  171. 0, // tp_free
  172. 0, // tp_is_gc
  173. 0 // tp_bases
  174. };
  175. static PyTypeObject g_MessageTableType = {
  176. PyVarObject_HEAD_INIT(NULL, 0)
  177. "cx_Oracle.MessageTable", // tp_name
  178. sizeof(udt_MessageTable), // tp_basicsize
  179. 0, // tp_itemsize
  180. (destructor) MessageTable_Free, // tp_dealloc
  181. 0, // tp_print
  182. 0, // tp_getattr
  183. 0, // tp_setattr
  184. 0, // tp_compare
  185. 0, // tp_repr
  186. 0, // tp_as_number
  187. 0, // tp_as_sequence
  188. 0, // tp_as_mapping
  189. 0, // tp_hash
  190. 0, // tp_call
  191. 0, // tp_str
  192. 0, // tp_getattro
  193. 0, // tp_setattro
  194. 0, // tp_as_buffer
  195. Py_TPFLAGS_DEFAULT, // tp_flags
  196. 0, // tp_doc
  197. 0, // tp_traverse
  198. 0, // tp_clear
  199. 0, // tp_richcompare
  200. 0, // tp_weaklistoffset
  201. 0, // tp_iter
  202. 0, // tp_iternext
  203. 0, // tp_methods
  204. g_MessageTableTypeMembers, // tp_members
  205. 0, // tp_getset
  206. 0, // tp_base
  207. 0, // tp_dict
  208. 0, // tp_descr_get
  209. 0, // tp_descr_set
  210. 0, // tp_dictoffset
  211. 0, // tp_init
  212. 0, // tp_alloc
  213. 0, // tp_new
  214. 0, // tp_free
  215. 0, // tp_is_gc
  216. 0 // tp_bases
  217. };
  218. static PyTypeObject g_MessageRowType = {
  219. PyVarObject_HEAD_INIT(NULL, 0)
  220. "cx_Oracle.MessageRow", // tp_name
  221. sizeof(udt_MessageRow), // tp_basicsize
  222. 0, // tp_itemsize
  223. (destructor) MessageRow_Free, // tp_dealloc
  224. 0, // tp_print
  225. 0, // tp_getattr
  226. 0, // tp_setattr
  227. 0, // tp_compare
  228. 0, // tp_repr
  229. 0, // tp_as_number
  230. 0, // tp_as_sequence
  231. 0, // tp_as_mapping
  232. 0, // tp_hash
  233. 0, // tp_call
  234. 0, // tp_str
  235. 0, // tp_getattro
  236. 0, // tp_setattro
  237. 0, // tp_as_buffer
  238. Py_TPFLAGS_DEFAULT, // tp_flags
  239. 0, // tp_doc
  240. 0, // tp_traverse
  241. 0, // tp_clear
  242. 0, // tp_richcompare
  243. 0, // tp_weaklistoffset
  244. 0, // tp_iter
  245. 0, // tp_iternext
  246. 0, // tp_methods
  247. g_MessageRowTypeMembers, // tp_members
  248. 0, // tp_getset
  249. 0, // tp_base
  250. 0, // tp_dict
  251. 0, // tp_descr_get
  252. 0, // tp_descr_set
  253. 0, // tp_dictoffset
  254. 0, // tp_init
  255. 0, // tp_alloc
  256. 0, // tp_new
  257. 0, // tp_free
  258. 0, // tp_is_gc
  259. 0 // tp_bases
  260. };
  261. //-----------------------------------------------------------------------------
  262. // MessageRow_Initialize()
  263. // Initialize a new message row with the information from the descriptor.
  264. //-----------------------------------------------------------------------------
  265. static int MessageRow_Initialize(
  266. udt_MessageRow *self, // object to initialize
  267. udt_Environment *env, // environment to use
  268. dvoid *descriptor) // descriptor to get information from
  269. {
  270. ub4 rowidLength;
  271. sword status;
  272. char *rowid;
  273. // determine operation
  274. status = OCIAttrGet(descriptor, OCI_DTYPE_ROW_CHDES, &self->operation,
  275. NULL, OCI_ATTR_CHDES_ROW_OPFLAGS, env->errorHandle);
  276. if (Environment_CheckForError(env, status,
  277. "MessageRow_Initialize(): get operation") < 0)
  278. return -1;
  279. // determine table name
  280. status = OCIAttrGet(descriptor, OCI_DTYPE_ROW_CHDES, &rowid, &rowidLength,
  281. OCI_ATTR_CHDES_ROW_ROWID, env->errorHandle);
  282. if (Environment_CheckForError(env, status,
  283. "MessageRow_Initialize(): get rowid") < 0)
  284. return -1;
  285. self->rowid = cxString_FromEncodedString(rowid, rowidLength,
  286. env->encoding);
  287. if (!self->rowid)
  288. return -1;
  289. return 0;
  290. }
  291. //-----------------------------------------------------------------------------
  292. // MessageTable_Initialize()
  293. // Initialize a new message table with the information from the descriptor.
  294. //-----------------------------------------------------------------------------
  295. static int MessageTable_Initialize(
  296. udt_MessageTable *self, // object to initialize
  297. udt_Environment *env, // environment to use
  298. dvoid *descriptor) // descriptor to get information from
  299. {
  300. dvoid **rowDescriptor, *indicator;
  301. ub4 nameLength, i;
  302. udt_MessageRow *row;
  303. boolean exists;
  304. OCIColl *rows;
  305. sword status;
  306. sb4 numRows;
  307. char *name;
  308. // determine operation
  309. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &self->operation,
  310. NULL, OCI_ATTR_CHDES_TABLE_OPFLAGS, env->errorHandle);
  311. if (Environment_CheckForError(env, status,
  312. "MessageTable_Initialize(): get operation") < 0)
  313. return -1;
  314. // determine table name
  315. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &name, &nameLength,
  316. OCI_ATTR_CHDES_TABLE_NAME, env->errorHandle);
  317. if (Environment_CheckForError(env, status,
  318. "MessageTable_Initialize(): get table name") < 0)
  319. return -1;
  320. self->name = cxString_FromEncodedString(name, nameLength, env->encoding);
  321. if (!self->name)
  322. return -1;
  323. // if change invalidated all rows, nothing to do
  324. if (self->operation & OCI_OPCODE_ALLROWS)
  325. return 0;
  326. // determine rows collection
  327. status = OCIAttrGet(descriptor, OCI_DTYPE_TABLE_CHDES, &rows, NULL,
  328. OCI_ATTR_CHDES_TABLE_ROW_CHANGES, env->errorHandle);
  329. if (Environment_CheckForError(env, status,
  330. "MessageTable_Initialize(): get rows collection") < 0)
  331. return -1;
  332. // determine number of rows in collection
  333. status = OCICollSize(env->handle, env->errorHandle, rows, &numRows);
  334. if (Environment_CheckForError(env, status,
  335. "MessageTable_Initialize(): get size of rows collection") < 0)
  336. return -1;
  337. // populate the rows attribute
  338. self->rows = PyList_New(numRows);
  339. if (!self->rows)
  340. return -1;
  341. for (i = 0; i < numRows; i++) {
  342. status = OCICollGetElem(env->handle, env->errorHandle, rows, i,
  343. &exists, (dvoid*) &rowDescriptor, &indicator);
  344. if (Environment_CheckForError(env, status,
  345. "MessageTable_Initialize(): get element from collection") < 0)
  346. return -1;
  347. row = (udt_MessageRow*)
  348. g_MessageRowType.tp_alloc(&g_MessageRowType, 0);
  349. if (!row)
  350. return -1;
  351. PyList_SET_ITEM(self->rows, i, (PyObject*) row);
  352. if (MessageRow_Initialize(row, env, *rowDescriptor) < 0)
  353. return -1;
  354. }
  355. return 0;
  356. }
  357. //-----------------------------------------------------------------------------
  358. // Message_Initialize()
  359. // Initialize a new message with the information from the descriptor.
  360. //-----------------------------------------------------------------------------
  361. static int Message_Initialize(
  362. udt_Message *self, // object to initialize
  363. udt_Environment *env, // environment to use
  364. dvoid *descriptor) // descriptor to get information from
  365. {
  366. dvoid **tableDescriptor, *indicator;
  367. udt_MessageTable *table;
  368. ub4 dbnameLength, i;
  369. OCIColl *tables;
  370. boolean exists;
  371. sb4 numTables;
  372. char *dbname;
  373. sword status;
  374. // determine type
  375. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &self->type, NULL,
  376. OCI_ATTR_CHDES_NFYTYPE, env->errorHandle);
  377. if (Environment_CheckForError(env, status,
  378. "Message_Initialize(): get type") < 0)
  379. return -1;
  380. // determine database name
  381. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &dbname, &dbnameLength,
  382. OCI_ATTR_CHDES_DBNAME, env->errorHandle);
  383. if (Environment_CheckForError(env, status,
  384. "Message_Initialize(): get database name") < 0)
  385. return -1;
  386. self->dbname = cxString_FromEncodedString(dbname, dbnameLength,
  387. env->encoding);
  388. if (!self->dbname)
  389. return -1;
  390. // determine table collection
  391. status = OCIAttrGet(descriptor, OCI_DTYPE_CHDES, &tables, NULL,
  392. OCI_ATTR_CHDES_TABLE_CHANGES, env->errorHandle);
  393. if (Environment_CheckForError(env, status,
  394. "Message_Initialize(): get tables collection") < 0)
  395. return -1;
  396. // determine number of tables
  397. if (!tables)
  398. numTables = 0;
  399. else {
  400. status = OCICollSize(env->handle, env->errorHandle, tables,
  401. &numTables);
  402. if (Environment_CheckForError(env, status,
  403. "Message_Initialize(): get size of collection") < 0)
  404. return -1;
  405. }
  406. // create list to hold results
  407. self->tables = PyList_New(numTables);
  408. if (!self->tables)
  409. return -1;
  410. // populate each entry with a message table instance
  411. for (i = 0; i < numTables; i++) {
  412. status = OCICollGetElem(env->handle, env->errorHandle, tables, i,
  413. &exists, (dvoid*) &tableDescriptor, &indicator);
  414. if (Environment_CheckForError(env, status,
  415. "Message_Initialize(): get element from collection") < 0)
  416. return -1;
  417. table = (udt_MessageTable*)
  418. g_MessageTableType.tp_alloc(&g_MessageTableType, 0);
  419. if (!table)
  420. return -1;
  421. PyList_SET_ITEM(self->tables, i, (PyObject*) table);
  422. if (MessageTable_Initialize(table, env, *tableDescriptor) < 0)
  423. return -1;
  424. }
  425. return 0;
  426. }
  427. //-----------------------------------------------------------------------------
  428. // Subscription_CallbackHandler()
  429. // Routine that performs the actual call.
  430. //-----------------------------------------------------------------------------
  431. static int Subscription_CallbackHandler(
  432. udt_Subscription *self, // subscription object
  433. udt_Environment *env, // environment to use
  434. dvoid *descriptor) // descriptor to get information from
  435. {
  436. PyObject *result, *args;
  437. udt_Message *message;
  438. // create the message
  439. message = (udt_Message*) g_MessageType.tp_alloc(&g_MessageType, 0);
  440. if (!message)
  441. return -1;
  442. if (Message_Initialize(message, env, descriptor) < 0) {
  443. Py_DECREF(message);
  444. return -1;
  445. }
  446. // create the arguments for the call
  447. args = PyTuple_Pack(1, message);
  448. Py_DECREF(message);
  449. if (!args)
  450. return -1;
  451. // make the actual call
  452. result = PyObject_Call(self->callback, args, NULL);
  453. Py_DECREF(args);
  454. if (!result)
  455. return -1;
  456. Py_DECREF(result);
  457. return 0;
  458. }
  459. //-----------------------------------------------------------------------------
  460. // Subscription_Callback()
  461. // Routine that is called when a callback needs to be invoked.
  462. //-----------------------------------------------------------------------------
  463. static void Subscription_Callback(
  464. udt_Subscription *self, // subscription object
  465. OCISubscription *handle, // subscription handle
  466. dvoid *payload, // payload
  467. ub4 *payloadLength, // payload length
  468. dvoid *descriptor, // descriptor
  469. ub4 mode) // mode used
  470. {
  471. #ifdef WITH_THREAD
  472. PyGILState_STATE gstate = PyGILState_Ensure();
  473. #endif
  474. udt_Environment *env;
  475. // perform the call
  476. env = Environment_NewFromScratch(0, 0, NULL, NULL);
  477. if (!env)
  478. PyErr_Print();
  479. else {
  480. if (Subscription_CallbackHandler(self, env, descriptor) < 0)
  481. PyErr_Print();
  482. Py_DECREF(env);
  483. }
  484. // restore thread state, if necessary
  485. #ifdef WITH_THREAD
  486. PyGILState_Release(gstate);
  487. #endif
  488. }
  489. //-----------------------------------------------------------------------------
  490. // Subscription_Register()
  491. // Register the subscription.
  492. //-----------------------------------------------------------------------------
  493. static int Subscription_Register(
  494. udt_Subscription *self) // subscription to register
  495. {
  496. udt_Environment *env;
  497. sword status;
  498. // create the subscription handle
  499. env = self->connection->environment;
  500. status = OCIHandleAlloc(env->handle, (dvoid**) &self->handle,
  501. OCI_HTYPE_SUBSCRIPTION, 0, 0);
  502. if (Environment_CheckForError(env, status,
  503. "Subscription_Register(): allocate handle") < 0)
  504. return -1;
  505. // set the namespace
  506. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  507. (dvoid*) &self->namespace, sizeof(ub4), OCI_ATTR_SUBSCR_NAMESPACE,
  508. env->errorHandle);
  509. if (Environment_CheckForError(env, status,
  510. "Subscription_Register(): set namespace") < 0)
  511. return -1;
  512. // set the protocol
  513. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  514. (dvoid*) &self->protocol, sizeof(ub4), OCI_ATTR_SUBSCR_RECPTPROTO,
  515. env->errorHandle);
  516. if (Environment_CheckForError(env, status,
  517. "Subscription_Register(): set protocol") < 0)
  518. return -1;
  519. // set the timeout
  520. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  521. (dvoid*) &self->timeout, sizeof(ub4), OCI_ATTR_SUBSCR_TIMEOUT,
  522. env->errorHandle);
  523. if (Environment_CheckForError(env, status,
  524. "Subscription_Register(): set timeout") < 0)
  525. return -1;
  526. // set the TCP port used on client to listen for callback from DB server
  527. if (self->port > 0) {
  528. status = OCIAttrSet(env->handle, OCI_HTYPE_ENV,
  529. (dvoid*) &(self->port), (ub4) 0, OCI_ATTR_SUBSCR_PORTNO,
  530. env->errorHandle);
  531. if (Environment_CheckForError(env, status,
  532. "Subscription_Register(): set port") < 0)
  533. return -1;
  534. }
  535. // set the context for the callback
  536. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  537. (dvoid*) self, 0, OCI_ATTR_SUBSCR_CTX, env->errorHandle);
  538. if (Environment_CheckForError(env, status,
  539. "Subscription_Register(): set context") < 0)
  540. return -1;
  541. // set the callback, if applicable
  542. if (self->callback) {
  543. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  544. (dvoid*) Subscription_Callback, 0, OCI_ATTR_SUBSCR_CALLBACK,
  545. env->errorHandle);
  546. if (Environment_CheckForError(env, status,
  547. "Subscription_Register(): set callback") < 0)
  548. return -1;
  549. }
  550. // set whether or not rowids are desired
  551. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  552. (dvoid*) &self->rowids, sizeof(ub4), OCI_ATTR_CHNF_ROWIDS,
  553. env->errorHandle);
  554. if (Environment_CheckForError(env, status,
  555. "Subscription_Register(): set rowids") < 0)
  556. return -1;
  557. // set which operations are desired
  558. status = OCIAttrSet(self->handle, OCI_HTYPE_SUBSCRIPTION,
  559. (dvoid*) &self->operations, sizeof(ub4), OCI_ATTR_CHNF_OPERATIONS,
  560. env->errorHandle);
  561. if (Environment_CheckForError(env, status,
  562. "Subscription_Register(): set operations") < 0)
  563. return -1;
  564. // register the subscription
  565. Py_BEGIN_ALLOW_THREADS
  566. status = OCISubscriptionRegister(self->connection->handle,
  567. &self->handle, 1, env->errorHandle, OCI_DEFAULT);
  568. Py_END_ALLOW_THREADS
  569. if (Environment_CheckForError(env, status,
  570. "Subscription_Register(): register") < 0)
  571. return -1;
  572. return 0;
  573. }
  574. //-----------------------------------------------------------------------------
  575. // Subscription_New()
  576. // Allocate a new subscription object.
  577. //-----------------------------------------------------------------------------
  578. static udt_Subscription *Subscription_New(
  579. udt_Connection *connection, // connection object
  580. ub4 namespace, // namespace to use
  581. ub4 protocol, // protocol to use
  582. ub4 port, // client port for callbacks
  583. PyObject *callback, // callback routine
  584. ub4 timeout, // timeout (in seconds)
  585. ub4 operations, // operations to notify
  586. int rowids) // retrieve rowids?
  587. {
  588. udt_Subscription *self;
  589. self = (udt_Subscription*)
  590. g_SubscriptionType.tp_alloc(&g_SubscriptionType, 0);
  591. if (!self)
  592. return NULL;
  593. Py_INCREF(connection);
  594. self->connection = connection;
  595. Py_XINCREF(callback);
  596. self->callback = callback;
  597. self->namespace = namespace;
  598. self->protocol = protocol;
  599. self->port = port;
  600. self->timeout = timeout;
  601. self->rowids = rowids;
  602. self->operations = operations;
  603. self->handle = NULL;
  604. if (Subscription_Register(self) < 0) {
  605. Py_DECREF(self);
  606. return NULL;
  607. }
  608. return self;
  609. }
  610. //-----------------------------------------------------------------------------
  611. // Subscription_Free()
  612. // Free the memory associated with a subscription.
  613. //-----------------------------------------------------------------------------
  614. static void Subscription_Free(
  615. udt_Subscription *self) // subscription to free
  616. {
  617. if (self->handle)
  618. OCISubscriptionUnRegister(self->connection->handle,
  619. self->handle, self->connection->environment->errorHandle,
  620. OCI_DEFAULT);
  621. Py_CLEAR(self->connection);
  622. Py_CLEAR(self->callback);
  623. Py_TYPE(self)->tp_free((PyObject*) self);
  624. }
  625. //-----------------------------------------------------------------------------
  626. // Subscription_Repr()
  627. // Return a string representation of the subscription.
  628. //-----------------------------------------------------------------------------
  629. static PyObject *Subscription_Repr(
  630. udt_Subscription *subscription) // subscription to repr
  631. {
  632. PyObject *connectionRepr, *module, *name, *result, *format, *formatArgs;
  633. format = cxString_FromAscii("<%s.%s on %s>");
  634. if (!format)
  635. return NULL;
  636. connectionRepr = PyObject_Repr((PyObject*) subscription->connection);
  637. if (!connectionRepr) {
  638. Py_DECREF(format);
  639. return NULL;
  640. }
  641. if (GetModuleAndName(Py_TYPE(subscription), &module, &name) < 0) {
  642. Py_DECREF(format);
  643. Py_DECREF(connectionRepr);
  644. return NULL;
  645. }
  646. formatArgs = PyTuple_Pack(3, module, name, connectionRepr);
  647. Py_DECREF(module);
  648. Py_DECREF(name);
  649. Py_DECREF(connectionRepr);
  650. if (!formatArgs) {
  651. Py_DECREF(format);
  652. return NULL;
  653. }
  654. result = cxString_Format(format, formatArgs);
  655. Py_DECREF(format);
  656. Py_DECREF(formatArgs);
  657. return result;
  658. }
  659. //-----------------------------------------------------------------------------
  660. // Subscription_RegisterQuery()
  661. // Register a query for database change notification.
  662. //-----------------------------------------------------------------------------
  663. static PyObject *Subscription_RegisterQuery(
  664. udt_Subscription *self, // subscription to use
  665. PyObject *args) // arguments
  666. {
  667. PyObject *statement, *executeArgs;
  668. udt_Buffer statementBuffer;
  669. udt_Environment *env;
  670. udt_Cursor *cursor;
  671. sword status;
  672. // parse arguments
  673. executeArgs = NULL;
  674. if (!PyArg_ParseTuple(args, "O!|O", cxString_Type, &statement,
  675. &executeArgs))
  676. return NULL;
  677. if (executeArgs) {
  678. if (!PyDict_Check(executeArgs) && !PySequence_Check(executeArgs)) {
  679. PyErr_SetString(PyExc_TypeError,
  680. "expecting a dictionary or sequence");
  681. return NULL;
  682. }
  683. }
  684. // create cursor to perform query
  685. env = self->connection->environment;
  686. cursor = (udt_Cursor*) Connection_NewCursor(self->connection, NULL);
  687. if (!cursor)
  688. return NULL;
  689. // allocate the handle so the subscription handle can be set
  690. if (Cursor_AllocateHandle(cursor) < 0) {
  691. Py_DECREF(cursor);
  692. return NULL;
  693. }
  694. // prepare the statement for execution
  695. if (cxBuffer_FromObject(&statementBuffer, statement,
  696. env->encoding) < 0) {
  697. Py_DECREF(cursor);
  698. return NULL;
  699. }
  700. status = OCIStmtPrepare(cursor->handle, env->errorHandle,
  701. (text*) statementBuffer.ptr, statementBuffer.size, OCI_NTV_SYNTAX,
  702. OCI_DEFAULT);
  703. cxBuffer_Clear(&statementBuffer);
  704. if (Environment_CheckForError(env, status,
  705. "Subscription_RegisterQuery(): prepare statement") < 0) {
  706. Py_DECREF(cursor);
  707. return NULL;
  708. }
  709. // perform binds
  710. if (executeArgs && Cursor_SetBindVariables(cursor, executeArgs, 1, 0,
  711. 0) < 0) {
  712. Py_DECREF(cursor);
  713. return NULL;
  714. }
  715. if (Cursor_PerformBind(cursor) < 0) {
  716. Py_DECREF(cursor);
  717. return NULL;
  718. }
  719. // parse the query in order to get the defined variables
  720. Py_BEGIN_ALLOW_THREADS
  721. status = OCIStmtExecute(self->connection->handle, cursor->handle,
  722. env->errorHandle, 0, 0, 0, 0, OCI_DESCRIBE_ONLY);
  723. Py_END_ALLOW_THREADS
  724. if (Environment_CheckForError(env, status,
  725. "Subscription_RegisterQuery(): parse statement") < 0) {
  726. Py_DECREF(cursor);
  727. return NULL;
  728. }
  729. // perform define as needed
  730. if (Cursor_PerformDefine(cursor) < 0) {
  731. Py_DECREF(cursor);
  732. return NULL;
  733. }
  734. // set the subscription handle
  735. status = OCIAttrSet(cursor->handle, OCI_HTYPE_STMT, self->handle, 0,
  736. OCI_ATTR_CHNF_REGHANDLE, env->errorHandle);
  737. if (Environment_CheckForError(env, status,
  738. "Subscription_RegisterQuery(): set subscription handle") < 0) {
  739. Py_DECREF(cursor);
  740. return NULL;
  741. }
  742. // execute the query which registers it
  743. if (Cursor_InternalExecute(cursor, 0) < 0) {
  744. Py_DECREF(cursor);
  745. return NULL;
  746. }
  747. Py_DECREF(cursor);
  748. Py_INCREF(Py_None);
  749. return Py_None;
  750. }
  751. //-----------------------------------------------------------------------------
  752. // Message_Free()
  753. // Free the memory associated with a message.
  754. //-----------------------------------------------------------------------------
  755. static void Message_Free(
  756. udt_Message *self) // object to free
  757. {
  758. Py_CLEAR(self->dbname);
  759. Py_CLEAR(self->tables);
  760. Py_TYPE(self)->tp_free((PyObject*) self);
  761. }
  762. //-----------------------------------------------------------------------------
  763. // MessageTable_Free()
  764. // Free the memory associated with a table in a message.
  765. //-----------------------------------------------------------------------------
  766. static void MessageTable_Free(
  767. udt_MessageTable *self) // object to free
  768. {
  769. Py_CLEAR(self->name);
  770. Py_TYPE(self)->tp_free((PyObject*) self);
  771. }
  772. //-----------------------------------------------------------------------------
  773. // MessageRow_Free()
  774. // Free the memory associated with a row in a message.
  775. //-----------------------------------------------------------------------------
  776. static void MessageRow_Free(
  777. udt_MessageRow *self) // object to free
  778. {
  779. Py_CLEAR(self->rowid);
  780. Py_TYPE(self)->tp_free((PyObject*) self);
  781. }