TestClient.cc 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750
  1. /**
  2. * Licensed to the Apache Software Foundation (ASF) under one
  3. * or more contributor license agreements. See the NOTICE file
  4. * distributed with this work for additional information
  5. * regarding copyright ownership. The ASF licenses this file
  6. * to you under the Apache License, Version 2.0 (the
  7. * "License"); you may not use this file except in compliance
  8. * with the License. You may obtain a copy of the License at
  9. *
  10. * http://www.apache.org/licenses/LICENSE-2.0
  11. *
  12. * Unless required by applicable law or agreed to in writing, software
  13. * distributed under the License is distributed on an "AS IS" BASIS,
  14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. * See the License for the specific language governing permissions and
  16. * limitations under the License.
  17. */
  18. #include <cppunit/extensions/HelperMacros.h>
  19. #include "CppAssertHelper.h"
  20. #include <stdlib.h>
  21. #include <sys/select.h>
  22. #include "CollectionUtil.h"
  23. #include "ThreadingUtil.h"
  24. using namespace Util;
  25. #include "Vector.h"
  26. using namespace std;
  27. #include <cstring>
  28. #include <list>
  29. #include <zookeeper.h>
  30. #ifdef THREADED
  31. static void yield(zhandle_t *zh, int i)
  32. {
  33. sleep(i);
  34. }
  35. #else
  36. static void yield(zhandle_t *zh, int seconds)
  37. {
  38. int fd;
  39. int interest;
  40. int events;
  41. struct timeval tv;
  42. int rc;
  43. time_t expires = time(0) + seconds;
  44. time_t timeLeft = seconds;
  45. fd_set rfds, wfds, efds;
  46. FD_ZERO(&rfds);
  47. FD_ZERO(&wfds);
  48. FD_ZERO(&efds);
  49. while(timeLeft >= 0) {
  50. zookeeper_interest(zh, &fd, &interest, &tv);
  51. if (fd != -1) {
  52. if (interest&ZOOKEEPER_READ) {
  53. FD_SET(fd, &rfds);
  54. } else {
  55. FD_CLR(fd, &rfds);
  56. }
  57. if (interest&ZOOKEEPER_WRITE) {
  58. FD_SET(fd, &wfds);
  59. } else {
  60. FD_CLR(fd, &wfds);
  61. }
  62. } else {
  63. fd = 0;
  64. }
  65. FD_SET(0, &rfds);
  66. if (tv.tv_sec > timeLeft) {
  67. tv.tv_sec = timeLeft;
  68. }
  69. rc = select(fd+1, &rfds, &wfds, &efds, &tv);
  70. timeLeft = expires - time(0);
  71. events = 0;
  72. if (FD_ISSET(fd, &rfds)) {
  73. events |= ZOOKEEPER_READ;
  74. }
  75. if (FD_ISSET(fd, &wfds)) {
  76. events |= ZOOKEEPER_WRITE;
  77. }
  78. zookeeper_process(zh, events);
  79. }
  80. }
  81. #endif
  82. typedef struct evt {
  83. string path;
  84. int type;
  85. } evt_t;
  86. typedef struct watchCtx {
  87. private:
  88. list<evt_t> events;
  89. watchCtx(const watchCtx&);
  90. watchCtx& operator=(const watchCtx&);
  91. public:
  92. bool connected;
  93. zhandle_t *zh;
  94. Mutex mutex;
  95. watchCtx() {
  96. connected = false;
  97. zh = 0;
  98. }
  99. ~watchCtx() {
  100. if (zh) {
  101. zookeeper_close(zh);
  102. zh = 0;
  103. }
  104. }
  105. evt_t getEvent() {
  106. evt_t evt;
  107. mutex.acquire();
  108. CPPUNIT_ASSERT( events.size() > 0);
  109. evt = events.front();
  110. events.pop_front();
  111. mutex.release();
  112. return evt;
  113. }
  114. int countEvents() {
  115. int count;
  116. mutex.acquire();
  117. count = events.size();
  118. mutex.release();
  119. return count;
  120. }
  121. void putEvent(evt_t evt) {
  122. mutex.acquire();
  123. events.push_back(evt);
  124. mutex.release();
  125. }
  126. bool waitForConnected(zhandle_t *zh) {
  127. time_t expires = time(0) + 10;
  128. while(!connected && time(0) < expires) {
  129. yield(zh, 1);
  130. }
  131. return connected;
  132. }
  133. bool waitForDisconnected(zhandle_t *zh) {
  134. time_t expires = time(0) + 15;
  135. while(connected && time(0) < expires) {
  136. yield(zh, 1);
  137. }
  138. return !connected;
  139. }
  140. } watchctx_t;
  141. class Zookeeper_simpleSystem : public CPPUNIT_NS::TestFixture
  142. {
  143. CPPUNIT_TEST_SUITE(Zookeeper_simpleSystem);
  144. CPPUNIT_TEST(testAsyncWatcherAutoReset);
  145. #ifdef THREADED
  146. CPPUNIT_TEST(testNullData);
  147. CPPUNIT_TEST(testPathValidation);
  148. CPPUNIT_TEST(testPing);
  149. CPPUNIT_TEST(testAcl);
  150. CPPUNIT_TEST(testAuth);
  151. CPPUNIT_TEST(testWatcherAutoResetWithGlobal);
  152. CPPUNIT_TEST(testWatcherAutoResetWithLocal);
  153. #endif
  154. CPPUNIT_TEST_SUITE_END();
  155. static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
  156. watchctx_t *ctx = (watchctx_t*)v;
  157. if (state == ZOO_CONNECTED_STATE) {
  158. ctx->connected = true;
  159. } else {
  160. ctx->connected = false;
  161. }
  162. if (type != ZOO_SESSION_EVENT) {
  163. evt_t evt;
  164. evt.path = path;
  165. evt.type = type;
  166. ctx->putEvent(evt);
  167. }
  168. }
  169. static const char hostPorts[];
  170. const char *getHostPorts() {
  171. return hostPorts;
  172. }
  173. zhandle_t *createClient(watchctx_t *ctx) {
  174. zhandle_t *zk = zookeeper_init(hostPorts, watcher, 10000, 0,
  175. ctx, 0);
  176. ctx->zh = zk;
  177. sleep(1);
  178. return zk;
  179. }
  180. public:
  181. void setUp()
  182. {
  183. char cmd[1024];
  184. sprintf(cmd, "%s startClean %s", ZKSERVER_CMD, getHostPorts());
  185. CPPUNIT_ASSERT(system(cmd) == 0);
  186. }
  187. void startServer() {
  188. char cmd[1024];
  189. sprintf(cmd, "%s start %s", ZKSERVER_CMD, getHostPorts());
  190. CPPUNIT_ASSERT(system(cmd) == 0);
  191. }
  192. void stopServer() {
  193. tearDown();
  194. }
  195. void tearDown()
  196. {
  197. char cmd[1024];
  198. sprintf(cmd, "%s stop %s", ZKSERVER_CMD, getHostPorts());
  199. CPPUNIT_ASSERT(system(cmd) == 0);
  200. }
  201. void testPing()
  202. {
  203. watchctx_t ctxIdle;
  204. watchctx_t ctxWC;
  205. zhandle_t *zkIdle = createClient(&ctxIdle);
  206. zhandle_t *zkWatchCreator = createClient(&ctxWC);
  207. int rc;
  208. char path[80];
  209. CPPUNIT_ASSERT(zkIdle);
  210. CPPUNIT_ASSERT(zkWatchCreator);
  211. for(int i = 0; i < 30; i++) {
  212. sprintf(path, "/%i", i);
  213. rc = zoo_create(zkWatchCreator, path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
  214. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  215. }
  216. for(int i = 0; i < 30; i++) {
  217. sprintf(path, "/%i", i);
  218. struct Stat stat;
  219. rc = zoo_exists(zkIdle, path, 1, &stat);
  220. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  221. }
  222. for(int i = 0; i < 30; i++) {
  223. sprintf(path, "/%i", i);
  224. usleep(500000);
  225. rc = zoo_delete(zkWatchCreator, path, -1);
  226. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  227. }
  228. struct Stat stat;
  229. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, zoo_exists(zkIdle, "/0", 0, &stat));
  230. }
  231. bool waitForEvent(zhandle_t *zh, watchctx_t *ctx, int seconds) {
  232. time_t expires = time(0) + seconds;
  233. while(ctx->countEvents() == 0 && time(0) < expires) {
  234. yield(zh, 1);
  235. }
  236. return ctx->countEvents() > 0;
  237. }
  238. #define COUNT 100
  239. static zhandle_t *async_zk;
  240. static volatile int count;
  241. static void statCompletion(int rc, const struct Stat *stat, const void *data) {
  242. int tmp = (int) (long) data;
  243. CPPUNIT_ASSERT_EQUAL(tmp, rc);
  244. }
  245. static void stringCompletion(int rc, const char *value, const void *data) {
  246. char *path = (char*)data;
  247. if (rc == ZCONNECTIONLOSS && path) {
  248. // Try again
  249. rc = zoo_acreate(async_zk, path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, stringCompletion, 0);
  250. } else if (rc != ZOK) {
  251. // fprintf(stderr, "rc = %d with path = %s\n", rc, (path ? path : "null"));
  252. }
  253. if (path) {
  254. free(path);
  255. }
  256. }
  257. static void waitForVoidCompletion(int seconds) {
  258. time_t expires = time(0) + seconds;
  259. while(count == 0 && time(0) < expires) {
  260. sleep(1);
  261. }
  262. count--;
  263. }
  264. static void voidCompletion(int rc, const void *data) {
  265. int tmp = (int) (long) data;
  266. CPPUNIT_ASSERT_EQUAL(tmp, rc);
  267. count++;
  268. }
  269. static void verifyCreateFails(const char *path, zhandle_t *zk) {
  270. CPPUNIT_ASSERT_EQUAL((int)ZBADARGUMENTS, zoo_create(zk,
  271. path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0));
  272. }
  273. static void verifyCreateOk(const char *path, zhandle_t *zk) {
  274. CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_create(zk,
  275. path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0));
  276. }
  277. static void verifyCreateFailsSeq(const char *path, zhandle_t *zk) {
  278. CPPUNIT_ASSERT_EQUAL((int)ZBADARGUMENTS, zoo_create(zk,
  279. path, "", 0, &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, 0, 0));
  280. }
  281. static void verifyCreateOkSeq(const char *path, zhandle_t *zk) {
  282. CPPUNIT_ASSERT_EQUAL((int)ZOK, zoo_create(zk,
  283. path, "", 0, &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, 0, 0));
  284. }
  285. /**
  286. returns false if the vectors dont match
  287. **/
  288. bool compareAcl(struct ACL_vector acl1, struct ACL_vector acl2) {
  289. if (acl1.count != acl2.count) {
  290. return false;
  291. }
  292. struct ACL *aclval1 = acl1.data;
  293. struct ACL *aclval2 = acl2.data;
  294. if (aclval1->perms != aclval2->perms) {
  295. return false;
  296. }
  297. struct Id id1 = aclval1->id;
  298. struct Id id2 = aclval2->id;
  299. if (strcmp(id1.scheme, id2.scheme) != 0) {
  300. return false;
  301. }
  302. if (strcmp(id1.id, id2.id) != 0) {
  303. return false;
  304. }
  305. return true;
  306. }
  307. void testAcl() {
  308. int rc;
  309. struct ACL_vector aclvec;
  310. struct Stat stat;
  311. watchctx_t ctx;
  312. zhandle_t *zk = createClient(&ctx);
  313. rc = zoo_create(zk, "/acl", "", 0,
  314. &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
  315. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  316. rc = zoo_get_acl(zk, "/acl", &aclvec, &stat );
  317. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  318. bool cmp = compareAcl(ZOO_OPEN_ACL_UNSAFE, aclvec);
  319. CPPUNIT_ASSERT_EQUAL(true, cmp);
  320. rc = zoo_set_acl(zk, "/acl", -1, &ZOO_READ_ACL_UNSAFE);
  321. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  322. rc = zoo_get_acl(zk, "/acl", &aclvec, &stat);
  323. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  324. cmp = compareAcl(ZOO_READ_ACL_UNSAFE, aclvec);
  325. CPPUNIT_ASSERT_EQUAL(true, cmp);
  326. }
  327. void testAuth() {
  328. int rc;
  329. count = 0;
  330. watchctx_t ctx1, ctx2, ctx3;
  331. zhandle_t *zk = createClient(&ctx1);
  332. struct ACL_vector nodeAcl;
  333. struct ACL acl_val;
  334. rc = zoo_add_auth(0, "", 0, 0, voidCompletion, (void*)-1);
  335. CPPUNIT_ASSERT_EQUAL((int) ZBADARGUMENTS, rc);
  336. rc = zoo_add_auth(zk, 0, 0, 0, voidCompletion, (void*)-1);
  337. CPPUNIT_ASSERT_EQUAL((int) ZBADARGUMENTS, rc);
  338. // auth as pat, create /tauth1, close session
  339. rc = zoo_add_auth(zk, "digest", "pat:passwd", 10, voidCompletion,
  340. (void*)ZOK);
  341. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  342. waitForVoidCompletion(3);
  343. CPPUNIT_ASSERT(count == 0);
  344. rc = zoo_create(zk, "/tauth1", "", 0, &ZOO_CREATOR_ALL_ACL, 0, 0, 0);
  345. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  346. zk = createClient(&ctx2);
  347. rc = zoo_add_auth(zk, "digest", "pat:passwd2", 11, voidCompletion,
  348. (void*)ZOK);
  349. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  350. waitForVoidCompletion(3);
  351. CPPUNIT_ASSERT(count == 0);
  352. char buf[1024];
  353. int blen = sizeof(buf);
  354. struct Stat stat;
  355. rc = zoo_get(zk, "/tauth1", 0, buf, &blen, &stat);
  356. CPPUNIT_ASSERT_EQUAL((int)ZNOAUTH, rc);
  357. // add auth pat w/correct pass verify success
  358. rc = zoo_add_auth(zk, "digest", "pat:passwd", 10, voidCompletion,
  359. (void*)ZOK);
  360. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  361. rc = zoo_get(zk, "/tauth1", 0, buf, &blen, &stat);
  362. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  363. waitForVoidCompletion(3);
  364. CPPUNIT_ASSERT(count == 0);
  365. //create a new client
  366. zk = createClient(&ctx3);
  367. rc = zoo_add_auth(zk, "digest", "pat:passwd", 10, voidCompletion, (void*) ZOK);
  368. waitForVoidCompletion(3);
  369. CPPUNIT_ASSERT(count == 0);
  370. rc = zoo_add_auth(zk, "ip", "none", 4, voidCompletion, (void*)ZOK);
  371. //make the server forget the auths
  372. waitForVoidCompletion(3);
  373. CPPUNIT_ASSERT(count == 0);
  374. stopServer();
  375. CPPUNIT_ASSERT(ctx3.waitForDisconnected(zk));
  376. startServer();
  377. CPPUNIT_ASSERT(ctx3.waitForConnected(zk));
  378. // now try getting the data
  379. rc = zoo_get(zk, "/tauth1", 0, buf, &blen, &stat);
  380. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  381. // also check for get
  382. rc = zoo_get_acl(zk, "/", &nodeAcl, &stat);
  383. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  384. // check if the acl has all the perms
  385. CPPUNIT_ASSERT_EQUAL((int)1, nodeAcl.count);
  386. acl_val = *(nodeAcl.data);
  387. CPPUNIT_ASSERT_EQUAL((int) acl_val.perms, ZOO_PERM_ALL);
  388. // verify on root node
  389. rc = zoo_set_acl(zk, "/", -1, &ZOO_CREATOR_ALL_ACL);
  390. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  391. rc = zoo_set_acl(zk, "/", -1, &ZOO_OPEN_ACL_UNSAFE);
  392. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  393. }
  394. void testNullData() {
  395. watchctx_t ctx;
  396. zhandle_t *zk = createClient(&ctx);
  397. CPPUNIT_ASSERT(zk);
  398. int rc = 0;
  399. rc = zoo_create(zk, "/mahadev", NULL, -1,
  400. &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
  401. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  402. char buffer[512];
  403. struct Stat stat;
  404. int len = 512;
  405. rc = zoo_wget(zk, "/mahadev", NULL, NULL, buffer, &len, &stat);
  406. CPPUNIT_ASSERT_EQUAL( -1, len);
  407. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  408. rc = zoo_set(zk, "/mahadev", NULL, -1, -1);
  409. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  410. rc = zoo_wget(zk, "/mahadev", NULL, NULL, buffer, &len, &stat);
  411. CPPUNIT_ASSERT_EQUAL( -1, len);
  412. CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
  413. }
  414. void testPathValidation() {
  415. watchctx_t ctx;
  416. zhandle_t *zk = createClient(&ctx);
  417. CPPUNIT_ASSERT(zk);
  418. verifyCreateFails(0, zk);
  419. verifyCreateFails("", zk);
  420. verifyCreateFails("//", zk);
  421. verifyCreateFails("///", zk);
  422. verifyCreateFails("////", zk);
  423. verifyCreateFails("/.", zk);
  424. verifyCreateFails("/..", zk);
  425. verifyCreateFails("/./", zk);
  426. verifyCreateFails("/../", zk);
  427. verifyCreateFails("/foo/./", zk);
  428. verifyCreateFails("/foo/../", zk);
  429. verifyCreateFails("/foo/.", zk);
  430. verifyCreateFails("/foo/..", zk);
  431. verifyCreateFails("/./.", zk);
  432. verifyCreateFails("/../..", zk);
  433. verifyCreateFails("/foo/bar/", zk);
  434. verifyCreateFails("/foo//bar", zk);
  435. verifyCreateFails("/foo/bar//", zk);
  436. verifyCreateFails("foo", zk);
  437. verifyCreateFails("a", zk);
  438. // verify that trailing fails, except for seq which adds suffix
  439. verifyCreateOk("/createseq", zk);
  440. verifyCreateFails("/createseq/", zk);
  441. verifyCreateOkSeq("/createseq/", zk);
  442. verifyCreateOkSeq("/createseq/.", zk);
  443. verifyCreateOkSeq("/createseq/..", zk);
  444. verifyCreateFailsSeq("/createseq//", zk);
  445. verifyCreateFailsSeq("/createseq/./", zk);
  446. verifyCreateFailsSeq("/createseq/../", zk);
  447. verifyCreateOk("/.foo", zk);
  448. verifyCreateOk("/.f.", zk);
  449. verifyCreateOk("/..f", zk);
  450. verifyCreateOk("/..f..", zk);
  451. verifyCreateOk("/f.c", zk);
  452. verifyCreateOk("/f", zk);
  453. verifyCreateOk("/f/.f", zk);
  454. verifyCreateOk("/f/f.", zk);
  455. verifyCreateOk("/f/..f", zk);
  456. verifyCreateOk("/f/f..", zk);
  457. verifyCreateOk("/f/.f/f", zk);
  458. verifyCreateOk("/f/f./f", zk);
  459. }
  460. void testAsyncWatcherAutoReset()
  461. {
  462. watchctx_t ctx;
  463. zhandle_t *zk = createClient(&ctx);
  464. watchctx_t lctx[COUNT];
  465. int i;
  466. char path[80];
  467. int rc;
  468. evt_t evt;
  469. async_zk = zk;
  470. for(i = 0; i < COUNT; i++) {
  471. sprintf(path, "/%d", i);
  472. rc = zoo_awexists(zk, path, watcher, &lctx[i], statCompletion, (void*)ZNONODE);
  473. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  474. }
  475. yield(zk, 0);
  476. for(i = 0; i < COUNT/2; i++) {
  477. sprintf(path, "/%d", i);
  478. rc = zoo_acreate(zk, path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, stringCompletion, strdup(path));
  479. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  480. }
  481. yield(zk, 3);
  482. for(i = 0; i < COUNT/2; i++) {
  483. sprintf(path, "/%d", i);
  484. CPPUNIT_ASSERT_MESSAGE(path, waitForEvent(zk, &lctx[i], 5));
  485. evt = lctx[i].getEvent();
  486. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path.c_str(), ZOO_CREATED_EVENT, evt.type);
  487. CPPUNIT_ASSERT_EQUAL(string(path), evt.path);
  488. }
  489. for(i = COUNT/2 + 1; i < COUNT*10; i++) {
  490. sprintf(path, "/%d", i);
  491. rc = zoo_acreate(zk, path, "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, stringCompletion, strdup(path));
  492. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  493. }
  494. yield(zk, 1);
  495. stopServer();
  496. CPPUNIT_ASSERT(ctx.waitForDisconnected(zk));
  497. startServer();
  498. CPPUNIT_ASSERT(ctx.waitForConnected(zk));
  499. yield(zk, 3);
  500. for(i = COUNT/2+1; i < COUNT; i++) {
  501. sprintf(path, "/%d", i);
  502. CPPUNIT_ASSERT_MESSAGE(path, waitForEvent(zk, &lctx[i], 5));
  503. evt = lctx[i].getEvent();
  504. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_CREATED_EVENT, evt.type);
  505. CPPUNIT_ASSERT_EQUAL(string(path), evt.path);
  506. }
  507. }
  508. void testWatcherAutoReset(zhandle_t *zk, watchctx_t *ctxGlobal,
  509. watchctx_t *ctxLocal)
  510. {
  511. bool isGlobal = (ctxGlobal == ctxLocal);
  512. int rc;
  513. struct Stat stat;
  514. char buf[1024];
  515. int blen;
  516. struct String_vector strings;
  517. const char *testName;
  518. rc = zoo_create(zk, "/watchtest", "", 0,
  519. &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
  520. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  521. rc = zoo_create(zk, "/watchtest/child", "", 0,
  522. &ZOO_OPEN_ACL_UNSAFE, ZOO_EPHEMERAL, 0, 0);
  523. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  524. if (isGlobal) {
  525. testName = "GlobalTest";
  526. rc = zoo_get_children(zk, "/watchtest", 1, &strings);
  527. deallocate_String_vector(&strings);
  528. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  529. blen = sizeof(buf);
  530. rc = zoo_get(zk, "/watchtest/child", 1, buf, &blen, &stat);
  531. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  532. rc = zoo_exists(zk, "/watchtest/child2", 1, &stat);
  533. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  534. } else {
  535. testName = "LocalTest";
  536. rc = zoo_wget_children(zk, "/watchtest", watcher, ctxLocal,
  537. &strings);
  538. deallocate_String_vector(&strings);
  539. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  540. blen = sizeof(buf);
  541. rc = zoo_wget(zk, "/watchtest/child", watcher, ctxLocal,
  542. buf, &blen, &stat);
  543. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  544. rc = zoo_wexists(zk, "/watchtest/child2", watcher, ctxLocal,
  545. &stat);
  546. CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
  547. }
  548. CPPUNIT_ASSERT(ctxLocal->countEvents() == 0);
  549. stopServer();
  550. CPPUNIT_ASSERT_MESSAGE(testName, ctxGlobal->waitForDisconnected(zk));
  551. startServer();
  552. CPPUNIT_ASSERT_MESSAGE(testName, ctxLocal->waitForConnected(zk));
  553. CPPUNIT_ASSERT(ctxLocal->countEvents() == 0);
  554. rc = zoo_set(zk, "/watchtest/child", "1", 1, -1);
  555. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  556. struct Stat stat1, stat2;
  557. rc = zoo_set2(zk, "/watchtest/child", "1", 1, -1, &stat1);
  558. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  559. CPPUNIT_ASSERT(stat1.version >= 0);
  560. rc = zoo_set2(zk, "/watchtest/child", "1", 1, stat1.version, &stat2);
  561. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  562. rc = zoo_set(zk, "/watchtest/child", "1", 1, stat2.version);
  563. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  564. rc = zoo_create(zk, "/watchtest/child2", "", 0,
  565. &ZOO_OPEN_ACL_UNSAFE, 0, 0, 0);
  566. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  567. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  568. evt_t evt = ctxLocal->getEvent();
  569. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_CHANGED_EVENT, evt.type);
  570. CPPUNIT_ASSERT_EQUAL(string("/watchtest/child"), evt.path);
  571. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  572. // The create will trigget the get children and the
  573. // exists watches
  574. evt = ctxLocal->getEvent();
  575. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_CREATED_EVENT, evt.type);
  576. CPPUNIT_ASSERT_EQUAL(string("/watchtest/child2"), evt.path);
  577. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  578. evt = ctxLocal->getEvent();
  579. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_CHILD_EVENT, evt.type);
  580. CPPUNIT_ASSERT_EQUAL(string("/watchtest"), evt.path);
  581. // Make sure Pings are giving us problems
  582. sleep(5);
  583. CPPUNIT_ASSERT(ctxLocal->countEvents() == 0);
  584. stopServer();
  585. CPPUNIT_ASSERT_MESSAGE(testName, ctxGlobal->waitForDisconnected(zk));
  586. startServer();
  587. CPPUNIT_ASSERT_MESSAGE(testName, ctxGlobal->waitForConnected(zk));
  588. if (isGlobal) {
  589. testName = "GlobalTest";
  590. rc = zoo_get_children(zk, "/watchtest", 1, &strings);
  591. deallocate_String_vector(&strings);
  592. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  593. blen = sizeof(buf);
  594. rc = zoo_get(zk, "/watchtest/child", 1, buf, &blen, &stat);
  595. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  596. rc = zoo_exists(zk, "/watchtest/child2", 1, &stat);
  597. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  598. } else {
  599. testName = "LocalTest";
  600. rc = zoo_wget_children(zk, "/watchtest", watcher, ctxLocal,
  601. &strings);
  602. deallocate_String_vector(&strings);
  603. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  604. blen = sizeof(buf);
  605. rc = zoo_wget(zk, "/watchtest/child", watcher, ctxLocal,
  606. buf, &blen, &stat);
  607. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  608. rc = zoo_wexists(zk, "/watchtest/child2", watcher, ctxLocal,
  609. &stat);
  610. CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
  611. }
  612. zoo_delete(zk, "/watchtest/child2", -1);
  613. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  614. evt = ctxLocal->getEvent();
  615. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_DELETED_EVENT, evt.type);
  616. CPPUNIT_ASSERT_EQUAL(string("/watchtest/child2"), evt.path);
  617. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  618. evt = ctxLocal->getEvent();
  619. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_CHILD_EVENT, evt.type);
  620. CPPUNIT_ASSERT_EQUAL(string("/watchtest"), evt.path);
  621. stopServer();
  622. CPPUNIT_ASSERT_MESSAGE(testName, ctxGlobal->waitForDisconnected(zk));
  623. startServer();
  624. CPPUNIT_ASSERT_MESSAGE(testName, ctxLocal->waitForConnected(zk));
  625. zoo_delete(zk, "/watchtest/child", -1);
  626. zoo_delete(zk, "/watchtest", -1);
  627. CPPUNIT_ASSERT_MESSAGE(testName, waitForEvent(zk, ctxLocal, 5));
  628. evt = ctxLocal->getEvent();
  629. CPPUNIT_ASSERT_EQUAL_MESSAGE(evt.path, ZOO_DELETED_EVENT, evt.type);
  630. CPPUNIT_ASSERT_EQUAL(string("/watchtest/child"), evt.path);
  631. // Make sure nothing is straggling
  632. sleep(1);
  633. CPPUNIT_ASSERT(ctxLocal->countEvents() == 0);
  634. }
  635. void testWatcherAutoResetWithGlobal()
  636. {
  637. watchctx_t ctx;
  638. zhandle_t *zk = createClient(&ctx);
  639. testWatcherAutoReset(zk, &ctx, &ctx);
  640. }
  641. void testWatcherAutoResetWithLocal()
  642. {
  643. watchctx_t ctx;
  644. watchctx_t lctx;
  645. zhandle_t *zk = createClient(&ctx);
  646. testWatcherAutoReset(zk, &ctx, &lctx);
  647. }
  648. };
  649. volatile int Zookeeper_simpleSystem::count;
  650. zhandle_t *Zookeeper_simpleSystem::async_zk;
  651. const char Zookeeper_simpleSystem::hostPorts[] = "127.0.0.1:22181";
  652. CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_simpleSystem);