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

源码开发语言/平台
当前位置：首页 > 源码/资料 > 中间件编程 > 查看源码
XATMIc.cxx
资源名称：blacktie-2.0.0.M3-src.zip　[点击查看]
上传用户：xfwatch
上传日期：2020-12-14
资源大小：872k
文件大小：29k
源码类别：

中间件编程

开发平台：

Java

XATMIc.cxx：源码内容
  1. /*
  2.  * JBoss, Home of Professional Open Source
  3.  * Copyright 2008, Red Hat, Inc., and others contributors as indicated
  4.  * by the @authors tag. All rights reserved.
  5.  * See the copyright.txt in the distribution for a
  6.  * full listing of individual contributors.
  7.  * This copyrighted material is made available to anyone wishing to use,
  8.  * modify, copy, or redistribute it subject to the terms and conditions
  9.  * of the GNU Lesser General Public License, v. 2.1.
  10.  * This program is distributed in the hope that it will be useful, but WITHOUT A
  11.  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
  12.  * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
  13.  * You should have received a copy of the GNU Lesser General Public License,
  14.  * v.2.1 along with this distribution; if not, write to the Free Software
  15.  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  16.  * MA  02110-1301, USA.
  17.  */
  19. #include <stdio.h>
  20. #include <stdlib.h>
  21. #include <string.h>
  22. #include <stdarg.h>
  23. #include <iostream>
  25. #include "log4cxx/logger.h"
  27. #include "ThreadLocalStorage.h"
  28. #include "txx.h"
  29. #include "xatmi.h"
  30. #include "Session.h"
  31. #include "AtmiBrokerClientControl.h"
  32. #include "AtmiBrokerServerControl.h"
  33. #include "AtmiBrokerClient.h"
  34. #include "AtmiBrokerServer.h"
  35. #include "AtmiBrokerMem.h"
  36. #include "AtmiBrokerEnv.h"
  37. #include "ServiceDispatcher.h"
  39. #include "txx.h"
  41. long DISCON = 0x00000003;
  42. bool warnedTPNOBLOCK = false;
  43. bool warnedTPGETANY = false;
  45. // Logger for XATMIc
  46. log4cxx::LoggerPtr loggerXATMI(log4cxx::Logger::getLogger("XATMIc"));
  48. int bufferSize(char* data, int suggestedSize) {
  49. if (data == NULL) {
  50. return 0;
  51. }
  52. int data_size = ::tptypes(data, NULL, NULL);
  53. if (data_size >= 0) {
  54. if (suggestedSize <= 0 || suggestedSize > data_size) {
  55. return data_size;
  56. } else {
  57. return suggestedSize;
  58. }
  59. } else {
  60. LOG4CXX_DEBUG(loggerXATMI,
  61. (char*) "A NON-BUFFER WAS ATTEMPTED TO BE SENT");
  62. setSpecific(TPE_KEY, TSS_TPEINVAL);
  63. return -1;
  64. }
  66. }
  68. void setTpurcode(long rcode) {
  69. char* retrieved = (char*) getSpecific(TPR_KEY);
  70. if (retrieved != NULL) {
  71. destroySpecific( TPR_KEY);
  72. free(retrieved);
  73. }
  74. if (rcode > 0) {
  75. char* toStore = (char*) malloc(8 * sizeof(long));
  76. sprintf(toStore, "%ld", rcode);
  77. setSpecific(TPR_KEY, toStore);
  78. }
  79. }
  81. int send(Session* session, const char* replyTo, char* idata, long ilen,
  82. int correlationId, long flags, long rval, long rcode) {
  83. LOG4CXX_DEBUG(loggerXATMI, (char*) "send - ilen: " << ilen << ": "
  84. << "cd: " << correlationId << "flags: " << flags);
  85. if (flags & TPNOBLOCK && !warnedTPNOBLOCK) {
  86. LOG4CXX_ERROR(loggerXATMI, (char*) "TPNOBLOCK NOT SUPPORTED FOR SENDS");
  87. warnedTPNOBLOCK = true;
  88. }
  89. int toReturn = -1;
  91. if (session->getCanSend() || rval == DISCON) {
  92. try {
  93. LOG4CXX_TRACE(loggerXATMI, (char*) "allocating data to go: "
  94. << ilen);
  96. MESSAGE message;
  97. message.replyto = replyTo;
  98. message.data = idata;
  99. message.len = ilen;
  100. message.correlationId = correlationId;
  101. message.flags = flags;
  102. message.rcode = rcode;
  103. message.rval = rval;
  104. message.type = NULL;
  105. message.subtype = NULL;
  106. if (message.data != NULL) {
  107. message.type = (char*) malloc(MAX_TYPE_SIZE + 1);
  108. memset(message.type, '', MAX_TYPE_SIZE + 1);
  109. message.subtype = (char*) malloc(MAX_SUBTYPE_SIZE + 1);
  110. memset(message.subtype, '', MAX_SUBTYPE_SIZE + 1);
  111. ::tptypes(idata, message.type, message.subtype);
  112. }
  113. if (message.type == NULL) {
  114. message.type = (char*) "";
  115. }
  116. if (message.subtype == NULL) {
  117. message.subtype = (char*) "";
  118. }
  119. message.control = (TPNOTRAN & flags) ? NULL : txx_serialize(&(message.ttl));
  120. if (message.control == NULL)
  121. message.ttl = mqConfig.timeToLive * 1000;
  123. session->blockSignals((flags & TPSIGRSTRT));
  125. if (session->send(message))
  126. toReturn = 0;
  128. if (session->unblockSignals() != 0 && (flags & TPSIGRSTRT) == 0)
  129. toReturn = -1;
  131. if (message.control)
  132. free(message.control);
  134. if (message.data != NULL) {
  135. free(message.type);
  136. free(message.subtype);
  137. }
  138. } catch (...) {
  139. LOG4CXX_ERROR(loggerXATMI, (char*) "send: call failed");
  140. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  141. }
  142. } else {
  143. LOG4CXX_ERROR(loggerXATMI, (char*) "Session " << correlationId
  144. << "can't send");
  145. setSpecific(TPE_KEY, TSS_TPEPROTO);
  146. }
  148. return toReturn;
  149. }
  151. int receive(int id, Session* session, char ** odata, long *olen, long flags,
  152. long* event, bool closeSession) {
  153. LOG4CXX_DEBUG(loggerXATMI, (char*) "Receive invoked");
  154. setTpurcode(0);
  155. int toReturn = -1;
  156. int len = ::bufferSize(*odata, *olen);
  157. if (len != -1) {
  158. LOG4CXX_DEBUG(loggerXATMI, (char*) "receive session: "
  159. << session->getId() << " olen: " << olen << " flags: " << flags);
  160. if (flags & TPGETANY && !warnedTPGETANY) {
  161. LOG4CXX_ERROR(loggerXATMI,
  162. (char*) "TPGETANY NOT SUPPORTED FOR RECEIVES");
  163. warnedTPGETANY = true;
  164. }
  165. if (session->getCanRecv()) {
  166. // TODO Make configurable Default wait time is blocking (x1000 in SynchronizableObject)
  167. long time;
  168. if (TPNOBLOCK & flags) { // NB flags override any XATMI or transaction timeouts
  169. time = 1;
  170. LOG4CXX_DEBUG(loggerXATMI, (char*) "Setting timeout to 1");
  171. } else if (TPNOTIME & flags) {
  172. time = 0;
  173. LOG4CXX_DEBUG(loggerXATMI, (char*) "TPNOTIME = BLOCKING CALL");
  174. } else {
  175. switch (txx_ttl(&time)) {
  176. case -1: // No transaction so use XATMI timeouts
  177. time = (long) mqConfig.requestTimeout + (long) mqConfig.timeToLive;
  178. {LOG4CXX_TRACE(loggerXATMI, (char*) "receive txx_ttl returned -1 time=" << time)};
  179. break;
  180. case 1: // txn not subject to a timeout so block
  181. time = 0;
  182. {LOG4CXX_TRACE(loggerXATMI, (char*) "receive txx_ttl returned 1 time=0"<< time)};
  183. break;
  184. default: /*FALLTHRU txx_ttl will only returns -1, 0 or 1*/
  185. case 0: // time has already been updated
  186. {LOG4CXX_TRACE(loggerXATMI, (char*) "receive txx_ttl returned 0 time=" << time)};
  187. break;
  188. }
  189. }
  191. LOG4CXX_DEBUG(loggerXATMI, (char*) "Setting timeout to: " << time);
  192. try {
  193. session->blockSignals((flags & TPSIGRSTRT));
  195. MESSAGE message = session->receive(time);
  197. if (session->unblockSignals() != 0 && (flags & TPSIGRSTRT) == 0) {
  198. // signalled during the receive call TPSIGRSTRT wasn't requested
  199. txx_rollback_only();
  200. } else if (message.received) {
  201. if (message.rval == DISCON) {
  202. *event = TPEV_DISCONIMM;
  203. txx_rollback_only();
  204. } else {
  205. char* type = message.type;
  206. if (type == NULL) {
  207. type = (char*) "";
  208. }
  209. char* subtype = message.subtype;
  210. if (subtype == NULL) {
  211. subtype = (char*) "";
  212. }
  213. char* messageType = (char*) malloc(MAX_TYPE_SIZE);
  214. char* messageSubtype = (char*) malloc(MAX_SUBTYPE_SIZE);
  215. tptypes(*odata, messageType, messageSubtype);
  216. bool typesChanged = strncmp(type, messageType,
  217. MAX_TYPE_SIZE) != 0 || strncmp(subtype,
  218. messageSubtype, MAX_SUBTYPE_SIZE) != 0;
  219. free(messageType);
  220. free(messageSubtype);
  222. if (flags & TPNOCHANGE && typesChanged) {
  223. // TODO rollback-only
  224. setSpecific(TPE_KEY, TSS_TPEOTYPE);
  225. setTpurcode(message.rcode);
  226. txx_rollback_only();
  227. free(message.data);
  228. free(message.type);
  229. free(message.subtype);
  230. free((char*) message.replyto);
  231. if (closeSession) {
  232. ptrAtmiBrokerClient->closeSession(id);
  233. LOG4CXX_TRACE(loggerXATMI,
  234. (char*) "receive session closed: "
  235. << id);
  236. }
  237. return toReturn;
  238. }
  240. if (len < message.len && !typesChanged) {
  241. *odata = AtmiBrokerMem::get_instance()->tprealloc(
  242. *odata, message.len, NULL, NULL, true);
  243. } else if (len < message.len && typesChanged) {
  244. *odata = AtmiBrokerMem::get_instance()->tprealloc(
  245. *odata, message.len, message.type,
  246. message.subtype, true);
  247. } else if (typesChanged) {
  248. *odata = AtmiBrokerMem::get_instance()->tprealloc(
  249. *odata, message.len, message.type,
  250. message.subtype, true);
  251. }
  253. *olen = message.len;
  254. if (message.len > 0) {
  255. memcpy(*odata, (char*) message.data, *olen);
  256. } else if (message.data == NULL) {
  257. *odata = NULL;
  258. }
  259. free(message.data);
  260. free(message.type);
  261. free(message.subtype);
  262. free((char*) message.replyto);
  264. if (message.rcode == TPESVCERR) {
  265. *event = TPEV_SVCERR;
  266. setSpecific(TPE_KEY, TSS_TPESVCERR);
  267. txx_rollback_only();
  268. ptrAtmiBrokerClient->closeSession(id);
  269. closeSession = false;
  270. } else if (message.rval == TPFAIL) {
  271. setTpurcode(message.rcode);
  272. *event = TPEV_SVCFAIL;
  273. setSpecific(TPE_KEY, TSS_TPESVCFAIL);
  274. txx_rollback_only();
  275. ptrAtmiBrokerClient->closeSession(id);
  276. closeSession = false;
  277. } else if (message.rval == TPSUCCESS) {
  278. toReturn = 0;
  279. setTpurcode(message.rcode);
  280. *event = TPEV_SVCSUCC;
  281. ptrAtmiBrokerClient->closeSession(id);
  282. closeSession = false;
  283. } else if (message.flags & TPRECVONLY) {
  284. toReturn = 0;
  285. *event = TPEV_SENDONLY;
  286. session->setCanSend(true);
  287. session->setCanRecv(false);
  288. LOG4CXX_DEBUG(
  289. loggerXATMI,
  290. (char*) "receive TPRECVONLY set constraints session: "
  291. << session->getId() << " send: "
  292. << session->getCanSend()
  293. << " recv: "
  294. << session->getCanRecv());
  295. } else if (message.correlationId >= 0) {
  296. toReturn = 0;
  297. } else {
  298. LOG4CXX_ERROR(loggerXATMI,
  299. (char*) "COULD NOT PARSE RECEIVED MESSAGE");
  300. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  301. }
  302. }
  303. } else if (TPNOBLOCK & flags) {
  304. LOG4CXX_DEBUG(loggerXATMI,
  305. (char*) "Message not immediately available");
  306. } else {
  307. setSpecific(TPE_KEY, TSS_TPETIME);
  308. txx_rollback_only();
  309. }
  310. } catch (...) {
  311. LOG4CXX_ERROR(
  312. loggerXATMI,
  313. (char*) "Could not set the receive from the destination");
  314. }
  315. } else {
  316. LOG4CXX_DEBUG(loggerXATMI, (char*) "Session can't receive");
  317. setSpecific(TPE_KEY, TSS_TPEPROTO);
  318. }
  319. }
  321. if (closeSession) {
  322. ptrAtmiBrokerClient->closeSession(id);
  323. LOG4CXX_TRACE(loggerXATMI, (char*) "receive session closed: " << id);
  324. }
  326. return toReturn;
  327. }
  329. int _get_tperrno(void) {
  330. LOG4CXX_TRACE(loggerXATMI, (char*) "_get_tperrno");
  331. char* err = (char*) getSpecific(TPE_KEY);
  332. int toReturn = 0;
  333. if (err != NULL) {
  334. LOG4CXX_TRACE(loggerXATMI, (char*) "found _get_tperrno" << err);
  335. toReturn = atoi(err);
  336. }
  337. LOG4CXX_TRACE(loggerXATMI, (char*) "returning _get_tperrno" << toReturn);
  338. return toReturn;
  339. }
  341. long _get_tpurcode(void) {
  342. LOG4CXX_TRACE(loggerXATMI, (char*) "_get_tpurcode");
  343. char* rcode = (char*) getSpecific(TPR_KEY);
  344. long toReturn = 0;
  345. if (rcode != NULL) {
  346. LOG4CXX_TRACE(loggerXATMI, (char*) "found _get_tpurcode" << rcode);
  347. toReturn = atol(rcode);
  348. }
  349. LOG4CXX_TRACE(loggerXATMI, (char*) "returning _get_tpurcode" << toReturn);
  350. return toReturn;
  351. }
  353. int tpadvertise(char * svcname, void(*func)(TPSVCINFO *)) {
  354. LOG4CXX_TRACE(loggerXATMI, (char*) "tpadvertise: " << svcname);
  355. setSpecific(TPE_KEY, TSS_TPERESET);
  356. int toReturn = -1;
  357. if (ptrServer != NULL) {
  358. if (ptrServer->advertiseService(svcname, func)) {
  359. toReturn = 0;
  360. }
  361. } else {
  362. LOG4CXX_ERROR(loggerXATMI, (char*) "server not initialized");
  363. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  364. }
  365. LOG4CXX_TRACE(loggerXATMI, (char*) "tpadvertise return: " << toReturn
  366. << " tperrno: " << tperrno);
  367. return toReturn;
  368. }
  370. int tpunadvertise(char * svcname) {
  371. LOG4CXX_TRACE(loggerXATMI, (char*) "tpunadvertise: " << svcname);
  372. setSpecific(TPE_KEY, TSS_TPERESET);
  373. int toReturn = -1;
  374. if (ptrServer != NULL) {
  375. if (svcname && strcmp(svcname, "") != 0) {
  376. if (ptrServer->isAdvertised(svcname)) {
  377. ptrServer->unadvertiseService(svcname, false);
  378. toReturn = 0;
  379. } else {
  380. setSpecific(TPE_KEY, TSS_TPENOENT);
  381. }
  382. } else {
  383. setSpecific(TPE_KEY, TSS_TPEINVAL);
  384. }
  385. } else {
  386. LOG4CXX_ERROR(loggerXATMI, (char*) "server not initialized");
  387. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  388. }
  389. LOG4CXX_TRACE(loggerXATMI, (char*) "tpunadvertise return: " << toReturn
  390. << " tperrno: " << tperrno);
  391. return toReturn;
  392. }
  394. char* tpalloc(char* type, char* subtype, long size) {
  395. LOG4CXX_TRACE(loggerXATMI, (char*) "tpalloc type: " << type << " size: "
  396. << size);
  397. if (subtype) {
  398. LOG4CXX_TRACE(loggerXATMI, (char*) "tpalloc subtype: " << type);
  399. }
  400. setSpecific(TPE_KEY, TSS_TPERESET);
  401. char* toReturn = NULL;
  402. if (clientinit() != -1) {
  403. toReturn = AtmiBrokerMem::get_instance()->tpalloc(type, subtype, size,
  404. false);
  405. }
  406. LOG4CXX_TRACE(loggerXATMI, (char*) "tpalloc returning" << " tperrno: "
  407. << tperrno);
  408. return toReturn;
  409. }
  411. char* tprealloc(char * addr, long size) {
  412. LOG4CXX_TRACE(loggerXATMI, (char*) "tprealloc: " << size);
  413. setSpecific(TPE_KEY, TSS_TPERESET);
  414. char* toReturn = NULL;
  415. if (clientinit() != -1) {
  416. toReturn = AtmiBrokerMem::get_instance()->tprealloc(addr, size, NULL,
  417. NULL, false);
  418. }
  419. LOG4CXX_TRACE(loggerXATMI, (char*) "tprealloc returning" << " tperrno: "
  420. << tperrno);
  421. return toReturn;
  422. }
  424. void tpfree(char* ptr) {
  425. LOG4CXX_TRACE(loggerXATMI, (char*) "tpfree");
  426. setSpecific(TPE_KEY, TSS_TPERESET);
  427. if (clientinit() != -1) {
  428. AtmiBrokerMem::get_instance()->tpfree(ptr, false);
  429. }
  430. LOG4CXX_TRACE(loggerXATMI, (char*) "tpfree returning" << " tperrno: "
  431. << tperrno);
  432. }
  434. long tptypes(char* ptr, char* type, char* subtype) {
  435. LOG4CXX_TRACE(loggerXATMI, (char*) "tptypes called");
  436. setSpecific(TPE_KEY, TSS_TPERESET);
  437. LOG4CXX_TRACE(loggerXATMI, (char*) "set the specific");
  438. long toReturn = -1;
  439. if (clientinit() != -1) {
  440. toReturn = AtmiBrokerMem::get_instance()->tptypes(ptr, type, subtype);
  441. }
  442. LOG4CXX_TRACE(loggerXATMI, (char*) "tptypes return: " << toReturn
  443. << " tperrno: " << tperrno);
  444. return toReturn;
  445. }
  447. int tpcall(char * svc, char* idata, long ilen, char ** odata, long *olen,
  448. long flags) {
  449. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcall: " << svc << " ilen: " << ilen
  450. << " flags: " << flags);
  451. int toReturn = -1;
  452. setSpecific(TPE_KEY, TSS_TPERESET);
  454. long toCheck = flags & ~(TPNOTRAN | TPNOCHANGE | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  456. if (toCheck != 0) {
  457. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  458. setSpecific(TPE_KEY, TSS_TPEINVAL);
  459. } else {
  460. if (clientinit() != -1) {
  461. long tpacallFlags = flags;
  462. tpacallFlags &= ~TPNOCHANGE;
  463. int cd = tpacall(svc, idata, ilen, tpacallFlags);
  465. if (cd != -1) {
  466. long tpgetrplyFlags = flags;
  467. tpgetrplyFlags &= ~TPNOTRAN;
  468. tpgetrplyFlags &= ~TPNOBLOCK;
  469. toReturn = tpgetrply(&cd, odata, olen, tpgetrplyFlags);
  470. }
  471. }
  472. }
  473. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcall return: " << toReturn
  474. << " tperrno: " << tperrno);
  475. return toReturn;
  476. }
  478. int tpacall(char * svc, char* idata, long ilen, long flags) {
  479. LOG4CXX_TRACE(loggerXATMI, (char*) "tpacall: " << svc << " ilen: " << ilen
  480. << " flags: 0x" << std::hex << flags);
  481. int toReturn = -1;
  482. setSpecific(TPE_KEY, TSS_TPERESET);
  484. long toCheck = flags & ~(TPNOTRAN | TPNOREPLY | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  486. if (toCheck != 0) {
  487. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  488. setSpecific(TPE_KEY, TSS_TPEINVAL);
  489. } else {
  490. bool transactional = !(flags & TPNOTRAN);
  491. if (transactional) {
  492. void *ctrl = txx_get_control();
  493. if (ctrl == NULL) {
  494. transactional = false;
  495. }
  496. txx_release_control(ctrl);
  497. }
  499. if (transactional && (flags & TPNOREPLY)) {
  500. LOG4CXX_ERROR(
  501. loggerXATMI,
  502. (char*) "TPNOREPLY CALLED WITHOUT TPNOTRAN DURING TRANSACTION");
  503. setSpecific(TPE_KEY, TSS_TPEINVAL);
  504. } else {
  505. int len = ::bufferSize(idata, ilen);
  506. if (len != -1) {
  507. if (clientinit() != -1) {
  508. Session* session = NULL;
  509. int cd = -1;
  510. try {
  511. session = ptrAtmiBrokerClient->createSession(cd, svc);
  512. LOG4CXX_TRACE(loggerXATMI, (char*) "new session: "
  513. << session << " cd: " << cd
  514. << " transactional: " << transactional);
  516. if (cd != -1) {
  517. if (transactional)
  518. txx_suspend(cd, tpdiscon);
  520. toReturn = ::send(session, session->getReplyTo(),
  521. idata, len, cd, flags, 0, 0);
  523. if (toReturn >= 0) {
  524. if (TPNOREPLY & flags) {
  525. toReturn = 0;
  526. ptrAtmiBrokerClient->closeSession(cd);
  527. } else {
  528. toReturn = cd;
  529. }
  530. } else {
  531. LOG4CXX_DEBUG(loggerXATMI,
  532. (char*) "Session got dudded: " << cd);
  533. ptrAtmiBrokerClient->closeSession(cd);
  534. }
  535. } else if (tperrno == 0) {
  536. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  537. ptrAtmiBrokerClient->closeSession(cd);
  538. }
  539. } catch (...) {
  540. LOG4CXX_ERROR(
  541. loggerXATMI,
  542. (char*) "tpacall failed to connect to service queue");
  543. setSpecific(TPE_KEY, TSS_TPENOENT);
  544. if (cd != -1) {
  545. ptrAtmiBrokerClient->closeSession(cd);
  546. }
  547. }
  549. if (transactional && toReturn < 0) {
  550. // txx_suspend was called but there was an error so
  551. // resume (note we didn't check for TPNOREPLY since we are in
  552. // the else arm of if (transactional && (flags & TPNOREPLY))
  553. LOG4CXX_DEBUG(loggerXATMI, (char*) "tpacall resume cd="
  554. << cd << " rv=" << toReturn);
  555. txx_resume(cd);
  556. }
  557. }
  558. }
  559. }
  560. }
  561. LOG4CXX_TRACE(loggerXATMI, (char*) "tpacall return: " << toReturn
  562. << " tperrno: " << tperrno);
  563. return toReturn;
  564. }
  566. int tpconnect(char * svc, char* idata, long ilen, long flags) {
  567. LOG4CXX_TRACE(loggerXATMI, (char*) "tpconnect: " << svc << " ilen: "
  568. << ilen << " flags: " << flags);
  569. int toReturn = -1;
  570. setSpecific(TPE_KEY, TSS_TPERESET);
  572. long toCheck = flags & ~(TPNOTRAN | TPSENDONLY | TPRECVONLY | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  574. if (toCheck != 0) {
  575. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  576. setSpecific(TPE_KEY, TSS_TPEINVAL);
  577. } else {
  578. if (!(flags & TPSENDONLY || flags & TPRECVONLY)) {
  579. setSpecific(TPE_KEY, TSS_TPEINVAL);
  580. } else {
  581. int len = 0;
  582. if (idata != NULL) {
  583. len = ::bufferSize(idata, ilen);
  584. }
  585. if (len != -1) {
  586. if (clientinit() != -1) {
  587. int cd = -1;
  588. Session* session = NULL;
  589. try {
  590. session = ptrAtmiBrokerClient->createSession(cd, svc);
  591. if (cd != -1) {
  592. int sendOk = ::send(session, session->getReplyTo(),
  593. idata, len, cd, flags | TPCONV, 0, 0);
  594. if (sendOk != -1) {
  595. long olen = 4;
  596. char* odata = (char*) tpalloc(
  597. (char*) "X_OCTET", NULL, olen);
  598. long event = 0;
  599. ::tprecv(cd, &odata, &olen, 0, &event);
  600. bool connected = strcmp(odata, "ACK") == 0;
  601. tpfree(odata);
  602. if (connected) {
  603. toReturn = cd;
  604. if (flags & TPRECVONLY) {
  605. session->setCanSend(false);
  606. LOG4CXX_DEBUG(
  607. loggerXATMI,
  608. (char*) "tpconnect set constraints session: "
  609. << session->getId()
  610. << " send: "
  611. << session->getCanSend()
  612. << " recv (not changed): "
  613. << session->getCanRecv());
  614. } else {
  615. session->setCanRecv(false);
  616. LOG4CXX_DEBUG(
  617. loggerXATMI,
  618. (char*) "tpconnect set constraints session: "
  619. << session->getId()
  620. << " send (not changed): "
  621. << session->getCanSend()
  622. << " recv: "
  623. << session->getCanRecv());
  624. }
  625. } else {
  626. LOG4CXX_DEBUG(loggerXATMI,
  627. (char*) "COULD NOT CONNECT: " << cd);
  628. ptrAtmiBrokerClient->closeSession(cd);
  629. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  630. }
  631. } else {
  632. LOG4CXX_DEBUG(loggerXATMI,
  633. (char*) "Session got dudded: " << cd);
  634. ptrAtmiBrokerClient->closeSession(cd);
  635. }
  636. } else if (tperrno == 0) {
  637. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  638. }
  639. } catch (...) {
  640. LOG4CXX_ERROR(
  641. loggerXATMI,
  642. (char*) "tpconnect failed to connect to service queue");
  643. setSpecific(TPE_KEY, TSS_TPENOENT);
  644. if (cd != -1) {
  645. ptrAtmiBrokerClient->closeSession(cd);
  646. }
  647. }
  648. }
  649. }
  650. }
  651. }
  652. LOG4CXX_TRACE(loggerXATMI, (char*) "tpconnect return: " << toReturn
  653. << " tperrno: " << tperrno);
  654. return toReturn;
  655. }
  657. int tpgetrply(int *id, char ** odata, long *olen, long flags) {
  658. LOG4CXX_TRACE(loggerXATMI, (char*) "tpgetrply " << id);
  659. int toReturn = -1;
  660. setSpecific(TPE_KEY, TSS_TPERESET);
  662. if (odata == NULL || *odata == NULL) {
  663. setSpecific(TPE_KEY, TSS_TPEINVAL);
  664. return toReturn;
  665. }
  667. long toCheck = flags & ~(TPGETANY | TPNOCHANGE | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  669. if (toCheck != 0) {
  670. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  671. setSpecific(TPE_KEY, TSS_TPEINVAL);
  672. } else {
  673. if (clientinit() != -1) {
  674. if (id && olen) {
  675. Session* session = ptrAtmiBrokerClient->getSession(*id);
  676. if (session == NULL) {
  677. setSpecific(TPE_KEY, TSS_TPEBADDESC);
  678. } else {
  679. long event = 0;
  680. toReturn = ::receive(*id, session, odata, olen, flags,
  681. &event, true);
  682. txx_resume(*id);
  683. }
  684. } else {
  685. setSpecific(TPE_KEY, TSS_TPEINVAL);
  686. }
  687. }
  688. }
  689. LOG4CXX_TRACE(loggerXATMI, (char*) "tpgetrply return: " << toReturn
  690. << " tperrno: " << tperrno);
  691. return toReturn;
  692. }
  694. int tpcancel(int id) {
  695. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcancel " << id);
  696. int toReturn = -1;
  698. setSpecific(TPE_KEY, TSS_TPERESET);
  699. if (::txx_isCdTransactional(id)) {
  700. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcancel not allowed (TSS_TPETRAN)");
  701. setSpecific(TPE_KEY, TSS_TPETRAN);
  702. } else if (clientinit() != -1) {
  703. if (getSpecific(TSS_KEY)) {
  704. setSpecific(TPE_KEY, TSS_TPETRAN);
  705. }
  706. if (ptrAtmiBrokerClient->getSession(id) != NULL) {
  707. ptrAtmiBrokerClient->closeSession(id);
  708. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcancel session closed");
  709. toReturn = 0;
  710. } else {
  711. setSpecific(TPE_KEY, TSS_TPEBADDESC);
  712. }
  713. }
  714. LOG4CXX_TRACE(loggerXATMI, (char*) "tpcancel return: " << toReturn
  715. << " tperrno: " << tperrno);
  716. return toReturn;
  717. }
  719. int tpsend(int id, char* idata, long ilen, long flags, long *revent) {
  720. LOG4CXX_TRACE(loggerXATMI, (char*) "tpsend " << id);
  721. int toReturn = -1;
  722. setSpecific(TPE_KEY, TSS_TPERESET);
  724. long toCheck = flags & ~(TPRECVONLY | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  726. if (toCheck != 0) {
  727. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  728. setSpecific(TPE_KEY, TSS_TPEINVAL);
  729. } else {
  730. int len = ::bufferSize(idata, ilen);
  731. if (len != -1) {
  732. if (clientinit() != -1) {
  733. Session* session = (Session*) getSpecific(SVC_SES);
  734. if (session != NULL) {
  735. if (session->getId() != id) {
  736. session = NULL;
  737. }
  738. }
  739. if (session == NULL) {
  740. if (clientinit() != -1) {
  741. session = ptrAtmiBrokerClient->getSession(id);
  742. if (session == NULL) {
  743. setSpecific(TPE_KEY, TSS_TPEBADDESC);
  744. len = -1;
  745. }
  746. }
  747. }
  748. if (len != -1) {
  749. if (session->getLastEvent() != 0) {
  750. if (revent != 0) {
  751. *revent = session->getLastEvent();
  752. LOG4CXX_DEBUG(
  753. loggerXATMI,
  754. (char*) "Session has event, will be closed: "
  755. << *revent);
  756. } else {
  757. LOG4CXX_ERROR(
  758. loggerXATMI,
  759. (char*) "Session has event, will be closed: "
  760. << session->getLastEvent());
  761. }
  763. if (session->getLastEvent() == TPEV_SVCFAIL) {
  764. setTpurcode(session->getLastRCode());
  765. } else if (session->getLastEvent() == TPEV_SVCSUCC
  766. || session->getLastEvent() == TPEV_DISCONIMM) {
  767. setSpecific(TPE_KEY, TSS_TPEEVENT);
  768. toReturn = -1;
  769. }
  770. ptrAtmiBrokerClient->closeSession(id);
  771. LOG4CXX_TRACE(loggerXATMI,
  772. (char*) "tpsend closed session");
  773. } else {
  774. toReturn = ::send(session, session->getReplyTo(),
  775. idata, len, id, flags, 0, 0);
  776. if (toReturn != -1 && flags & TPRECVONLY) {
  777. session->setCanSend(false);
  778. session->setCanRecv(true);
  779. LOG4CXX_DEBUG(loggerXATMI,
  780. (char*) "tpsend set constraints session: "
  781. << session->getId() << " send: "
  782. << session->getCanSend()
  783. << " recv: "
  784. << session->getCanRecv());
  785. }
  786. }
  787. }
  788. }
  789. }
  790. }
  791. LOG4CXX_TRACE(loggerXATMI, (char*) "tpsend return: " << toReturn
  792. << " tperrno: " << tperrno);
  793. return toReturn;
  794. }
  796. int tprecv(int id, char ** odata, long *olen, long flags, long* event) {
  797. LOG4CXX_TRACE(loggerXATMI, (char*) "tprecv " << id);
  798. int toReturn = -1;
  799. setSpecific(TPE_KEY, TSS_TPERESET);
  800. *event = 0;
  802. long toCheck = flags & ~(TPNOCHANGE | TPNOBLOCK | TPNOTIME | TPSIGRSTRT);
  804. if (toCheck != 0) {
  805. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  806. setSpecific(TPE_KEY, TSS_TPEINVAL);
  807. } else {
  808. if (clientinit() != -1) {
  809. Session* session = (Session*) getSpecific(SVC_SES);
  810. if (session != NULL && session->getId() != id) {
  811. session = NULL;
  812. }
  813. if (session == NULL) {
  814. if (clientinit() != -1) {
  815. session = ptrAtmiBrokerClient->getSession(id);
  816. }
  817. }
  818. if (session == NULL) {
  819. setSpecific(TPE_KEY, TSS_TPEBADDESC);
  820. } else {
  821. toReturn = ::receive(id, session, odata, olen, flags, event,
  822. false);
  823. if (*event == TPEV_SVCSUCC || *event == TPEV_DISCONIMM
  824. || *event == TPEV_SENDONLY || *event == TPEV_SVCFAIL
  825. || *event == TPEV_SVCERR) {
  826. setSpecific(TPE_KEY, TSS_TPEEVENT);
  827. toReturn = -1;
  828. }
  829. }
  830. }
  831. }
  832. LOG4CXX_TRACE(loggerXATMI, (char*) "tprecv return: " << toReturn
  833. << " tperrno: " << tperrno);
  834. return toReturn;
  835. }
  837. void tpreturn(int rval, long rcode, char* idata, long ilen, long flags) {
  838. LOG4CXX_TRACE(loggerXATMI, (char*) "tpreturn " << rval);
  839. setSpecific(TPE_KEY, TSS_TPERESET);
  841. long toCheck = flags;
  843. if (toCheck != 0) {
  844. LOG4CXX_TRACE(loggerXATMI, (char*) "invalid flags remain: " << toCheck);
  845. setSpecific(TPE_KEY, TSS_TPEINVAL);
  846. } else {
  847. if (clientinit() != -1) {
  848. int len = 0;
  849. if (idata != NULL) {
  850. len = ::bufferSize(idata, ilen);
  851. }
  852. Session* session = (Session*) getSpecific(SVC_SES);
  853. ServiceDispatcher* dispatcher = (ServiceDispatcher*) getSpecific(SVC_KEY);
  854. if (session != NULL) {
  855. if (!session->getCanSend()
  856. && !(rval == TPFAIL && idata == NULL)) {
  857. LOG4CXX_TRACE(loggerXATMI, (char*) "generating TPESVCERR");
  858. rcode = TPESVCERR;
  859. rval = TPFAIL;
  860. }
  861. if (rcode == TPESVCERR || len == -1) {
  862. rval = TPFAIL;
  863. }
  864. if (!session->getCanSend()) {
  865. rval = DISCON;
  866. }
  867. session->setCanRecv(false);
  869. if (rcode == TPESVCERR || len == -1) {
  870. // mark rollback only and disassociate tx if present
  871. txx_rollback_only();
  872. if (getSpecific(TSS_KEY) != NULL)
  873. txx_release_control(txx_unbind(true));
  874. if (idata != NULL) {
  875. ::tpfree(idata);
  876. }
  877. ::send(session, "", NULL, 0, 0, flags, rval, TPESVCERR);
  878. LOG4CXX_TRACE(loggerXATMI, (char*) "sent TPESVCERR");
  879. if(dispatcher != NULL) {
  880. LOG4CXX_TRACE(loggerXATMI, (char*) "update error counter");
  881. dispatcher->updateErrorCounter();
  882. }
  883. } else {
  884. if (rval != TPSUCCESS && rval != TPFAIL) {
  885. rval = TPFAIL;
  886. LOG4CXX_TRACE(loggerXATMI, (char*) "generating TPFAIL");
  887. }
  888. if (rval == TPFAIL) {
  889. txx_rollback_only();
  890. LOG4CXX_TRACE(loggerXATMI, (char*) "will send TPFAIL");
  891. if(dispatcher != NULL) {
  892. LOG4CXX_TRACE(loggerXATMI, (char*) "update error counter");
  893. dispatcher->updateErrorCounter();
  894. }
  895. }
  897. // TODO send a fail if there are any outstanding replies or
  898. // open connections, or if any work done within the service
  899. // caused its transaction to be marked rollback-only
  902. // mark rollback only and disassociate tx if present
  903. if (getSpecific(TSS_KEY) != NULL)
  904. txx_release_control(txx_unbind((rval == TPFAIL)));
  906. ::send(session, "", idata, len, 0, flags, rval, rcode);
  907. LOG4CXX_TRACE(loggerXATMI, (char*) "sent response");
  908. }
  910. ::tpfree(idata);
  911. session->setSendTo(NULL);
  912. session->setCanSend(false);
  913. LOG4CXX_DEBUG(loggerXATMI,
  914. (char*) "tpreturn set constraints session: "
  915. << session->getId() << " send: "
  916. << session->getCanSend() << " recv: "
  917. << session->getCanRecv());
  919. } else {
  920. LOG4CXX_DEBUG(loggerXATMI, (char*) "Session is null");
  921. setSpecific(TPE_KEY, TSS_TPEPROTO);
  922. }
  923. }
  924. }
  925. LOG4CXX_TRACE(loggerXATMI, (char*) "tpreturn returning" << " tperrno: "
  926. << tperrno);
  927. }
  929. int tpdiscon(int id) {
  930. LOG4CXX_TRACE(loggerXATMI, (char*) "tpdiscon " << id);
  931. setSpecific(TPE_KEY, TSS_TPERESET);
  932. int toReturn = -1;
  933. if (clientinit() != -1) {
  934. LOG4CXX_DEBUG(loggerXATMI, (char*) "end - id: " << id);
  935. Session* session = ptrAtmiBrokerClient->getSession(id);
  936. if (session == NULL) {
  937. setSpecific(TPE_KEY, TSS_TPEBADDESC);
  938. } else {
  939. // CHECK TO MAKE SURE THE REMOTE SIDE IS "EXPECTING" DISCONNECTS STILL
  940. if (session->getLastEvent() == 0) {
  941. // SEND THE DISCONNECT TO THE REMOTE SIDE
  942. ::send(session, "", NULL, 0, id, TPNOTRAN, DISCON, 0);
  943. }
  944. try {
  945. if (getSpecific(TSS_KEY)) {
  946. toReturn = txx_rollback_only();
  947. }
  948. ptrAtmiBrokerClient->closeSession(id);
  949. LOG4CXX_TRACE(loggerXATMI, (char*) "tpdiscon session closed");
  950. } catch (...) {
  951. LOG4CXX_ERROR(loggerXATMI, (char*) "tpdiscon: call failed");
  952. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  953. }
  954. }
  955. }
  956. LOG4CXX_TRACE(loggerXATMI, (char*) "tpdiscon return: " << toReturn
  957. << " tperrno: " << tperrno);
  958. return toReturn;
  959. }