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

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

中间件编程

开发平台：

Java

AtmiBrokerServer.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.  */
  18. #include <string>
  19. #include <sstream>
  20. #include <queue>
  22. #ifdef TAO_COMP
  23. #include <orbsvcs/CosNamingS.h>
  24. #endif
  26. #include "log4cxx/logger.h"
  27. #include "AtmiBrokerInit.h"
  28. #include "AtmiBrokerServer.h"
  29. #include "AtmiBrokerPoaFac.h"
  30. #include "AtmiBrokerEnv.h"
  31. #include "AtmiBrokerServerControl.h"
  32. #include "AtmiBrokerMem.h"
  33. #include "txx.h"
  34. #include "OrbManagement.h"
  35. #include "SymbolLoader.h"
  36. #include "ace/Get_Opt.h"
  37. #include "ace/OS_NS_stdio.h"
  38. #include "ace/OS_NS_stdlib.h"
  39. #include "ace/OS_NS_string.h"
  40. #include "ace/Default_Constants.h"
  41. #include "ace/Signal.h"
  42. #include "ThreadLocalStorage.h"
  43. #include "xatmi.h"
  45. // WORK AROUND FOR NO tx.h
  46. #define TX_OK              0
  47. #ifdef __cplusplus
  48. extern "C" {
  49. #endif
  50. extern BLACKTIE_TX_DLL int tx_open(void);
  51. #ifdef __cplusplus
  52. }
  53. #endif
  55. extern void ADMIN(TPSVCINFO* svcinfo);
  56. extern const char* version;
  58. log4cxx::LoggerPtr loggerAtmiBrokerServer(log4cxx::Logger::getLogger(
  59. "AtmiBrokerServer"));
  60. AtmiBrokerServer * ptrServer = NULL;
  61. bool serverInitialized = false;
  62. PortableServer::POA_var server_poa;
  63. bool configFromCmdline = false;
  64. int errorBootAdminService = 0;
  65. char configDir[256];
  66. char server[30];
  67. int serverid = 0;
  69. int server_sigint_handler_callback(int sig_type) {
  70. LOG4CXX_INFO(
  71. loggerAtmiBrokerServer,
  72. (char*) "SIGINT Detected: Shutting down server this may take several minutes");
  73. if (ptrServer != NULL)
  74. ptrServer->shutdown();
  75. LOG4CXX_INFO(loggerAtmiBrokerServer,
  76. (char*) "SIGINT Detected: Shutdown complete");
  77. return -1;
  79. }
  81. int serverrun() {
  82. setSpecific(TPE_KEY, TSS_TPERESET);
  83. return ptrServer->block();
  84. }
  86. int parsecmdline(int argc, char** argv) {
  87. ACE_Get_Opt getopt(argc, argv, ACE_TEXT("c:i:"));
  88. int c;
  89. int r = 0;
  90. bool isSetServerId = false;
  92. configFromCmdline = false;
  93. while ((c = getopt()) != -1) {
  94. switch ((char) c) {
  95. case 'c':
  96. configFromCmdline = true;
  97. ACE_OS::strncpy(configDir, getopt.opt_arg(), 256);
  98. break;
  99. case 'i':
  100. serverid = atoi(getopt.opt_arg());
  101. if (serverid <= 0 || serverid > 9) {
  102. r = -1;
  103. } else {
  104. isSetServerId = true;
  105. }
  106. break;
  107. default:
  108. r = -1;
  109. }
  110. }
  112. int last = getopt.opt_ind();
  113. if (r == 0 && last < argc) {
  114. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "opt_ind is " << last
  115. << ", server is " << argv[last]);
  116. ACE_OS::strncpy(server, argv[last], 30);
  117. }
  119. if (isSetServerId == false) {
  120. fprintf(stderr,
  121. "you must specify a server id with -i greater than 0 and less than 10n");
  122. r = -1;
  123. }
  125. return r;
  126. }
  128. const char* getConfiguration() {
  129. const char* dir = NULL;
  131. if (configFromCmdline) {
  132. dir = configDir;
  133. }
  135. return dir;
  136. }
  138. int serverinit(int argc, char** argv) {
  139. AtmiBrokerInitSingleton::instance();
  140. setSpecific(TPE_KEY, TSS_TPERESET);
  141. int toReturn = 0;
  143. ACE_OS::strncpy(server, "default", 30);
  145. if (argc > 0 && parsecmdline(argc, argv) != 0) {
  146. fprintf(stderr, "usage:%s [-c config] -i id [server]n", argv[0]);
  147. toReturn = -1;
  148. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  149. }
  151. if (toReturn != -1 && ptrServer == NULL) {
  152. const char* configuration = getConfiguration();
  153. if (configuration != NULL) {
  154. AtmiBrokerEnv::set_configuration(configuration);
  155. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  156. (char*) "set AtmiBrokerEnv configuration type "
  157. << configuration);
  158. }
  160. try {
  161. AtmiBrokerEnv* env = AtmiBrokerEnv::get_instance();
  162. std::stringstream sname;
  163. std::stringstream sid;
  164. sname << "BLACKTIE_SERVER_NAME=" << domain << server << serverid;
  165. sid << "BLACKTIE_SERVER_ID=" << serverid;
  166. env->putenv((char *) (sname.str().c_str()));
  167. env->putenv((char *) (sid.str().c_str()));
  169. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "serverinit called");
  170. ptrServer = new AtmiBrokerServer();
  171. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "serverInitialized=" << serverInitialized);
  173. if (!serverInitialized) {
  174. ::serverdone();
  175. toReturn = -1;
  176. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  177. } else {
  178. ptrServer->advertiseAtBootime();
  180. // install a handler for the default set of signals (namely, SIGINT and SIGTERM)
  181. (env->getSignalHandler()).addSignalHandler(
  182. server_sigint_handler_callback, true);
  184. LOG4CXX_INFO(loggerAtmiBrokerServer, (char*) "Server "
  185. << serverid << " Running");
  186. }
  187. } catch (...) {
  188. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  189. (char*) "Server startup failed");
  190. toReturn = -1;
  191. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  192. }
  193. }
  194. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "serverinit returning: "
  195. << toReturn);
  196. if (toReturn != 0) {
  197. LOG4CXX_FATAL(loggerAtmiBrokerServer, (char*) "serverinit failed");
  198. }
  199. return toReturn;
  200. }
  202. int serverdone() {
  203. setSpecific(TPE_KEY, TSS_TPERESET);
  204. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "serverdone called");
  205. if (ptrServer) {
  206. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  207. (char*) "serverdone deleting Corba server");
  208. delete ptrServer;
  209. ptrServer = NULL;
  210. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  211. (char*) "serverdone deleted Corba server");
  213. }
  214. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "serverdone returning 0");
  215. // if (configFromCmdline) {
  216. // char* toFree = ACE_OS::getenv("BLACKTIE_CONFIGURATION");
  217. // free(toFree);
  218. // }
  219. return 0;
  220. }
  222. int isadvertised(char* name) {
  223. if (ptrServer) {
  224. if (ptrServer->isAdvertised(name)) {
  225. return 0;
  226. }
  227. }
  228. return -1;
  229. }
  231. int getServiceStatus(char** str, char* svc) {
  232. if (ptrServer) {
  233. return ptrServer->getServiceStatus(str, svc);
  234. }
  236. return -1;
  237. }
  239. long getServiceMessageCounter(char* serviceName) {
  240. if (ptrServer) {
  241. return ptrServer->getServiceMessageCounter(serviceName);
  242. }
  244. return 0;
  245. }
  247. long getServiceErrorCounter(char* serviceName) {
  248. if (ptrServer) {
  249. return ptrServer->getServiceErrorCounter(serviceName);
  250. }
  252. return 0;
  253. }
  255. void getResponseTime(char* serviceName, unsigned long* min, unsigned long* avg,
  256. unsigned long* max) {
  257. if (ptrServer) {
  258. ptrServer->getResponseTime(serviceName, min, avg, max);
  259. }
  260. }
  262. int advertiseByAdmin(char* name) {
  263. if (isadvertised(name) == 0) {
  264. return 0;
  265. }
  267. if (ptrServer) {
  268. if (ptrServer->advertiseService(name)) {
  269. return 0;
  270. }
  271. }
  272. return -1;
  273. }
  275. int pauseServerByAdmin() {
  276. if (ptrServer) {
  277. return ptrServer->pause();
  278. }
  280. return -1;
  281. }
  283. int resumeServerByAdmin() {
  284. if (ptrServer) {
  285. return ptrServer->resume();
  286. }
  288. return -1;
  289. }
  291. // AtmiBrokerServer constructor
  292. //
  293. // Note: since we use virtual inheritance, we must include an
  294. // initialiser for all the virtual base class constructors that
  295. // require arguments, even those that we inherit indirectly.
  296. //
  297. AtmiBrokerServer::AtmiBrokerServer() {
  298. try {
  299. finish = new SynchronizableObject();
  300. serverName = server;
  301. isPause = false;
  302. unsigned int i;
  304. serverInfo.serverName = NULL;
  306. for (i = 0; i < servers.size(); i++) {
  307. if (strcmp(servers[i]->serverName, serverName) == 0) {
  308. serverInfo.serverName = strdup(servers[i]->serverName);
  309. // add service ADMIN
  310. char adm[XATMI_SERVICE_NAME_LENGTH + 1];
  311. ACE_OS::snprintf(adm, XATMI_SERVICE_NAME_LENGTH + 1,
  312. "%s_ADMIN_%d", server, serverid);
  313. ServiceInfo service;
  314. memset(&service, 0, sizeof(ServiceInfo));
  315. service.serviceName = strdup(adm);
  316. #ifdef WIN32
  317. service.transportLib = strdup("atmibroker-hybrid.dll");
  318. #else
  319. service.transportLib = strdup("libatmibroker-hybrid.so");
  320. #endif
  321. service.poolSize = 1;
  322. service.advertised = false;
  323. serverInfo.serviceVector.push_back(service);
  325. for (unsigned int j = 0; j < servers[i]->serviceVector.size(); j++) {
  326. ServiceInfo service;
  327. memset(&service, 0, sizeof(ServiceInfo));
  328. service.serviceName = strdup(
  329. servers[i]->serviceVector[j].serviceName);
  330. service.transportLib = strdup(
  331. servers[i]->serviceVector[j].transportLib);
  333. if (servers[i]->serviceVector[j].function_name) {
  334. service.function_name = strdup(
  335. servers[i]->serviceVector[j].function_name);
  336. } else {
  337. service.function_name = NULL;
  338. }
  339. if (servers[i]->serviceVector[j].library_name) {
  340. service.library_name = strdup(
  341. servers[i]->serviceVector[j].library_name);
  342. } else {
  343. service.library_name = NULL;
  344. }
  345. service.poolSize = servers[i]->serviceVector[j].poolSize;
  346. service.advertised
  347. = servers[i]->serviceVector[j].advertised;
  348. serverInfo.serviceVector.push_back(service);
  349. }
  350. break;
  351. }
  352. }
  354. if (i == servers.size()) {
  355. LOG4CXX_WARN(loggerAtmiBrokerServer, serverName
  356. << " has no configuration");
  357. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  358. return;
  359. }
  361. if (tx_open() != TX_OK) {
  362. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  364. LOG4CXX_ERROR(
  365. loggerAtmiBrokerServer,
  366. serverName
  367. << (char *) " transaction configuration error, aborting server startup");
  368. } else {
  369. // make ADMIN service mandatory for server
  370. char adm[XATMI_SERVICE_NAME_LENGTH + 1];
  371. ACE_OS::snprintf(adm, XATMI_SERVICE_NAME_LENGTH + 1, "%s_ADMIN_%d",
  372. server, serverid);
  373. if (!advertiseService(adm, ADMIN)) {
  374. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertise admin service failed");
  375. return;
  376. }
  378. if (errorBootAdminService == 2) {
  379. LOG4CXX_WARN(loggerAtmiBrokerServer,
  380. (char*) "Maybe the same server id running");
  381. throw std::exception();
  382. }
  383. serverInitialized = true;
  385. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  386. (char*) "server_init(): finished.");
  387. }
  388. } catch (CORBA::Exception& e) {
  389. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  390. (char*) "serverinit - Unexpected CORBA exception: "
  391. << e._name());
  392. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  393. }
  394. }
  396. // ~AtmiBrokerServer destructor.
  397. //
  398. AtmiBrokerServer::~AtmiBrokerServer() {
  399. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "destructor");
  400. server_done();
  401. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "Server done");
  403. if (finish != NULL) {
  404. delete finish;
  405. finish = NULL;
  406. }
  408. serviceData.clear();
  409. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "deleted service array");
  411. for (unsigned int i = 0; i < serverInfo.serviceVector.size(); i++) {
  412. ServiceInfo* service = &serverInfo.serviceVector[i];
  413. free(service->serviceName);
  414. free(service->transportLib);
  415. if (service->function_name != NULL) {
  416. free(service->function_name);
  417. }
  418. if (service->library_name != NULL) {
  419. free(service->library_name);
  420. }
  421. }
  422. if (serverInfo.serverName != NULL) {
  423. free(serverInfo.serverName);
  424. }
  426. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "deleting services");
  427. AtmiBrokerMem::discard_instance();
  428. txx_stop();
  429. AtmiBrokerEnv::discard_instance();
  430. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "deleted services");
  432. connections.closeConnections();
  433. serverInitialized = false;
  434. }
  436. void AtmiBrokerServer::advertiseAtBootime() {
  437. for (unsigned int i = 0; i < serverInfo.serviceVector.size(); i++) {
  438. ServiceInfo* service = &serverInfo.serviceVector[i];
  439. SVCFUNC func = NULL;
  440. bool status = false;
  442. if (service->library_name != NULL) {
  443. func = (SVCFUNC) ::lookup_symbol(service->library_name,
  444. service->function_name);
  445. if (func == NULL) {
  446. LOG4CXX_WARN(loggerAtmiBrokerServer, "can not find "
  447. << service->function_name << " in "
  448. << service->library_name);
  449. }
  450. }
  452. if (service->advertised && func != NULL) {
  453. LOG4CXX_DEBUG(loggerAtmiBrokerServer, "begin advertise "
  454. << service->serviceName);
  455. status = advertiseService((char*) service->serviceName, func);
  456. LOG4CXX_DEBUG(loggerAtmiBrokerServer, "end advertise "
  457. << service->serviceName);
  458. }
  460. updateServiceStatus(service, func, status);
  461. }
  462. }
  464. int AtmiBrokerServer::block() {
  466. int toReturn = 0;
  467. if (errorBootAdminService == 3) {
  468. LOG4CXX_INFO(loggerAtmiBrokerServer, "Domain is paused");
  469. pause();
  470. } else {
  471. LOG4CXX_INFO(loggerAtmiBrokerServer, "Server waiting for requests...");
  472. }
  474. try {
  475. finish->lock();
  476. finish->wait(0);
  477. finish->unlock();
  478. } catch (...) {
  479. LOG4CXX_ERROR(loggerAtmiBrokerServer, "Unexpected exception");
  480. toReturn = -1;
  481. }
  482. return toReturn;
  483. }
  485. void AtmiBrokerServer::shutdown() {
  486. LOG4CXX_INFO(loggerAtmiBrokerServer, "Server prepare to shutdown");
  487. // server_done(); You can't do this here as the service dispatcher will be cleaned up that is handling
  488. // the cleanup for an admin call
  489. finish->lock();
  490. finish->notify();
  491. finish->unlock();
  492. }
  494. int AtmiBrokerServer::pause() {
  495. if (!isPause) {
  496. char adm[XATMI_SERVICE_NAME_LENGTH + 1];
  497. ACE_OS::snprintf(adm, XATMI_SERVICE_NAME_LENGTH + 1, "%s_ADMIN_%d",
  498. server, serverid);
  499. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  500. != serviceData.end(); i++) {
  501. if (ACE_OS::strcmp((*i).serviceInfo->serviceName, adm) != 0) {
  502. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "pausing service"
  503. << (*i).serviceInfo->serviceName);
  504. for (std::vector<ServiceDispatcher*>::iterator j =
  505. (*i).dispatchers.begin(); j != (*i).dispatchers.end(); j++) {
  506. ServiceDispatcher* dispatcher = (*j);
  507. if (dispatcher->pause() != 0) {
  508. LOG4CXX_WARN(loggerAtmiBrokerServer,
  509. (char*) "pause service dispatcher "
  510. << dispatcher << " failed");
  511. }
  512. }
  513. }
  514. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "pause service"
  515. << (*i).serviceInfo->serviceName << " done");
  516. }
  517. isPause = true;
  518. LOG4CXX_INFO(loggerAtmiBrokerServer, (char*) "Server Pause");
  519. }
  520. return 0;
  521. }
  523. int AtmiBrokerServer::resume() {
  524. if (isPause) {
  525. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  526. != serviceData.end(); i++) {
  527. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "resuming service"
  528. << (*i).serviceInfo->serviceName);
  529. for (std::vector<ServiceDispatcher*>::iterator j =
  530. (*i).dispatchers.begin(); j != (*i).dispatchers.end(); j++) {
  531. ServiceDispatcher* dispatcher = (*j);
  532. if (dispatcher->resume() != 0) {
  533. LOG4CXX_WARN(loggerAtmiBrokerServer,
  534. (char*) "resume service dispatcher " << dispatcher
  535. << " failed");
  536. }
  537. }
  538. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "resume service"
  539. << (*i).serviceInfo->serviceName << " done");
  540. }
  541. isPause = false;
  542. LOG4CXX_INFO(loggerAtmiBrokerServer, (char*) "Server Resume");
  543. }
  544. return 0;
  545. }
  547. char *
  548. AtmiBrokerServer::getServerName() {
  549. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "getServerName");
  550. return serverName;
  551. }
  553. int AtmiBrokerServer::getServiceStatus(char** toReturn, char* svc) {
  554. int len = 0;
  555. char* str;
  556. int size = sizeof(char) * (9 + 14 + strlen(serverName) + 11 + 12 + 10);
  557. char adm[XATMI_SERVICE_NAME_LENGTH + 1];
  558. ACE_OS::snprintf(adm, XATMI_SERVICE_NAME_LENGTH + 1, "%s_ADMIN_%d", server,
  559. serverid);
  561. str = (char*) malloc(size);
  562. len += ACE_OS::sprintf(str + len, "<server>");
  563. len += ACE_OS::sprintf(str + len, "<name>%s</name>", serverName);
  564. len += ACE_OS::sprintf(str + len, "<services>");
  565. for (std::vector<ServiceStatus>::iterator i = serviceStatus.begin(); i
  566. != serviceStatus.end(); i++) {
  567. if (strcmp(adm, (*i).name) != 0 && (svc == NULL || ACE_OS::strcmp(svc,
  568. (*i).name) == 0)) {
  569. int svcsize = sizeof(char) * (50 + strlen((*i).name));
  570. size += svcsize;
  571. str = (char*) realloc(str, size);
  573. len += ACE_OS::sprintf(str + len,
  574. "<service><name>%.15s</name><status>%d</status></service>",
  575. (*i).name, isPause && (*i).status ? 2 : (*i).status);
  576. if (svc != NULL)
  577. break;
  578. }
  579. }
  581. len += ACE_OS::sprintf(str + len, "</services>");
  582. len += ACE_OS::sprintf(str + len, "</server>");
  583. *toReturn = str;
  584. return len;
  585. }
  587. void AtmiBrokerServer::updateServiceStatus(ServiceInfo* service, SVCFUNC func,
  588. bool status) {
  589. bool found = false;
  591. for (std::vector<ServiceStatus>::iterator i = serviceStatus.begin(); i
  592. != serviceStatus.end(); i++) {
  593. if (strncmp((*i).name, service->serviceName, XATMI_SERVICE_NAME_LENGTH)
  594. == 0) {
  595. (*i).func = func;
  596. (*i).status = status;
  597. found = true;
  598. break;
  599. }
  600. }
  602. if (found == false) {
  603. ServiceStatus aServiceStatus;
  604. memset(&aServiceStatus, 0, sizeof(aServiceStatus));
  605. ACE_OS::strncpy(aServiceStatus.name, service->serviceName,
  606. XATMI_SERVICE_NAME_LENGTH);
  607. aServiceStatus.func = func;
  608. aServiceStatus.status = status;
  609. serviceStatus.push_back(aServiceStatus);
  610. }
  611. }
  613. bool AtmiBrokerServer::advertiseService(char * svcname) {
  614. for (std::vector<ServiceStatus>::iterator i = serviceStatus.begin(); i
  615. != serviceStatus.end(); i++) {
  616. if (strncmp((*i).name, svcname, XATMI_SERVICE_NAME_LENGTH) == 0) {
  617. return advertiseService(svcname, (*i).func);
  618. }
  619. }
  620. LOG4CXX_WARN(
  621. loggerAtmiBrokerServer,
  622. (char*) "Could not advertise service, was not registered in btconfig.xml: "
  623. << svcname);
  624. return false;
  625. }
  627. bool AtmiBrokerServer::advertiseService(char * svcname,
  628. void(*func)(TPSVCINFO *)) {
  630. if (!svcname || strlen(svcname) == 0) {
  631. setSpecific(TPE_KEY, TSS_TPEINVAL);
  632. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertiseService invalid service name");
  633. return false;
  634. }
  636. char* serviceName = (char*) ::malloc(XATMI_SERVICE_NAME_LENGTH + 1);
  637. memset(serviceName, '', XATMI_SERVICE_NAME_LENGTH + 1);
  638. strncat(serviceName, svcname, XATMI_SERVICE_NAME_LENGTH);
  640. bool found = false;
  641. unsigned int i;
  642. ServiceInfo* service;
  643. for (i = 0; i < serverInfo.serviceVector.size(); i++) {
  644. if (strncmp(serverInfo.serviceVector[i].serviceName, serviceName,
  645. XATMI_SERVICE_NAME_LENGTH) == 0) {
  646. found = true;
  647. service = &serverInfo.serviceVector[i];
  648. break;
  649. }
  650. }
  651. if (!found) {
  652. LOG4CXX_WARN(
  653. loggerAtmiBrokerServer,
  654. (char*) "Could not advertise service, was not registered for server in btconfig.xml: "
  655. << svcname);
  656. setSpecific(TPE_KEY, TSS_TPELIMIT);
  657. free(serviceName);
  658. return false;
  659. }
  660. void (*serviceFunction)(TPSVCINFO*) = getServiceMethod(serviceName);
  661. if (serviceFunction != NULL) {
  662. if (serviceFunction == func) {
  663. free(serviceName);
  664. return true;
  665. } else {
  666. setSpecific(TPE_KEY, TSS_TPEMATCH);
  667. free(serviceName);
  668. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertiseService same service function");
  669. return false;
  670. }
  671. } else if (serviceFunction == NULL && func == NULL) {
  672. LOG4CXX_WARN(loggerAtmiBrokerServer,
  673. (char*) "Could not advertise service, no function available: "
  674. << svcname);
  675. return false;
  676. }
  678. Connection* connection = connections.getServerConnection();
  679. if (connection == NULL) {
  680. setSpecific(TPE_KEY, TSS_TPESYSTEM);
  681. free(serviceName);
  682. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertiseService no server connection");
  683. return false;
  684. }
  686. bool toReturn = false;
  688. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertiseService(): "
  689. << serviceName);
  691. // create reference for Service Queue and cache
  692. try {
  693. toReturn = createAdminDestination(serviceName);
  694. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "advertiseService status=" << toReturn);
  695. if (toReturn) {
  696. Destination* destination;
  697. destination = connection->createDestination(serviceName);
  699. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  700. (char*) "created destination: " << serviceName);
  702. addDestination(destination, func, service);
  703. updateServiceStatus(service, func, true);
  704. advertisedServices.push_back(serviceName);
  705. LOG4CXX_INFO(loggerAtmiBrokerServer, (char*) "advertised service "
  706. << serviceName);
  707. } else {
  708. free(serviceName);
  709. }
  710. } catch (CORBA::Exception& e) {
  711. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  712. (char*) "CORBA::Exception creating the destination: "
  713. << serviceName << " Exception: " << e._name());
  714. setSpecific(TPE_KEY, TSS_TPEMATCH);
  715. try {
  716. removeAdminDestination(serviceName, true);
  717. } catch (...) {
  718. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  719. (char*) "Could not remove the destination: " << serviceName);
  720. }
  721. free(serviceName);
  722. } catch (...) {
  723. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  724. (char*) "Could not create the destination: " << serviceName);
  725. setSpecific(TPE_KEY, TSS_TPEMATCH);
  726. try {
  727. removeAdminDestination(serviceName, true);
  728. } catch (...) {
  729. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  730. (char*) "Could not remove the destination: " << serviceName);
  731. }
  732. free(serviceName);
  733. }
  734. return toReturn;
  735. }
  737. void AtmiBrokerServer::unadvertiseService(char * svcname, bool decrement) {
  738. char* serviceName = (char*) ::malloc(XATMI_SERVICE_NAME_LENGTH + 1);
  739. memset(serviceName, '', XATMI_SERVICE_NAME_LENGTH + 1);
  740. strncat(serviceName, svcname, XATMI_SERVICE_NAME_LENGTH);
  742. Connection* connection = connections.getServerConnection();
  743. if (connection == NULL) {
  744. return;
  745. }
  747. // Connection* connz = connections.getServerConnection("BAR");
  748. // delete connz;
  750. for (std::vector<char*>::iterator i = advertisedServices.begin(); i
  751. != advertisedServices.end(); i++) {
  752. char* name = (*i);
  753. if (strcmp(serviceName, name) == 0) {
  754. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  755. (char*) "remove_service_queue: " << serviceName);
  756. Destination * destination = removeDestination(serviceName);
  757. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  758. (char*) "preparing to destroy" << serviceName);
  760. connection->destroyDestination(destination);
  761. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "destroyed"
  762. << serviceName);
  763. advertisedServices.erase(i);
  764. free(name);
  765. removeAdminDestination(serviceName, decrement);
  766. LOG4CXX_INFO(loggerAtmiBrokerServer,
  767. (char*) "unadvertised service " << serviceName);
  768. break;
  769. }
  770. }
  771. free(serviceName);
  772. }
  774. bool AtmiBrokerServer::createAdminDestination(char* serviceName) {
  775. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "Creating admin queue for: "
  776. << serviceName);
  778. bool isadm = false;
  779. char adm[XATMI_SERVICE_NAME_LENGTH + 1];
  780. ACE_OS::snprintf(adm, XATMI_SERVICE_NAME_LENGTH + 1, "%s_ADMIN_%d", server,
  781. serverid);
  782. if (strcmp(adm, serviceName) == 0) {
  783. isadm = true;
  784. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  785. (char*) "advertising admin service");
  786. }
  788. long commandLength;
  789. long responseLength = 1;
  791. commandLength = strlen(serverName) + strlen(serviceName) + strlen(version)
  792. + 15 + 1;
  794. char* command = (char*) ::tpalloc((char*) "X_OCTET", NULL, commandLength);
  795. char* response = (char*) ::tpalloc((char*) "X_OCTET", NULL, responseLength);
  796. memset(command, '', commandLength);
  798. sprintf(command, "tpadvertise,%s,%s,%s,", serverName, serviceName, version);
  800. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "createAdminDestination with command " << command);
  802. if (tpcall((char*) "BTStompAdmin", command, commandLength, &response,
  803. &responseLength, TPNOTRAN) != 0) {
  804. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  805. "Could not advertise service with command: " << command);
  806. tpfree(command);
  807. tpfree(response);
  808. return false;
  809. } else if (responseLength != 1) {
  810. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  811. "Service returned with unexpected response: " << response
  812. << " with length " << responseLength);
  813. tpfree(command);
  814. tpfree(response);
  815. return false;
  816. } else if (response[0] == 4) {
  817. LOG4CXX_WARN(loggerAtmiBrokerServer, (char*) "Server vresion "
  818. << version << " can not main Domain version");
  819. tpfree(command);
  820. tpfree(response);
  821. return false;
  822. } else if (response[0] == 3) {
  823. // Dispatcher needs to be paused
  824. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  825. (char*) "Created paused admin queue for: " << serviceName);
  826. tpfree(command);
  827. tpfree(response);
  828. return true;
  829. } else if (response[0] == 1) {
  830. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  831. (char*) "Created admin queue for: " << serviceName);
  832. tpfree(command);
  833. tpfree(response);
  834. return true;
  835. } else {
  836. int r = (int) response[0];
  837. char c = response[0];
  838. // REMOVED BY TOM, NOT CLEAR WHAT THIS IS REQUIRED FOR,
  839. // IF COMMENTED BACK IN, PLEASE PROVIDE A COMMENT
  840. // if (!isadm || (errorBootAdminService = response[0]) == 2) {
  841. LOG4CXX_ERROR(loggerAtmiBrokerServer, "Service returned with error: "
  842. << response[0] << " command was " << command << " r=" << r << " c=" << c);
  843. tpfree(command);
  844. tpfree(response);
  845. return false;
  846. // }
  847. }
  848. }
  850. void AtmiBrokerServer::removeAdminDestination(char* serviceName, bool decrement) {
  851. long commandLength;
  852. long responseLength = 1;
  853. char* command;
  855. if (decrement) {
  856. commandLength = strlen(serverName) + strlen(serviceName) + 21;
  857. command = (char*) ::tpalloc((char*) "X_OCTET", NULL, commandLength);
  858. sprintf(command, "decrementconsumer,%s,%s,", serverName, serviceName);
  859. } else {
  860. commandLength = strlen(serverName) + strlen(serviceName) + strlen(
  861. "tpunadvertise,,, ");
  862. command = (char*) ::tpalloc((char*) "X_OCTET", NULL, commandLength);
  863. sprintf(command, "tpunadvertise,%s,%s,", serverName, serviceName);
  864. }
  866. char* response = (char*) ::tpalloc((char*) "X_OCTET", NULL, responseLength);
  868. LOG4CXX_DEBUG(loggerAtmiBrokerServer, "Unadvertise with command: "
  869. << command);
  870. if (tpcall((char*) "BTStompAdmin", command, commandLength, &response,
  871. &responseLength, TPNOTRAN) != 0) {
  872. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  873. "Could not unadvertise service with command: " << command);
  874. } else if (responseLength != 1) {
  875. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  876. "Service returned with unexpected response: " << response
  877. << " with length " << responseLength);
  878. } else if (response[0] == 0) {
  879. LOG4CXX_ERROR(loggerAtmiBrokerServer, "Service returned with error: "
  880. << command);
  881. } else {
  882. LOG4CXX_DEBUG(loggerAtmiBrokerServer, "Unadvertise ok");
  883. }
  884. tpfree(command);
  885. tpfree(response);
  886. }
  888. bool AtmiBrokerServer::isAdvertised(char * serviceName) {
  889. bool toReturn = false;
  890. for (std::vector<char*>::iterator i = advertisedServices.begin(); i
  891. != advertisedServices.end(); i++) {
  892. if (strncmp(serviceName, (*i), XATMI_SERVICE_NAME_LENGTH) == 0) {
  893. toReturn = true;
  894. }
  895. }
  896. return toReturn;
  897. }
  899. void AtmiBrokerServer::addDestination(Destination* destination, void(*func)(
  900. TPSVCINFO *), ServiceInfo* service) {
  901. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "addDestination: "
  902. << destination->getName());
  904. ServiceData entry;
  905. entry.destination = destination;
  906. entry.func = func;
  907. entry.serviceInfo = service;
  909. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "constructor: "
  910. << destination->getName());
  912. Connection* connection = connections.getServerConnection();
  913. if (connection == NULL) {
  914. return;
  915. }
  917. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "createPool");
  918. SynchronizableObject* reconnect = new SynchronizableObject();
  919. for (int i = 0; i < entry.serviceInfo->poolSize; i++) {
  920. ServiceDispatcher* dispatcher = new ServiceDispatcher(this,
  921. destination, connection, destination->getName(), func, isPause,
  922. reconnect);
  923. if (dispatcher->activate(THR_NEW_LWP | THR_JOINABLE, 1, 0,
  924. ACE_DEFAULT_THREAD_PRIORITY, -1, 0, 0, 0, 0, 0, 0) != 0) {
  925. delete dispatcher;
  926. LOG4CXX_ERROR(loggerAtmiBrokerServer,
  927. (char*) "Could not start thread pool");
  928. } else {
  929. entry.dispatchers.push_back(dispatcher);
  930. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) " destination "
  931. << destination);
  932. }
  933. }
  935. serviceData.push_back(entry);
  936. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "added: "
  937. << destination->getName());
  938. }
  940. Destination* AtmiBrokerServer::removeDestination(const char * aServiceName) {
  941. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "removeDestination: "
  942. << aServiceName);
  943. Destination* toReturn = NULL;
  944. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  945. != serviceData.end(); i++) {
  946. if (strncmp((*i).serviceInfo->serviceName, aServiceName,
  947. XATMI_SERVICE_NAME_LENGTH) == 0) {
  948. toReturn = (*i).destination;
  949. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "removing service "
  950. << aServiceName);
  951. for (std::vector<ServiceDispatcher*>::iterator j =
  952. (*i).dispatchers.begin(); j != (*i).dispatchers.end(); j++) {
  953. ServiceDispatcher* dispatcher = (*j);
  954. dispatcher->shutdown();
  955. }
  956. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "shutdown notified "
  957. << aServiceName);
  959. for (std::vector<ServiceDispatcher*>::iterator j =
  960. (*i).dispatchers.begin(); j != (*i).dispatchers.end(); j++) {
  961. toReturn->disconnect();
  962. }
  963. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  964. (char*) "disconnect notified " << aServiceName);
  966. SynchronizableObject* reconnect = NULL;
  968. for (std::vector<ServiceDispatcher*>::iterator j =
  969. (*i).dispatchers.begin(); j != (*i).dispatchers.end();) {
  970. ServiceDispatcher* dispatcher = (*j);
  971. if (dispatcher != NULL) {
  972. LOG4CXX_TRACE(loggerAtmiBrokerServer,
  973. (char*) "Waiting for dispatcher notified "
  974. << aServiceName);
  975. dispatcher->wait();
  976. LOG4CXX_TRACE(loggerAtmiBrokerServer,
  977. (char*) "Deleting dispatcher " << aServiceName);
  978. reconnect = dispatcher->getReconnect();
  979. delete dispatcher;
  980. LOG4CXX_TRACE(loggerAtmiBrokerServer,
  981. (char*) "Dispatcher deleted " << aServiceName);
  982. } else {
  983. LOG4CXX_TRACE(loggerAtmiBrokerServer,
  984. (char*) "NULL Dispatcher detected for"
  985. << aServiceName);
  986. }
  987. LOG4CXX_TRACE(loggerAtmiBrokerServer,
  988. (char*) "Erasing dispatcher " << aServiceName);
  989. j = (*i).dispatchers.erase(j);
  990. }
  991. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  992. (char*) "waited for dispatcher: " << aServiceName);
  993. if (reconnect != NULL) {
  994. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  995. (char*) "Deleting reconnect");
  996. delete reconnect;
  997. }
  999. updateServiceStatus((*i).serviceInfo, (*i).func, false);
  1000. serviceData.erase(i);
  1001. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "removed: "
  1002. << aServiceName);
  1003. break;
  1004. }
  1005. }
  1006. return toReturn;
  1007. }
  1009. void AtmiBrokerServer::server_done() {
  1010. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "server_done()");
  1012. for (int i = serverInfo.serviceVector.size() - 1; i >= 0; i--) {
  1013. char* svcname = (char*) serverInfo.serviceVector[i].serviceName;
  1014. if (isAdvertised(svcname)) {
  1015. unadvertiseService(svcname, true);
  1016. }
  1017. }
  1019. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "server_done(): returning.");
  1020. }
  1022. long AtmiBrokerServer::getServiceMessageCounter(char* serviceName) {
  1023. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  1024. != serviceData.end(); i++) {
  1025. if (strncmp((*i).destination->getName(), serviceName,
  1026. XATMI_SERVICE_NAME_LENGTH) == 0) {
  1027. long counter = 0;
  1028. for (std::vector<ServiceDispatcher*>::iterator dispatcher =
  1029. (*i).dispatchers.begin(); dispatcher
  1030. != (*i).dispatchers.end(); dispatcher++) {
  1031. counter += (*dispatcher)->getCounter();
  1032. }
  1033. return counter;
  1034. }
  1035. }
  1036. return 0;
  1037. }
  1039. long AtmiBrokerServer::getServiceErrorCounter(char* serviceName) {
  1040. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  1041. != serviceData.end(); i++) {
  1042. if (strncmp((*i).destination->getName(), serviceName,
  1043. XATMI_SERVICE_NAME_LENGTH) == 0) {
  1044. long counter = 0;
  1045. for (std::vector<ServiceDispatcher*>::iterator dispatcher =
  1046. (*i).dispatchers.begin(); dispatcher
  1047. != (*i).dispatchers.end(); dispatcher++) {
  1048. counter += (*dispatcher)->getErrorCounter();
  1049. }
  1050. return counter;
  1051. }
  1052. }
  1053. return 0;
  1054. }
  1056. void AtmiBrokerServer::getResponseTime(char* serviceName, unsigned long* min,
  1057. unsigned long* avg, unsigned long* max) {
  1058. *min = 0;
  1059. *avg = 0;
  1060. *max = 0;
  1062. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  1063. != serviceData.end(); i++) {
  1064. if (strncmp((*i).destination->getName(), serviceName,
  1065. XATMI_SERVICE_NAME_LENGTH) == 0) {
  1066. long counter = 0;
  1067. long total = 0;
  1068. unsigned long min_time;
  1069. unsigned long max_time;
  1070. unsigned long avg_time;
  1072. for (std::vector<ServiceDispatcher*>::iterator dispatcher =
  1073. (*i).dispatchers.begin(); dispatcher
  1074. != (*i).dispatchers.end(); dispatcher++) {
  1075. counter = (*dispatcher)->getCounter();
  1076. (*dispatcher)->getResponseTime(&min_time, &avg_time, &max_time);
  1077. if (*min == 0 || min_time < *min) {
  1078. *min = min_time;
  1079. }
  1081. if (total != 0 || counter != 0) {
  1082. *avg = ((*avg) * total + avg_time * counter) / (total
  1083. + counter);
  1084. }
  1086. if (max_time > *max) {
  1087. *max = max_time;
  1088. }
  1089. }
  1090. }
  1091. }
  1092. }
  1094. void (*AtmiBrokerServer::getServiceMethod(const char * aServiceName))(TPSVCINFO *) {
  1095. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "getServiceMethod: "
  1096. << aServiceName);
  1098. for (std::vector<ServiceData>::iterator i = serviceData.begin(); i
  1099. != serviceData.end(); i++) {
  1100. if (strncmp((*i).destination->getName(), aServiceName,
  1101. XATMI_SERVICE_NAME_LENGTH) == 0) {
  1102. LOG4CXX_DEBUG(loggerAtmiBrokerServer, (char*) "found: "
  1103. << aServiceName);
  1104. return (*i).func;
  1105. }
  1106. }
  1107. LOG4CXX_DEBUG(loggerAtmiBrokerServer,
  1108. (char*) "getServiceMethod out: " << aServiceName);
  1109. return NULL;
  1110. }