开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
ndb_async2.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：22k
源码类别：

MySQL数据库

开发平台：

Visual C++

ndb_async2.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. //#define DEBUG_ON
  19. #include <string.h>
  20. #include "userInterface.h"
  22. #include "macros.h"
  23. #include "ndb_schema.hpp"
  24. #include "ndb_error.hpp"
  25. #include <NdbSleep.h>
  27. #include <NdbApi.hpp>
  29. void T1_Callback(int result, NdbConnection * pCon, void * threadData);
  30. void T2_Callback(int result, NdbConnection * pCon, void * threadData);
  31. void T3_Callback_1(int result, NdbConnection * pCon, void * threadData);
  32. void T3_Callback_2(int result, NdbConnection * pCon, void * threadData);
  33. void T3_Callback_3(int result, NdbConnection * pCon, void * threadData);
  34. void T4_Callback_1(int result, NdbConnection * pCon, void * threadData);
  35. void T4_Callback_2(int result, NdbConnection * pCon, void * threadData);
  36. void T4_Callback_3(int result, NdbConnection * pCon, void * threadData);
  37. void T5_Callback_1(int result, NdbConnection * pCon, void * threadData);
  38. void T5_Callback_2(int result, NdbConnection * pCon, void * threadData);
  39. void T5_Callback_3(int result, NdbConnection * pCon, void * threadData);
  41. static int stat_async = 0;
  43. /**
  44.  * Transaction 1 - T1 
  45.  *
  46.  * Update location and changed by/time on a subscriber
  47.  *
  48.  * Input: 
  49.  *   SubscriberNumber,
  50.  *   Location,
  51.  *   ChangedBy,
  52.  *   ChangedTime
  53.  *
  54.  * Output:
  55.  */
  57. #define SFX_START (SUBSCRIBER_NUMBER_LENGTH - SUBSCRIBER_NUMBER_SUFFIX_LENGTH)
  59. inline
  60. NdbConnection *
  61. startTransaction(Ndb * pNDB, ThreadData * td){
  62.   return pNDB->startTransaction();
  63. #ifdef OLD_CODE
  64.   return pNDB->startTransactionDGroup (0, 
  65.        &td->transactionData.number[SFX_START],
  66.        1);
  67. #endif
  68. }
  70. void
  71. start_T1(Ndb * pNDB, ThreadData * td, int async){
  73.   DEBUG2("T1(%.*s): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  74.  td->transactionData.number); 
  76.   NdbConnection * pCON = 0;
  77.   while((pCON = startTransaction(pNDB, td)) == 0){
  78.     CHECK_ALLOWED_ERROR("T1: startTransaction", td, pNDB->getNdbError());
  79.     NdbSleep_MilliSleep(10);
  80.   }
  82.   NdbOperation *MyOp = pCON->getNdbOperation(SUBSCRIBER_TABLE);
  83.   if (MyOp != NULL) {  
  84.     MyOp->updateTuple();  
  85.     MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  86. td->transactionData.number);
  87.     MyOp->setValue(IND_SUBSCRIBER_LOCATION, 
  88.    (char *)&td->transactionData.location);
  89.     MyOp->setValue(IND_SUBSCRIBER_CHANGED_BY, 
  90.    td->transactionData.changed_by);
  91.     MyOp->setValue(IND_SUBSCRIBER_CHANGED_TIME, 
  92.    td->transactionData.changed_time);
  93.     if (async == 1) {
  94.       pCON->executeAsynchPrepare( Commit , T1_Callback, td);
  95.     } else {
  96.       int result = pCON->execute(Commit);
  97.       T1_Callback(result, pCON, (void*)td);
  98.       return;
  99.     }//if
  100.   } else {
  101.     CHECK_NULL(MyOp, "T1: getNdbOperation", td, pCON->getNdbError());
  102.   }//if
  103. }
  105. void
  106. T1_Callback(int result, NdbConnection * pCON, void * threadData) {
  107.   ThreadData * td = (ThreadData *)threadData;
  108.   
  109.   DEBUG2("T1(%.*s): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  110.  td->transactionData.number); 
  112.   if (result == -1) {
  113.     CHECK_ALLOWED_ERROR("T1: Commit", td, pCON->getNdbError());
  114.     td->pNDB->closeTransaction(pCON);
  115.     start_T1(td->pNDB, td, stat_async);
  116.     return;
  117.   }//if
  118.   td->pNDB->closeTransaction(pCON);
  119.   complete_T1(td);
  120. }
  122. /**
  123.  * Transaction 2 - T2
  124.  *
  125.  * Read from Subscriber:
  126.  *
  127.  * Input: 
  128.  *   SubscriberNumber
  129.  *
  130.  * Output:
  131.  *   Location
  132.  *   Changed by
  133.  *   Changed Timestamp
  134.  *   Name
  135.  */
  136. void
  137. start_T2(Ndb * pNDB, ThreadData * td, int async){
  139.   DEBUG3("T2(%.*s, %d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  140.  td->transactionData.number, 
  141.  td->transactionData.location);
  142.   
  143.   NdbConnection * pCON = 0;
  144.   
  145.   while((pCON = startTransaction(pNDB, td)) == 0){
  146.     CHECK_ALLOWED_ERROR("T2-1: startTransaction", td, pNDB->getNdbError());
  147.     NdbSleep_MilliSleep(10);
  148.   }
  150.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  151.   CHECK_NULL(MyOp, "T2: getNdbOperation", td,
  152.      pCON->getNdbError());
  153.   
  154.   MyOp->readTuple();
  155.   MyOp->equal(IND_SUBSCRIBER_NUMBER,
  156.       td->transactionData.number);
  157.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  158.  (char *)&td->transactionData.location);
  159.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  160.  td->transactionData.changed_by);
  161.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  162.  td->transactionData.changed_time);
  163.   MyOp->getValue(IND_SUBSCRIBER_NAME, 
  164.  td->transactionData.name);
  165.   if (async == 1) {
  166.     pCON->executeAsynchPrepare( Commit , T2_Callback, td);
  167.   } else {
  168.     int result = pCON->execute(Commit);
  169.     T2_Callback(result, pCON, (void*)td);
  170.     return;
  171.   }//if
  172. }
  174. void
  175. T2_Callback(int result, NdbConnection * pCON, void * threadData){
  176.   ThreadData * td = (ThreadData *)threadData;
  177.   DEBUG3("T2(%.*s, %d): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  178.  td->transactionData.number, 
  179.  td->transactionData.location);
  180.   
  181.   if (result == -1) {
  182.     CHECK_ALLOWED_ERROR("T2: Commit", td, pCON->getNdbError());
  183.     td->pNDB->closeTransaction(pCON);
  184.     start_T2(td->pNDB, td, stat_async);
  185.     return;
  186.   }//if
  187.   td->pNDB->closeTransaction(pCON);
  188.   complete_T2(td);
  189. }
  191. /**
  192.  * Transaction 3 - T3
  193.  *
  194.  * Read session details
  195.  *
  196.  * Input:
  197.  *   SubscriberNumber
  198.  *   ServerId
  199.  *   ServerBit
  200.  *
  201.  * Output:
  202.  *   BranchExecuted
  203.  *   SessionDetails
  204.  *   ChangedBy
  205.  *   ChangedTime
  206.  *   Location
  207.  */
  208. void
  209. start_T3(Ndb * pNDB, ThreadData * td, int async){
  211.   DEBUG3("T3(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  212.  td->transactionData.number, 
  213.  td->transactionData.server_id);
  214.   
  215.   NdbConnection * pCON = 0;
  217.   while((pCON = startTransaction(pNDB, td)) == 0){
  218.     CHECK_ALLOWED_ERROR("T3-1: startTransaction", td, pNDB->getNdbError());
  219.     NdbSleep_MilliSleep(10);
  220.   }
  221.   
  222.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  223.   CHECK_NULL(MyOp, "T3-1: getNdbOperation", td,
  224.      pCON->getNdbError());
  225.   
  226.   MyOp->readTuple();
  227.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  228.       td->transactionData.number);
  229.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  230.  (char *)&td->transactionData.location);
  231.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  232.  td->transactionData.changed_by);
  233.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  234.  td->transactionData.changed_time);
  235.   MyOp->getValue(IND_SUBSCRIBER_GROUP, 
  236.  (char *)&td->transactionData.group_id);
  237.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS, 
  238.  (char *)&td->transactionData.sessions);
  239.   stat_async = async;
  240.   if (async == 1) {
  241.     pCON->executeAsynchPrepare( NoCommit , T3_Callback_1, td);
  242.   } else {
  243.     int result = pCON->execute( NoCommit );
  244.     T3_Callback_1(result, pCON, (void*)td);
  245.     return;
  246.   }//if
  247. }
  249. void
  250. T3_Callback_1(int result, NdbConnection * pCON, void * threadData){
  251.   ThreadData * td = (ThreadData *)threadData;
  252.   DEBUG3("T3(%.*s, %.2d): - Callback 1", SUBSCRIBER_NUMBER_LENGTH, 
  253.  td->transactionData.number, 
  254.  td->transactionData.server_id);
  256.   if (result == -1) {
  257.     CHECK_ALLOWED_ERROR("T3-1: execute", td, pCON->getNdbError());
  258.     td->pNDB->closeTransaction(pCON);
  259.     start_T3(td->pNDB, td, stat_async);
  260.     return;
  261.   }//if
  263.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  264.   CHECK_NULL(MyOp, "T3-2: getNdbOperation", td,
  265.      pCON->getNdbError());
  266.     
  267.   MyOp->readTuple();
  268.   MyOp->equal(IND_GROUP_ID,
  269.       (char*)&td->transactionData.group_id);
  270.   MyOp->getValue(IND_GROUP_ALLOW_READ, 
  271.  (char *)&td->transactionData.permission);
  272.   if (stat_async == 1) {
  273.     pCON->executeAsynchPrepare( NoCommit , T3_Callback_2, td);
  274.   } else {
  275.     int result = pCON->execute( NoCommit );
  276.     T3_Callback_2(result, pCON, (void*)td);
  277.     return;
  278.   }//if
  279. }
  281. void
  282. T3_Callback_2(int result, NdbConnection * pCON, void * threadData){
  283.   ThreadData * td = (ThreadData *)threadData;
  284.   
  285.   if (result == -1) {
  286.     CHECK_ALLOWED_ERROR("T3-2: execute", td, pCON->getNdbError());
  287.     td->pNDB->closeTransaction(pCON);
  288.     start_T3(td->pNDB, td, stat_async);
  289.     return;
  290.   }//if
  291.   
  292.   Uint32 permission = td->transactionData.permission;
  293.   Uint32 sessions   = td->transactionData.sessions;
  294.   Uint32 server_bit = td->transactionData.server_bit;
  296.   if(((permission & server_bit) == server_bit) &&
  297.      ((sessions   & server_bit) == server_bit)){
  298.     
  299.     memcpy(td->transactionData.suffix,
  300.    &td->transactionData.number[SFX_START],
  301.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  302.     DEBUG5("T3(%.*s, %.2d): - Callback 2 - reading(%.*s)", 
  303.    SUBSCRIBER_NUMBER_LENGTH, 
  304.    td->transactionData.number, 
  305.    td->transactionData.server_id,
  306.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  307.    td->transactionData.suffix);
  308.     
  309.     /* Operation 3 */
  310.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  311.     CHECK_NULL(MyOp, "T3-3: getNdbOperation", td,
  312.        pCON->getNdbError());
  313.     
  314.     MyOp->simpleRead();
  315.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  316. (char*)td->transactionData.number);
  317.     MyOp->equal(IND_SESSION_SERVER,
  318. (char*)&td->transactionData.server_id);
  319.     MyOp->getValue(IND_SESSION_DATA, 
  320.    (char *)td->transactionData.session_details);
  321.     
  322.     /* Operation 4 */
  323.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  324.     CHECK_NULL(MyOp, "T3-4: getNdbOperation", td,
  325.        pCON->getNdbError());
  326.     
  327.     MyOp->interpretedUpdateTuple();
  328.     MyOp->equal(IND_SERVER_ID,
  329. (char*)&td->transactionData.server_id);
  330.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  331. (char*)td->transactionData.suffix);
  332.     MyOp->incValue(IND_SERVER_READS, (uint32)1);
  333.     td->transactionData.branchExecuted = 1;
  334.   } else {
  335.     DEBUG3("T3(%.*s, %.2d): - Callback 2 - no read",
  336.    SUBSCRIBER_NUMBER_LENGTH, 
  337.    td->transactionData.number, 
  338.    td->transactionData.server_id);
  339.     td->transactionData.branchExecuted = 0;
  340.   }
  341.   if (stat_async == 1) {
  342.     pCON->executeAsynchPrepare( Commit , T3_Callback_3, td);
  343.   } else {
  344.     int result = pCON->execute( Commit );
  345.     T3_Callback_3(result, pCON, (void*)td);
  346.     return;
  347.   }//if
  348. }
  350. void
  351. T3_Callback_3(int result, NdbConnection * pCON, void * threadData){
  352.   ThreadData * td = (ThreadData *)threadData;  
  353.   DEBUG3("T3(%.*s, %.2d): - Completing", SUBSCRIBER_NUMBER_LENGTH, 
  354.  td->transactionData.number, 
  355.  td->transactionData.server_id);
  356.   
  357.   if (result == -1) {
  358.     CHECK_ALLOWED_ERROR("T3-3: Commit", td, pCON->getNdbError());
  359.     td->pNDB->closeTransaction(pCON);
  360.     start_T3(td->pNDB, td, stat_async);
  361.     return;
  362.   }//if
  363.   td->pNDB->closeTransaction(pCON);
  364.   complete_T3(td);
  365. }
  367. /**
  368.  * Transaction 4 - T4
  369.  * 
  370.  * Create session
  371.  *
  372.  * Input:
  373.  *   SubscriberNumber
  374.  *   ServerId
  375.  *   ServerBit
  376.  *   SessionDetails,
  377.  *   DoRollback
  378.  * Output:
  379.  *   ChangedBy
  380.  *   ChangedTime
  381.  *   Location
  382.  *   BranchExecuted
  383.  */
  384. void
  385. start_T4(Ndb * pNDB, ThreadData * td, int async){
  387.   DEBUG3("T4(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  388.  td->transactionData.number, 
  389.  td->transactionData.server_id);
  390.   
  391.   NdbConnection * pCON = 0;
  392.   while((pCON = startTransaction(pNDB, td)) == 0){
  393.     CHECK_ALLOWED_ERROR("T4-1: startTransaction", td, pNDB->getNdbError());
  394.     NdbSleep_MilliSleep(10);
  395.   }
  396.   
  397.   NdbOperation *MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  398.   CHECK_NULL(MyOp, "T4-1: getNdbOperation", td,
  399.      pCON->getNdbError());
  400.   
  401.   MyOp->interpretedUpdateTuple();
  402.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  403.       td->transactionData.number);
  404.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  405.  (char *)&td->transactionData.location);
  406.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  407.  td->transactionData.changed_by);
  408.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  409.  td->transactionData.changed_time);
  410.   MyOp->getValue(IND_SUBSCRIBER_GROUP,
  411.  (char *)&td->transactionData.group_id);
  412.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
  413.  (char *)&td->transactionData.sessions); 
  414.   MyOp->incValue(IND_SUBSCRIBER_SESSIONS, 
  415.  (uint32)td->transactionData.server_bit);
  416.   stat_async = async;
  417.   if (async == 1) {
  418.     pCON->executeAsynchPrepare( NoCommit , T4_Callback_1, td);
  419.   } else {
  420.     int result = pCON->execute( NoCommit );
  421.     T4_Callback_1(result, pCON, (void*)td);
  422.     return;
  423.   }//if
  424. }
  426. void
  427. T4_Callback_1(int result, NdbConnection * pCON, void * threadData){
  428.   ThreadData * td = (ThreadData *)threadData;  
  429.   if (result == -1) {
  430.     CHECK_ALLOWED_ERROR("T4-1: execute", td, pCON->getNdbError());
  431.     td->pNDB->closeTransaction(pCON);
  432.     start_T4(td->pNDB, td, stat_async);
  433.     return;
  434.   }//if
  435.   
  436.   DEBUG3("T4(%.*s, %.2d): - Callback 1", 
  437.  SUBSCRIBER_NUMBER_LENGTH, 
  438.  td->transactionData.number, 
  439.  td->transactionData.server_id);
  442.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  443.   CHECK_NULL(MyOp, "T4-2: getNdbOperation", td,
  444.      pCON->getNdbError());
  445.   
  446.   MyOp->readTuple();
  447.   MyOp->equal(IND_GROUP_ID,
  448.       (char*)&td->transactionData.group_id);
  449.   MyOp->getValue(IND_GROUP_ALLOW_INSERT, 
  450.  (char *)&td->transactionData.permission);
  451.   if (stat_async == 1) {
  452.     pCON->executeAsynchPrepare( NoCommit , T4_Callback_2, td);
  453.   } else {
  454.     int result = pCON->execute( NoCommit );
  455.     T4_Callback_2(result, pCON, (void*)td);
  456.     return;
  457.   }//if
  458. }
  460. void
  461. T4_Callback_2(int result, NdbConnection * pCON, void * threadData){
  462.   ThreadData * td = (ThreadData *)threadData;  
  463.   if (result == -1) {
  464.     CHECK_ALLOWED_ERROR("T4-2: execute", td, pCON->getNdbError());
  465.     td->pNDB->closeTransaction(pCON);
  466.     start_T4(td->pNDB, td, stat_async);
  467.     return;
  468.   }//if
  470.   Uint32 permission = td->transactionData.permission;
  471.   Uint32 sessions   = td->transactionData.sessions;
  472.   Uint32 server_bit = td->transactionData.server_bit;
  473.   
  474.   if(((permission & server_bit) == server_bit) &&
  475.      ((sessions   & server_bit) == 0)){
  476.     
  477.     memcpy(td->transactionData.suffix,
  478.    &td->transactionData.number[SFX_START],
  479.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  480.     
  481.     DEBUG5("T4(%.*s, %.2d): - Callback 2 - inserting(%.*s)", 
  482.    SUBSCRIBER_NUMBER_LENGTH, 
  483.    td->transactionData.number, 
  484.    td->transactionData.server_id,
  485.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  486.    td->transactionData.suffix);
  487.     
  488.     /* Operation 3 */
  489.     
  490.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  491.     CHECK_NULL(MyOp, "T4-3: getNdbOperation", td,
  492.        pCON->getNdbError());
  493.     
  494.     MyOp->insertTuple();
  495.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  496. (char*)td->transactionData.number);
  497.     MyOp->equal(IND_SESSION_SERVER,
  498. (char*)&td->transactionData.server_id);
  499.     MyOp->setValue(SESSION_DATA, 
  500.    (char *)td->transactionData.session_details);
  501.     /* Operation 4 */
  502.     
  503.     /* Operation 5 */
  504.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  505.     CHECK_NULL(MyOp, "T4-5: getNdbOperation", td,
  506.        pCON->getNdbError());
  507.     
  508.     MyOp->interpretedUpdateTuple();
  509.     MyOp->equal(IND_SERVER_ID,
  510. (char*)&td->transactionData.server_id);
  511.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  512. (char*)td->transactionData.suffix);
  513.     MyOp->incValue(IND_SERVER_INSERTS, (uint32)1);
  514.     td->transactionData.branchExecuted = 1;
  515.   } else {
  516.     td->transactionData.branchExecuted = 0;
  517.     DEBUG5("T4(%.*s, %.2d): - Callback 2 - %s %s",
  518.    SUBSCRIBER_NUMBER_LENGTH, 
  519.    td->transactionData.number, 
  520.    td->transactionData.server_id,
  521.    ((permission & server_bit) ? 
  522.     "permission - " : "no permission - "),
  523.    ((sessions   & server_bit) ? 
  524.     "in session - " : "no in session - "));
  525.   }
  526.   
  527.   if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
  528.     if (stat_async == 1) {
  529.       pCON->executeAsynchPrepare( Commit , T4_Callback_3, td);
  530.     } else {
  531.       int result = pCON->execute( Commit );
  532.       T4_Callback_3(result, pCON, (void*)td);
  533.       return;
  534.     }//if
  535.   } else {
  536.     if (stat_async == 1) {
  537.       pCON->executeAsynchPrepare( Rollback , T4_Callback_3, td);
  538.     } else {
  539.       int result = pCON->execute( Rollback );
  540.       T4_Callback_3(result, pCON, (void*)td);
  541.       return;
  542.     }//if
  543.   }
  544. }
  546. void
  547. T4_Callback_3(int result, NdbConnection * pCON, void * threadData){
  548.   ThreadData * td = (ThreadData *)threadData;  
  549.   if (result == -1) {
  550.     CHECK_ALLOWED_ERROR("T4-3: Commit", td, pCON->getNdbError());
  551.     td->pNDB->closeTransaction(pCON);
  552.     start_T4(td->pNDB, td, stat_async);
  553.     return;
  554.   }//if
  555.   
  556.   DEBUG3("T4(%.*s, %.2d): - Completing", 
  557.  SUBSCRIBER_NUMBER_LENGTH, 
  558.  td->transactionData.number, 
  559.  td->transactionData.server_id);
  561.   td->pNDB->closeTransaction(pCON);
  562.   complete_T4(td);
  563. }
  565. /**
  566.  * Transaction 5 - T5
  567.  * 
  568.  * Delete session
  569.  *
  570.  * Input:
  571.  *   SubscriberNumber
  572.  *   ServerId
  573.  *   ServerBit
  574.  *   DoRollback
  575.  * Output:
  576.  *   ChangedBy
  577.  *   ChangedTime
  578.  *   Location
  579.  *   BranchExecuted
  580.  */
  581. void
  582. start_T5(Ndb * pNDB, ThreadData * td, int async){
  584.   DEBUG3("T5(%.*s, %.2d): - Starting", SUBSCRIBER_NUMBER_LENGTH, 
  585.  td->transactionData.number, 
  586.  td->transactionData.server_id);
  588.   NdbConnection * pCON = 0;
  589.   while((pCON = startTransaction(pNDB, td)) == 0){
  590.     CHECK_ALLOWED_ERROR("T5-1: startTransaction", td, pNDB->getNdbError());
  591.     NdbSleep_MilliSleep(10);
  592.   }
  593.   
  594.   NdbOperation * MyOp= pCON->getNdbOperation(SUBSCRIBER_TABLE);
  595.   CHECK_NULL(MyOp, "T5-1: getNdbOperation", td,
  596.      pCON->getNdbError());
  597.   
  598.   MyOp->interpretedUpdateTuple();
  599.   MyOp->equal(IND_SUBSCRIBER_NUMBER, 
  600.       td->transactionData.number);
  601.   MyOp->getValue(IND_SUBSCRIBER_LOCATION, 
  602.  (char *)&td->transactionData.location);
  603.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_BY, 
  604.  td->transactionData.changed_by);
  605.   MyOp->getValue(IND_SUBSCRIBER_CHANGED_TIME, 
  606.  td->transactionData.changed_time);
  607.   MyOp->getValue(IND_SUBSCRIBER_GROUP,
  608.  (char *)&td->transactionData.group_id);
  609.   MyOp->getValue(IND_SUBSCRIBER_SESSIONS,
  610.  (char *)&td->transactionData.sessions);
  611.   MyOp->subValue(IND_SUBSCRIBER_SESSIONS, 
  612.  (uint32)td->transactionData.server_bit);
  613.   stat_async = async;
  614.   if (async == 1) {
  615.     pCON->executeAsynchPrepare( NoCommit , T5_Callback_1, td);
  616.   } else {
  617.     int result = pCON->execute( NoCommit );
  618.     T5_Callback_1(result, pCON, (void*)td);
  619.     return;
  620.   }//if
  621. }
  623. void
  624. T5_Callback_1(int result, NdbConnection * pCON, void * threadData){
  625.   ThreadData * td = (ThreadData *)threadData;  
  626.   if (result == -1) {
  627.     CHECK_ALLOWED_ERROR("T5-1: execute", td, pCON->getNdbError());
  628.     td->pNDB->closeTransaction(pCON);
  629.     start_T5(td->pNDB, td, stat_async);
  630.     return;
  631.   }//if
  633.   DEBUG3("T5(%.*s, %.2d): - Callback 1", 
  634.  SUBSCRIBER_NUMBER_LENGTH, 
  635.  td->transactionData.number, 
  636.  td->transactionData.server_id);
  637.   
  638.   NdbOperation * MyOp = pCON->getNdbOperation(GROUP_TABLE);
  639.   CHECK_NULL(MyOp, "T5-2: getNdbOperation", td,
  640.      pCON->getNdbError());
  641.   
  642.   MyOp->readTuple();
  643.   MyOp->equal(IND_GROUP_ID,
  644.       (char*)&td->transactionData.group_id);
  645.   MyOp->getValue(IND_GROUP_ALLOW_DELETE, 
  646.  (char *)&td->transactionData.permission);
  647.   if (stat_async == 1) {
  648.     pCON->executeAsynchPrepare( NoCommit , T5_Callback_2, td);
  649.   } else {
  650.     int result = pCON->execute( NoCommit );
  651.     T5_Callback_2(result, pCON, (void*)td);
  652.     return;
  653.   }//if
  654. }
  656. void
  657. T5_Callback_2(int result, NdbConnection * pCON, void * threadData){
  658.   ThreadData * td = (ThreadData *)threadData;  
  659.   if (result == -1) {
  660.     CHECK_ALLOWED_ERROR("T5-2: execute", td, pCON->getNdbError());
  661.     td->pNDB->closeTransaction(pCON);
  662.     start_T5(td->pNDB, td, stat_async);
  663.     return;
  664.   }//if
  666.   Uint32 permission = td->transactionData.permission;
  667.   Uint32 sessions   = td->transactionData.sessions;
  668.   Uint32 server_bit = td->transactionData.server_bit;
  670.   if(((permission & server_bit) == server_bit) &&
  671.      ((sessions   & server_bit) == server_bit)){
  672.     
  673.     memcpy(td->transactionData.suffix,
  674.    &td->transactionData.number[SFX_START],
  675.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH);
  676.     
  677.     DEBUG5("T5(%.*s, %.2d): - Callback 2 - deleting(%.*s)", 
  678.    SUBSCRIBER_NUMBER_LENGTH, 
  679.    td->transactionData.number, 
  680.    td->transactionData.server_id,
  681.    SUBSCRIBER_NUMBER_SUFFIX_LENGTH, 
  682.    td->transactionData.suffix);
  683.     
  684.     /* Operation 3 */
  685.     NdbOperation * MyOp = pCON->getNdbOperation(SESSION_TABLE);
  686.     CHECK_NULL(MyOp, "T5-3: getNdbOperation", td,
  687.        pCON->getNdbError());
  688.     
  689.     MyOp->deleteTuple();
  690.     MyOp->equal(IND_SESSION_SUBSCRIBER,
  691. (char*)td->transactionData.number);
  692.     MyOp->equal(IND_SESSION_SERVER,
  693. (char*)&td->transactionData.server_id);
  694.     /* Operation 4 */
  695.     
  696.     /* Operation 5 */
  697.     MyOp = pCON->getNdbOperation(SERVER_TABLE);
  698.     CHECK_NULL(MyOp, "T5-5: getNdbOperation", td,
  699.        pCON->getNdbError());
  700.     
  701.     MyOp->interpretedUpdateTuple();
  702.     MyOp->equal(IND_SERVER_ID,
  703. (char*)&td->transactionData.server_id);
  704.     MyOp->equal(IND_SERVER_SUBSCRIBER_SUFFIX,
  705. (char*)td->transactionData.suffix);
  706.     MyOp->incValue(IND_SERVER_DELETES, (uint32)1);
  707.     td->transactionData.branchExecuted = 1;
  708.   } else {
  709.     td->transactionData.branchExecuted = 0;
  711.     DEBUG5("T5(%.*s, %.2d): - Callback 2 - no delete - %s %s", 
  712.    SUBSCRIBER_NUMBER_LENGTH, 
  713.    td->transactionData.number, 
  714.    td->transactionData.server_id,
  715.    ((permission & server_bit) ? 
  716.     "permission - " : "no permission - "),
  717.    ((sessions   & server_bit) ? 
  718.     "in session - " : "no in session - "));
  719.   }
  720.   
  721.   if(!td->transactionData.do_rollback && td->transactionData.branchExecuted){
  722.     if (stat_async == 1) {
  723.       pCON->executeAsynchPrepare( Commit , T5_Callback_3, td);
  724.     } else {
  725.       int result = pCON->execute( Commit );
  726.       T5_Callback_3(result, pCON, (void*)td);
  727.       return;
  728.     }//if
  729.   } else {
  730.     if (stat_async == 1) {
  731.       pCON->executeAsynchPrepare( Rollback , T5_Callback_3, td);
  732.     } else {
  733.       int result = pCON->execute( Rollback );
  734.       T5_Callback_3(result, pCON, (void*)td);
  735.       return;
  736.     }//if
  737.   }
  738. }
  740. void
  741. T5_Callback_3(int result, NdbConnection * pCON, void * threadData){
  742.   ThreadData * td = (ThreadData *)threadData;  
  743.   if (result == -1) {
  744.     CHECK_ALLOWED_ERROR("T5-3: Commit", td, pCON->getNdbError());
  745.     td->pNDB->closeTransaction(pCON);
  746.     start_T5(td->pNDB, td, stat_async);
  747.     return;
  748.   }//if
  749.   
  750.   DEBUG3("T5(%.*s, %.2d): - Completing", 
  751.  SUBSCRIBER_NUMBER_LENGTH, 
  752.  td->transactionData.number, 
  753.  td->transactionData.server_id);
  754.   
  755.   td->pNDB->closeTransaction(pCON);
  756.   complete_T5(td);
  757. }