You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

1585 lines
49 KiB

  1. /*
  2. * This file is part of GNUnet
  3. * Copyright (C) 2013 GNUnet e.V.
  4. *
  5. * GNUnet is free software: you can redistribute it and/or modify it
  6. * under the terms of the GNU Affero General Public License as published
  7. * by the Free Software Foundation, either version 3 of the License,
  8. * or (at your option) any later version.
  9. *
  10. * GNUnet is distributed in the hope that it will be useful, but
  11. * WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  13. * Affero General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU Affero General Public License
  16. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  17. SPDX-License-Identifier: AGPL3.0-or-later
  18. */
  19. /**
  20. * @file psyc/psyc_api.c
  21. * @brief PSYC service; high-level access to the PSYC protocol
  22. * note that clients of this API are NOT expected to
  23. * understand the PSYC message format, only the semantics!
  24. * Parsing (and serializing) the PSYC stream format is done
  25. * within the implementation of the libgnunetpsyc library,
  26. * and this API deliberately exposes as little as possible
  27. * of the actual data stream format to the application!
  28. * @author Gabor X Toth
  29. */
  30. #include <inttypes.h>
  31. #include "platform.h"
  32. #include "gnunet_util_lib.h"
  33. #include "gnunet_multicast_service.h"
  34. #include "gnunet_psyc_service.h"
  35. #include "gnunet_psyc_util_lib.h"
  36. #include "psyc.h"
  37. #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
  38. /**
  39. * Handle to access PSYC channel operations for both the master and slaves.
  40. */
  41. struct GNUNET_PSYC_Channel
  42. {
  43. /**
  44. * Configuration to use.
  45. */
  46. const struct GNUNET_CONFIGURATION_Handle *cfg;
  47. /**
  48. * Client connection to the service.
  49. */
  50. struct GNUNET_MQ_Handle *mq;
  51. /**
  52. * Message to send on connect.
  53. */
  54. struct GNUNET_MQ_Envelope *connect_env;
  55. /**
  56. * Time to wait until we try to reconnect on failure.
  57. */
  58. struct GNUNET_TIME_Relative reconnect_delay;
  59. /**
  60. * Task for reconnecting when the listener fails.
  61. */
  62. struct GNUNET_SCHEDULER_Task *reconnect_task;
  63. /**
  64. * Async operations.
  65. */
  66. struct GNUNET_OP_Handle *op;
  67. /**
  68. * Transmission handle;
  69. */
  70. struct GNUNET_PSYC_TransmitHandle *tmit;
  71. /**
  72. * Receipt handle;
  73. */
  74. struct GNUNET_PSYC_ReceiveHandle *recv;
  75. /**
  76. * Function called after disconnected from the service.
  77. */
  78. GNUNET_ContinuationCallback disconnect_cb;
  79. /**
  80. * Closure for @a disconnect_cb.
  81. */
  82. void *disconnect_cls;
  83. /**
  84. * Are we polling for incoming messages right now?
  85. */
  86. uint8_t in_receive;
  87. /**
  88. * Is this a master or slave channel?
  89. */
  90. uint8_t is_master;
  91. /**
  92. * Is this channel in the process of disconnecting from the service?
  93. * #GNUNET_YES or #GNUNET_NO
  94. */
  95. uint8_t is_disconnecting;
  96. };
  97. /**
  98. * Handle for the master of a PSYC channel.
  99. */
  100. struct GNUNET_PSYC_Master
  101. {
  102. struct GNUNET_PSYC_Channel chn;
  103. GNUNET_PSYC_MasterStartCallback start_cb;
  104. /**
  105. * Join request callback.
  106. */
  107. GNUNET_PSYC_JoinRequestCallback join_req_cb;
  108. /**
  109. * Closure for the callbacks.
  110. */
  111. void *cb_cls;
  112. };
  113. /**
  114. * Handle for a PSYC channel slave.
  115. */
  116. struct GNUNET_PSYC_Slave
  117. {
  118. struct GNUNET_PSYC_Channel chn;
  119. GNUNET_PSYC_SlaveConnectCallback connect_cb;
  120. GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb;
  121. /**
  122. * Closure for the callbacks.
  123. */
  124. void *cb_cls;
  125. };
  126. /**
  127. * Handle that identifies a join request.
  128. *
  129. * Used to match calls to #GNUNET_PSYC_JoinRequestCallback to the
  130. * corresponding calls to GNUNET_PSYC_join_decision().
  131. */
  132. struct GNUNET_PSYC_JoinHandle
  133. {
  134. struct GNUNET_PSYC_Master *mst;
  135. struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
  136. };
  137. /**
  138. * Handle for a pending PSYC transmission operation.
  139. */
  140. struct GNUNET_PSYC_SlaveTransmitHandle
  141. {
  142. };
  143. struct GNUNET_PSYC_HistoryRequest
  144. {
  145. /**
  146. * Channel.
  147. */
  148. struct GNUNET_PSYC_Channel *chn;
  149. /**
  150. * Operation ID.
  151. */
  152. uint64_t op_id;
  153. /**
  154. * Message handler.
  155. */
  156. struct GNUNET_PSYC_ReceiveHandle *recv;
  157. /**
  158. * Function to call when the operation finished.
  159. */
  160. GNUNET_ResultCallback result_cb;
  161. /**
  162. * Closure for @a result_cb.
  163. */
  164. void *cls;
  165. };
  166. struct GNUNET_PSYC_StateRequest
  167. {
  168. /**
  169. * Channel.
  170. */
  171. struct GNUNET_PSYC_Channel *chn;
  172. /**
  173. * Operation ID.
  174. */
  175. uint64_t op_id;
  176. /**
  177. * State variable result callback.
  178. */
  179. GNUNET_PSYC_StateVarCallback var_cb;
  180. /**
  181. * Function to call when the operation finished.
  182. */
  183. GNUNET_ResultCallback result_cb;
  184. /**
  185. * Closure for @a result_cb.
  186. */
  187. void *cls;
  188. };
  189. static int
  190. check_channel_result (void *cls,
  191. const struct GNUNET_OperationResultMessage *res)
  192. {
  193. return GNUNET_OK;
  194. }
  195. static void
  196. handle_channel_result (void *cls,
  197. const struct GNUNET_OperationResultMessage *res)
  198. {
  199. struct GNUNET_PSYC_Channel *chn = cls;
  200. uint16_t size = ntohs (res->header.size);
  201. if (size < sizeof (*res))
  202. { /* Error, message too small. */
  203. GNUNET_break (0);
  204. return;
  205. }
  206. uint16_t data_size = size - sizeof (*res);
  207. const char *data = (0 < data_size) ? (void *) &res[1] : NULL;
  208. GNUNET_OP_result (chn->op, GNUNET_ntohll (res->op_id),
  209. GNUNET_ntohll (res->result_code),
  210. data, data_size, NULL);
  211. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  212. "handle_channel_result: Received result message with OP ID %" PRIu64 "\n",
  213. GNUNET_ntohll (res->op_id));
  214. }
  215. static void
  216. op_recv_history_result (void *cls, int64_t result,
  217. const void *data, uint16_t data_size)
  218. {
  219. LOG (GNUNET_ERROR_TYPE_DEBUG,
  220. "Received history replay result: %" PRId64 ".\n", result);
  221. struct GNUNET_PSYC_HistoryRequest *hist = cls;
  222. if (NULL != hist->result_cb)
  223. hist->result_cb (hist->cls, result, data, data_size);
  224. GNUNET_PSYC_receive_destroy (hist->recv);
  225. GNUNET_free (hist);
  226. }
  227. static void
  228. op_recv_state_result (void *cls, int64_t result,
  229. const void *data, uint16_t data_size)
  230. {
  231. LOG (GNUNET_ERROR_TYPE_DEBUG,
  232. "Received state request result: %" PRId64 ".\n", result);
  233. struct GNUNET_PSYC_StateRequest *sr = cls;
  234. if (NULL != sr->result_cb)
  235. sr->result_cb (sr->cls, result, data, data_size);
  236. GNUNET_free (sr);
  237. }
  238. static int
  239. check_channel_history_result (void *cls,
  240. const struct GNUNET_OperationResultMessage *res)
  241. {
  242. struct GNUNET_PSYC_MessageHeader *
  243. pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
  244. uint16_t size = ntohs (res->header.size);
  245. if ( (NULL == pmsg) ||
  246. (size < sizeof (*res) + sizeof (*pmsg)) )
  247. { /* Error, message too small. */
  248. GNUNET_break_op (0);
  249. return GNUNET_SYSERR;
  250. }
  251. return GNUNET_OK;
  252. }
  253. static void
  254. handle_channel_history_result (void *cls,
  255. const struct GNUNET_OperationResultMessage *res)
  256. {
  257. struct GNUNET_PSYC_Channel *chn = cls;
  258. struct GNUNET_PSYC_MessageHeader *
  259. pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
  260. GNUNET_ResultCallback result_cb = NULL;
  261. struct GNUNET_PSYC_HistoryRequest *hist = NULL;
  262. LOG (GNUNET_ERROR_TYPE_DEBUG,
  263. "%p Received historic fragment for message #%" PRIu64 ".\n",
  264. chn,
  265. GNUNET_ntohll (pmsg->message_id));
  266. if (GNUNET_YES != GNUNET_OP_get (chn->op,
  267. GNUNET_ntohll (res->op_id),
  268. &result_cb, (void *) &hist, NULL))
  269. { /* Operation not found. */
  270. LOG (GNUNET_ERROR_TYPE_WARNING,
  271. "%p Replay operation not found for historic fragment of message #%"
  272. PRIu64 ".\n",
  273. chn, GNUNET_ntohll (pmsg->message_id));
  274. return;
  275. }
  276. GNUNET_PSYC_receive_message (hist->recv,
  277. (const struct GNUNET_PSYC_MessageHeader *) pmsg);
  278. }
  279. static int
  280. check_channel_state_result (void *cls,
  281. const struct GNUNET_OperationResultMessage *res)
  282. {
  283. const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
  284. uint16_t mod_size;
  285. uint16_t size;
  286. if (NULL == mod)
  287. {
  288. GNUNET_break_op (0);
  289. return GNUNET_SYSERR;
  290. }
  291. mod_size = ntohs (mod->size);
  292. size = ntohs (res->header.size);
  293. if (size - sizeof (*res) != mod_size)
  294. {
  295. GNUNET_break_op (0);
  296. return GNUNET_SYSERR;
  297. }
  298. return GNUNET_OK;
  299. }
  300. static void
  301. handle_channel_state_result (void *cls,
  302. const struct GNUNET_OperationResultMessage *res)
  303. {
  304. struct GNUNET_PSYC_Channel *chn = cls;
  305. GNUNET_ResultCallback result_cb = NULL;
  306. struct GNUNET_PSYC_StateRequest *sr = NULL;
  307. if (GNUNET_YES != GNUNET_OP_get (chn->op,
  308. GNUNET_ntohll (res->op_id),
  309. &result_cb, (void *) &sr, NULL))
  310. { /* Operation not found. */
  311. return;
  312. }
  313. const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
  314. if (NULL == mod)
  315. {
  316. GNUNET_break_op (0);
  317. return;
  318. }
  319. uint16_t mod_size = ntohs (mod->size);
  320. switch (ntohs (mod->type))
  321. {
  322. case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
  323. {
  324. const struct GNUNET_PSYC_MessageModifier *
  325. pmod = (const struct GNUNET_PSYC_MessageModifier *) mod;
  326. const char *name = (const char *) &pmod[1];
  327. uint16_t name_size = ntohs (pmod->name_size);
  328. if (0 == name_size
  329. || mod_size - sizeof (*pmod) < name_size
  330. || '\0' != name[name_size - 1])
  331. {
  332. GNUNET_break_op (0);
  333. return;
  334. }
  335. sr->var_cb (sr->cls, mod, name, name + name_size,
  336. ntohs (pmod->header.size) - sizeof (*pmod),
  337. ntohs (pmod->value_size));
  338. break;
  339. }
  340. case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
  341. sr->var_cb (sr->cls, mod, NULL, (const char *) &mod[1],
  342. mod_size - sizeof (*mod), 0);
  343. break;
  344. }
  345. }
  346. static int
  347. check_channel_message (void *cls,
  348. const struct GNUNET_PSYC_MessageHeader *pmsg)
  349. {
  350. return GNUNET_OK;
  351. }
  352. static void
  353. handle_channel_message (void *cls,
  354. const struct GNUNET_PSYC_MessageHeader *pmsg)
  355. {
  356. struct GNUNET_PSYC_Channel *chn = cls;
  357. GNUNET_PSYC_receive_message (chn->recv, pmsg);
  358. }
  359. static void
  360. handle_channel_message_ack (void *cls,
  361. const struct GNUNET_MessageHeader *msg)
  362. {
  363. struct GNUNET_PSYC_Channel *chn = cls;
  364. GNUNET_PSYC_transmit_got_ack (chn->tmit);
  365. }
  366. static void
  367. handle_master_start_ack (void *cls,
  368. const struct GNUNET_PSYC_CountersResultMessage *cres)
  369. {
  370. struct GNUNET_PSYC_Master *mst = cls;
  371. int32_t result = ntohl (cres->result_code);
  372. if (GNUNET_OK != result && GNUNET_NO != result)
  373. {
  374. LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master: %ld\n", result);
  375. GNUNET_break (0);
  376. /* FIXME: disconnect */
  377. }
  378. if (NULL != mst->start_cb)
  379. mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
  380. }
  381. static int
  382. check_master_join_request (void *cls,
  383. const struct GNUNET_PSYC_JoinRequestMessage *req)
  384. {
  385. if ( ((sizeof (*req) + sizeof (struct GNUNET_PSYC_Message)) <= ntohs (req->header.size)) &&
  386. (NULL == GNUNET_MQ_extract_nested_mh (req)) )
  387. {
  388. GNUNET_break_op (0);
  389. return GNUNET_SYSERR;
  390. }
  391. return GNUNET_OK;
  392. }
  393. static void
  394. handle_master_join_request (void *cls,
  395. const struct GNUNET_PSYC_JoinRequestMessage *req)
  396. {
  397. struct GNUNET_PSYC_Master *mst = cls;
  398. if (NULL == mst->join_req_cb)
  399. return;
  400. const struct GNUNET_PSYC_Message *join_msg = NULL;
  401. if (sizeof (*req) + sizeof (*join_msg) <= ntohs (req->header.size))
  402. {
  403. join_msg = (struct GNUNET_PSYC_Message *) GNUNET_MQ_extract_nested_mh (req);
  404. LOG (GNUNET_ERROR_TYPE_DEBUG,
  405. "Received join_msg of type %u and size %u.\n",
  406. ntohs (join_msg->header.type),
  407. ntohs (join_msg->header.size));
  408. }
  409. struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh));
  410. jh->mst = mst;
  411. jh->slave_pub_key = req->slave_pub_key;
  412. if (NULL != mst->join_req_cb)
  413. mst->join_req_cb (mst->cb_cls, req, &req->slave_pub_key, join_msg, jh);
  414. }
  415. static void
  416. handle_slave_join_ack (void *cls,
  417. const struct GNUNET_PSYC_CountersResultMessage *cres)
  418. {
  419. struct GNUNET_PSYC_Slave *slv = cls;
  420. int32_t result = ntohl (cres->result_code);
  421. if (GNUNET_YES != result && GNUNET_NO != result)
  422. {
  423. LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n");
  424. GNUNET_break (0);
  425. /* FIXME: disconnect */
  426. }
  427. if (NULL != slv->connect_cb)
  428. slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
  429. }
  430. static int
  431. check_slave_join_decision (void *cls,
  432. const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
  433. {
  434. return GNUNET_OK;
  435. }
  436. static void
  437. handle_slave_join_decision (void *cls,
  438. const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
  439. {
  440. struct GNUNET_PSYC_Slave *slv = cls;
  441. struct GNUNET_PSYC_Message *pmsg = NULL;
  442. if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg))
  443. pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1];
  444. if (NULL != slv->join_dcsn_cb)
  445. slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg);
  446. }
  447. static void
  448. channel_cleanup (struct GNUNET_PSYC_Channel *chn)
  449. {
  450. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  451. "cleaning up channel %p\n",
  452. chn);
  453. if (NULL != chn->tmit)
  454. {
  455. GNUNET_PSYC_transmit_destroy (chn->tmit);
  456. chn->tmit = NULL;
  457. }
  458. if (NULL != chn->recv)
  459. {
  460. GNUNET_PSYC_receive_destroy (chn->recv);
  461. chn->recv = NULL;
  462. }
  463. if (NULL != chn->connect_env)
  464. {
  465. GNUNET_MQ_discard (chn->connect_env);
  466. chn->connect_env = NULL;
  467. }
  468. if (NULL != chn->mq)
  469. {
  470. GNUNET_MQ_destroy (chn->mq);
  471. chn->mq = NULL;
  472. }
  473. if (NULL != chn->disconnect_cb)
  474. {
  475. chn->disconnect_cb (chn->disconnect_cls);
  476. chn->disconnect_cb = NULL;
  477. }
  478. GNUNET_free (chn);
  479. }
  480. static void
  481. handle_channel_part_ack (void *cls,
  482. const struct GNUNET_MessageHeader *msg)
  483. {
  484. struct GNUNET_PSYC_Channel *chn = cls;
  485. channel_cleanup (chn);
  486. }
  487. /*** MASTER ***/
  488. static void
  489. master_connect (struct GNUNET_PSYC_Master *mst);
  490. static void
  491. master_reconnect (void *cls)
  492. {
  493. master_connect (cls);
  494. }
  495. /**
  496. * Master client disconnected from service.
  497. *
  498. * Reconnect after backoff period.
  499. */
  500. static void
  501. master_disconnected (void *cls, enum GNUNET_MQ_Error error)
  502. {
  503. struct GNUNET_PSYC_Master *mst = cls;
  504. struct GNUNET_PSYC_Channel *chn = &mst->chn;
  505. LOG (GNUNET_ERROR_TYPE_DEBUG,
  506. "Master client disconnected (%d), re-connecting\n",
  507. (int) error);
  508. if (NULL != chn->tmit)
  509. {
  510. GNUNET_PSYC_transmit_destroy (chn->tmit);
  511. chn->tmit = NULL;
  512. }
  513. if (NULL != chn->mq)
  514. {
  515. GNUNET_MQ_destroy (chn->mq);
  516. chn->mq = NULL;
  517. }
  518. chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
  519. master_reconnect,
  520. mst);
  521. chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
  522. }
  523. static void
  524. master_connect (struct GNUNET_PSYC_Master *mst)
  525. {
  526. struct GNUNET_PSYC_Channel *chn = &mst->chn;
  527. struct GNUNET_MQ_MessageHandler handlers[] = {
  528. GNUNET_MQ_hd_fixed_size (master_start_ack,
  529. GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK,
  530. struct GNUNET_PSYC_CountersResultMessage,
  531. mst),
  532. GNUNET_MQ_hd_var_size (master_join_request,
  533. GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
  534. struct GNUNET_PSYC_JoinRequestMessage,
  535. mst),
  536. GNUNET_MQ_hd_fixed_size (channel_part_ack,
  537. GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
  538. struct GNUNET_MessageHeader,
  539. chn),
  540. GNUNET_MQ_hd_var_size (channel_message,
  541. GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
  542. struct GNUNET_PSYC_MessageHeader,
  543. chn),
  544. GNUNET_MQ_hd_fixed_size (channel_message_ack,
  545. GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
  546. struct GNUNET_MessageHeader,
  547. chn),
  548. GNUNET_MQ_hd_var_size (channel_history_result,
  549. GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
  550. struct GNUNET_OperationResultMessage,
  551. chn),
  552. GNUNET_MQ_hd_var_size (channel_state_result,
  553. GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
  554. struct GNUNET_OperationResultMessage,
  555. chn),
  556. GNUNET_MQ_hd_var_size (channel_result,
  557. GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
  558. struct GNUNET_OperationResultMessage,
  559. chn),
  560. GNUNET_MQ_handler_end ()
  561. };
  562. chn->mq = GNUNET_CLIENT_connect (chn->cfg,
  563. "psyc",
  564. handlers,
  565. &master_disconnected,
  566. mst);
  567. GNUNET_assert (NULL != chn->mq);
  568. chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
  569. GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
  570. }
  571. /**
  572. * Start a PSYC master channel.
  573. *
  574. * Will start a multicast group identified by the given ECC key. Messages
  575. * received from group members will be given to the respective handler methods.
  576. * If a new member wants to join a group, the "join" method handler will be
  577. * invoked; the join handler must then generate a "join" message to approve the
  578. * joining of the new member. The channel can also change group membership
  579. * without explicit requests. Note that PSYC doesn't itself "understand" join
  580. * or part messages, the respective methods must call other PSYC functions to
  581. * inform PSYC about the meaning of the respective events.
  582. *
  583. * @param cfg Configuration to use (to connect to PSYC service).
  584. * @param channel_key ECC key that will be used to sign messages for this
  585. * PSYC session. The public key is used to identify the PSYC channel.
  586. * Note that end-users will usually not use the private key directly, but
  587. * rather look it up in GNS for places managed by other users, or select
  588. * a file with the private key(s) when setting up their own channels
  589. * FIXME: we'll likely want to use NOT the p521 curve here, but a cheaper
  590. * one in the future.
  591. * @param policy Channel policy specifying join and history restrictions.
  592. * Used to automate join decisions.
  593. * @param message_cb Function to invoke on message parts received from slaves.
  594. * @param join_request_cb Function to invoke when a slave wants to join.
  595. * @param master_start_cb Function to invoke after the channel master started.
  596. * @param cls Closure for @a method and @a join_cb.
  597. *
  598. * @return Handle for the channel master, NULL on error.
  599. */
  600. struct GNUNET_PSYC_Master *
  601. GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
  602. const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key,
  603. enum GNUNET_PSYC_Policy policy,
  604. GNUNET_PSYC_MasterStartCallback start_cb,
  605. GNUNET_PSYC_JoinRequestCallback join_request_cb,
  606. GNUNET_PSYC_MessageCallback message_cb,
  607. GNUNET_PSYC_MessagePartCallback message_part_cb,
  608. void *cls)
  609. {
  610. struct GNUNET_PSYC_Master *mst = GNUNET_new (struct GNUNET_PSYC_Master);
  611. struct GNUNET_PSYC_Channel *chn = &mst->chn;
  612. struct MasterStartRequest *req;
  613. chn->connect_env = GNUNET_MQ_msg (req,
  614. GNUNET_MESSAGE_TYPE_PSYC_MASTER_START);
  615. req->channel_key = *channel_key;
  616. req->policy = policy;
  617. chn->cfg = cfg;
  618. chn->is_master = GNUNET_YES;
  619. chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
  620. chn->op = GNUNET_OP_create ();
  621. chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
  622. mst->start_cb = start_cb;
  623. mst->join_req_cb = join_request_cb;
  624. mst->cb_cls = cls;
  625. master_connect (mst);
  626. return mst;
  627. }
  628. /**
  629. * Stop a PSYC master channel.
  630. *
  631. * @param master PSYC channel master to stop.
  632. * @param keep_active FIXME
  633. */
  634. void
  635. GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst,
  636. int keep_active,
  637. GNUNET_ContinuationCallback stop_cb,
  638. void *stop_cls)
  639. {
  640. struct GNUNET_PSYC_Channel *chn = &mst->chn;
  641. struct GNUNET_MQ_Envelope *env;
  642. chn->is_disconnecting = GNUNET_YES;
  643. chn->disconnect_cb = stop_cb;
  644. chn->disconnect_cls = stop_cls;
  645. env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
  646. GNUNET_MQ_send (chn->mq, env);
  647. }
  648. /**
  649. * Function to call with the decision made for a join request.
  650. *
  651. * Must be called once and only once in response to an invocation of the
  652. * #GNUNET_PSYC_JoinCallback.
  653. *
  654. * @param jh Join request handle.
  655. * @param is_admitted #GNUNET_YES if the join is approved,
  656. * #GNUNET_NO if it is disapproved,
  657. * #GNUNET_SYSERR if we cannot answer the request.
  658. * @param relay_count Number of relays given.
  659. * @param relays Array of suggested peers that might be useful relays to use
  660. * when joining the multicast group (essentially a list of peers that
  661. * are already part of the multicast group and might thus be willing
  662. * to help with routing). If empty, only this local peer (which must
  663. * be the multicast origin) is a good candidate for building the
  664. * multicast tree. Note that it is unnecessary to specify our own
  665. * peer identity in this array.
  666. * @param join_resp Application-dependent join response message.
  667. *
  668. * @return #GNUNET_OK on success,
  669. * #GNUNET_SYSERR if the message is too large.
  670. */
  671. int
  672. GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
  673. int is_admitted,
  674. uint32_t relay_count,
  675. const struct GNUNET_PeerIdentity *relays,
  676. const struct GNUNET_PSYC_Message *join_resp)
  677. {
  678. struct GNUNET_PSYC_Channel *chn = &jh->mst->chn;
  679. struct GNUNET_PSYC_JoinDecisionMessage *dcsn;
  680. uint16_t join_resp_size
  681. = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0;
  682. uint16_t relay_size = relay_count * sizeof (*relays);
  683. if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
  684. < sizeof (*dcsn) + relay_size + join_resp_size)
  685. return GNUNET_SYSERR;
  686. struct GNUNET_MQ_Envelope *
  687. env = GNUNET_MQ_msg_extra (dcsn, relay_size + join_resp_size,
  688. GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
  689. dcsn->is_admitted = htonl (is_admitted);
  690. dcsn->slave_pub_key = jh->slave_pub_key;
  691. if (0 < join_resp_size)
  692. GNUNET_memcpy (&dcsn[1], join_resp, join_resp_size);
  693. GNUNET_MQ_send (chn->mq, env);
  694. GNUNET_free (jh);
  695. return GNUNET_OK;
  696. }
  697. /**
  698. * Send a message to call a method to all members in the PSYC channel.
  699. *
  700. * @param master Handle to the PSYC channel.
  701. * @param method_name Which method should be invoked.
  702. * @param notify_mod Function to call to obtain modifiers.
  703. * @param notify_data Function to call to obtain fragments of the data.
  704. * @param notify_cls Closure for @a notify_mod and @a notify_data.
  705. * @param flags Flags for the message being transmitted.
  706. *
  707. * @return Transmission handle, NULL on error (i.e. more than one request queued).
  708. */
  709. struct GNUNET_PSYC_MasterTransmitHandle *
  710. GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *mst,
  711. const char *method_name,
  712. GNUNET_PSYC_TransmitNotifyModifier notify_mod,
  713. GNUNET_PSYC_TransmitNotifyData notify_data,
  714. void *notify_cls,
  715. enum GNUNET_PSYC_MasterTransmitFlags flags)
  716. {
  717. if (GNUNET_OK
  718. == GNUNET_PSYC_transmit_message (mst->chn.tmit, method_name, NULL,
  719. notify_mod, notify_data, notify_cls,
  720. flags))
  721. return (struct GNUNET_PSYC_MasterTransmitHandle *) mst->chn.tmit;
  722. else
  723. return NULL;
  724. }
  725. /**
  726. * Resume transmission to the channel.
  727. *
  728. * @param tmit Handle of the request that is being resumed.
  729. */
  730. void
  731. GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
  732. {
  733. GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
  734. }
  735. /**
  736. * Abort transmission request to the channel.
  737. *
  738. * @param tmit Handle of the request that is being aborted.
  739. */
  740. void
  741. GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
  742. {
  743. GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
  744. }
  745. /**
  746. * Convert a channel @a master to a @e channel handle to access the @e channel
  747. * APIs.
  748. *
  749. * @param master Channel master handle.
  750. *
  751. * @return Channel handle, valid for as long as @a master is valid.
  752. */
  753. struct GNUNET_PSYC_Channel *
  754. GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
  755. {
  756. return &master->chn;
  757. }
  758. /*** SLAVE ***/
  759. static void
  760. slave_connect (struct GNUNET_PSYC_Slave *slv);
  761. static void
  762. slave_reconnect (void *cls)
  763. {
  764. slave_connect (cls);
  765. }
  766. /**
  767. * Slave client disconnected from service.
  768. *
  769. * Reconnect after backoff period.
  770. */
  771. static void
  772. slave_disconnected (void *cls,
  773. enum GNUNET_MQ_Error error)
  774. {
  775. struct GNUNET_PSYC_Slave *slv = cls;
  776. struct GNUNET_PSYC_Channel *chn = &slv->chn;
  777. LOG (GNUNET_ERROR_TYPE_DEBUG,
  778. "Slave client disconnected (%d), re-connecting\n",
  779. (int) error);
  780. if (NULL != chn->tmit)
  781. {
  782. GNUNET_PSYC_transmit_destroy (chn->tmit);
  783. chn->tmit = NULL;
  784. }
  785. if (NULL != chn->mq)
  786. {
  787. GNUNET_MQ_destroy (chn->mq);
  788. chn->mq = NULL;
  789. }
  790. chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
  791. &slave_reconnect,
  792. slv);
  793. chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
  794. }
  795. static void
  796. slave_connect (struct GNUNET_PSYC_Slave *slv)
  797. {
  798. struct GNUNET_PSYC_Channel *chn = &slv->chn;
  799. struct GNUNET_MQ_MessageHandler handlers[] = {
  800. GNUNET_MQ_hd_fixed_size (slave_join_ack,
  801. GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK,
  802. struct GNUNET_PSYC_CountersResultMessage,
  803. slv),
  804. GNUNET_MQ_hd_var_size (slave_join_decision,
  805. GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
  806. struct GNUNET_PSYC_JoinDecisionMessage,
  807. slv),
  808. GNUNET_MQ_hd_fixed_size (channel_part_ack,
  809. GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
  810. struct GNUNET_MessageHeader,
  811. chn),
  812. GNUNET_MQ_hd_var_size (channel_message,
  813. GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
  814. struct GNUNET_PSYC_MessageHeader,
  815. chn),
  816. GNUNET_MQ_hd_fixed_size (channel_message_ack,
  817. GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
  818. struct GNUNET_MessageHeader,
  819. chn),
  820. GNUNET_MQ_hd_var_size (channel_history_result,
  821. GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
  822. struct GNUNET_OperationResultMessage,
  823. chn),
  824. GNUNET_MQ_hd_var_size (channel_state_result,
  825. GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
  826. struct GNUNET_OperationResultMessage,
  827. chn),
  828. GNUNET_MQ_hd_var_size (channel_result,
  829. GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
  830. struct GNUNET_OperationResultMessage,
  831. chn),
  832. GNUNET_MQ_handler_end ()
  833. };
  834. chn->mq = GNUNET_CLIENT_connect (chn->cfg,
  835. "psyc",
  836. handlers,
  837. &slave_disconnected,
  838. slv);
  839. if (NULL == chn->mq)
  840. {
  841. chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
  842. &slave_reconnect,
  843. slv);
  844. chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
  845. return;
  846. }
  847. chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
  848. GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
  849. }
  850. /**
  851. * Join a PSYC channel.
  852. *
  853. * The entity joining is always the local peer. The user must immediately use
  854. * the GNUNET_PSYC_slave_transmit() functions to transmit a @e join_msg to the
  855. * channel; if the join request succeeds, the channel state (and @e recent
  856. * method calls) will be replayed to the joining member. There is no explicit
  857. * notification on failure (as the channel may simply take days to approve,
  858. * and disapproval is simply being ignored).
  859. *
  860. * @param cfg
  861. * Configuration to use.
  862. * @param channel_key ECC public key that identifies the channel we wish to join.
  863. * @param slave_key ECC private-public key pair that identifies the slave, and
  864. * used by multicast to sign the join request and subsequent unicast
  865. * requests sent to the master.
  866. * @param origin Peer identity of the origin.
  867. * @param relay_count Number of peers in the @a relays array.
  868. * @param relays Peer identities of members of the multicast group, which serve
  869. * as relays and used to join the group at.
  870. * @param message_cb Function to invoke on message parts received from the
  871. * channel, typically at least contains method handlers for @e join and
  872. * @e part.
  873. * @param slave_connect_cb Function invoked once we have connected to the
  874. * PSYC service.
  875. * @param join_decision_cb Function invoked once we have received a join
  876. * decision.
  877. * @param cls Closure for @a message_cb and @a slave_joined_cb.
  878. * @param method_name Method name for the join request.
  879. * @param env Environment containing transient variables for the request, or NULL.
  880. * @param data Payload for the join message.
  881. * @param data_size Number of bytes in @a data.
  882. *
  883. * @return Handle for the slave, NULL on error.
  884. */
  885. struct GNUNET_PSYC_Slave *
  886. GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
  887. const struct GNUNET_CRYPTO_EddsaPublicKey *channel_pub_key,
  888. const struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key,
  889. enum GNUNET_PSYC_SlaveJoinFlags flags,
  890. const struct GNUNET_PeerIdentity *origin,
  891. uint32_t relay_count,
  892. const struct GNUNET_PeerIdentity *relays,
  893. GNUNET_PSYC_MessageCallback message_cb,
  894. GNUNET_PSYC_MessagePartCallback message_part_cb,
  895. GNUNET_PSYC_SlaveConnectCallback connect_cb,
  896. GNUNET_PSYC_JoinDecisionCallback join_decision_cb,
  897. void *cls,
  898. const struct GNUNET_PSYC_Message *join_msg)
  899. {
  900. struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv));
  901. struct GNUNET_PSYC_Channel *chn = &slv->chn;
  902. uint16_t relay_size = relay_count * sizeof (*relays);
  903. uint16_t join_msg_size;
  904. if (NULL == join_msg)
  905. join_msg_size = 0;
  906. else
  907. join_msg_size = ntohs (join_msg->header.size);
  908. struct SlaveJoinRequest *req;
  909. chn->connect_env = GNUNET_MQ_msg_extra (req, relay_size + join_msg_size,
  910. GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
  911. req->channel_pub_key = *channel_pub_key;
  912. req->slave_key = *slave_key;
  913. req->origin = *origin;
  914. req->relay_count = htonl (relay_count);
  915. req->flags = htonl (flags);
  916. if (0 < relay_size)
  917. GNUNET_memcpy (&req[1], relays, relay_size);
  918. if (NULL != join_msg)
  919. GNUNET_memcpy ((char *) &req[1] + relay_size, join_msg, join_msg_size);
  920. chn->cfg = cfg;
  921. chn->is_master = GNUNET_NO;
  922. chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
  923. chn->op = GNUNET_OP_create ();
  924. chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
  925. slv->connect_cb = connect_cb;
  926. slv->join_dcsn_cb = join_decision_cb;
  927. slv->cb_cls = cls;
  928. slave_connect (slv);
  929. return slv;
  930. }
  931. /**
  932. * Part a PSYC channel.
  933. *
  934. * Will terminate the connection to the PSYC service. Polite clients should
  935. * first explicitly send a part request (via GNUNET_PSYC_slave_transmit()).
  936. *
  937. * @param slave Slave handle.
  938. */
  939. void
  940. GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv,
  941. int keep_active,
  942. GNUNET_ContinuationCallback part_cb,
  943. void *part_cls)
  944. {
  945. struct GNUNET_PSYC_Channel *chn = &slv->chn;
  946. struct GNUNET_MQ_Envelope *env;
  947. chn->is_disconnecting = GNUNET_YES;
  948. chn->disconnect_cb = part_cb;
  949. chn->disconnect_cls = part_cls;
  950. env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
  951. GNUNET_MQ_send (chn->mq, env);
  952. }
  953. /**
  954. * Request a message to be sent to the channel master.
  955. *
  956. * @param slave Slave handle.
  957. * @param method_name Which (PSYC) method should be invoked (on host).
  958. * @param notify_mod Function to call to obtain modifiers.
  959. * @param notify_data Function to call to obtain fragments of the data.
  960. * @param notify_cls Closure for @a notify.
  961. * @param flags Flags for the message being transmitted.
  962. *
  963. * @return Transmission handle, NULL on error (i.e. more than one request
  964. * queued).
  965. */
  966. struct GNUNET_PSYC_SlaveTransmitHandle *
  967. GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slv,
  968. const char *method_name,
  969. GNUNET_PSYC_TransmitNotifyModifier notify_mod,
  970. GNUNET_PSYC_TransmitNotifyData notify_data,
  971. void *notify_cls,
  972. enum GNUNET_PSYC_SlaveTransmitFlags flags)
  973. {
  974. if (GNUNET_OK
  975. == GNUNET_PSYC_transmit_message (slv->chn.tmit, method_name, NULL,
  976. notify_mod, notify_data, notify_cls,
  977. flags))
  978. return (struct GNUNET_PSYC_SlaveTransmitHandle *) slv->chn.tmit;
  979. else
  980. return NULL;
  981. }
  982. /**
  983. * Resume transmission to the master.
  984. *
  985. * @param tmit Handle of the request that is being resumed.
  986. */
  987. void
  988. GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
  989. {
  990. GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
  991. }
  992. /**
  993. * Abort transmission request to master.
  994. *
  995. * @param tmit Handle of the request that is being aborted.
  996. */
  997. void
  998. GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
  999. {
  1000. GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
  1001. }
  1002. /**
  1003. * Convert @a slave to a @e channel handle to access the @e channel APIs.
  1004. *
  1005. * @param slv Slave handle.
  1006. *
  1007. * @return Channel handle, valid for as long as @a slave is valid.
  1008. */
  1009. struct GNUNET_PSYC_Channel *
  1010. GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slv)
  1011. {
  1012. return &slv->chn;
  1013. }
  1014. /**
  1015. * Add a slave to the channel's membership list.
  1016. *
  1017. * Note that this will NOT generate any PSYC traffic, it will merely update the
  1018. * local database to modify how we react to <em>membership test</em> queries.
  1019. * The channel master still needs to explicitly transmit a @e join message to
  1020. * notify other channel members and they then also must still call this function
  1021. * in their respective methods handling the @e join message. This way, how @e
  1022. * join and @e part operations are exactly implemented is still up to the
  1023. * application; for example, there might be a @e part_all method to kick out
  1024. * everyone.
  1025. *
  1026. * Note that channel slaves are explicitly trusted to execute such methods
  1027. * correctly; not doing so correctly will result in either denying other slaves
  1028. * access or offering access to channel data to non-members.
  1029. *
  1030. * @param chn
  1031. * Channel handle.
  1032. * @param slave_pub_key
  1033. * Identity of channel slave to add.
  1034. * @param announced_at
  1035. * ID of the message that announced the membership change.
  1036. * @param effective_since
  1037. * Addition of slave is in effect since this message ID.
  1038. * @param result_cb
  1039. * Function to call with the result of the operation.
  1040. * The @e result_code argument is #GNUNET_OK on success, or
  1041. * #GNUNET_SYSERR on error. In case of an error, the @e data argument
  1042. * can contain an optional error message.
  1043. * @param cls
  1044. * Closure for @a result_cb.
  1045. */
  1046. void
  1047. GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
  1048. const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
  1049. uint64_t announced_at,
  1050. uint64_t effective_since,
  1051. GNUNET_ResultCallback result_cb,
  1052. void *cls)
  1053. {
  1054. struct ChannelMembershipStoreRequest *req;
  1055. struct GNUNET_MQ_Envelope *
  1056. env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
  1057. req->slave_pub_key = *slave_pub_key;
  1058. req->announced_at = GNUNET_htonll (announced_at);
  1059. req->effective_since = GNUNET_htonll (effective_since);
  1060. req->did_join = GNUNET_YES;
  1061. req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
  1062. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  1063. "GNUNET_PSYC_channel_slave_add, OP ID: %" PRIu64 "\n",
  1064. GNUNET_ntohll (req->op_id));
  1065. GNUNET_MQ_send (chn->mq, env);
  1066. }
  1067. /**
  1068. * Remove a slave from the channel's membership list.
  1069. *
  1070. * Note that this will NOT generate any PSYC traffic, it will merely update the
  1071. * local database to modify how we react to <em>membership test</em> queries.
  1072. * The channel master still needs to explicitly transmit a @e part message to
  1073. * notify other channel members and they then also must still call this function
  1074. * in their respective methods handling the @e part message. This way, how
  1075. * @e join and @e part operations are exactly implemented is still up to the
  1076. * application; for example, there might be a @e part_all message to kick out
  1077. * everyone.
  1078. *
  1079. * Note that channel members are explicitly trusted to perform these
  1080. * operations correctly; not doing so correctly will result in either
  1081. * denying members access or offering access to channel data to
  1082. * non-members.
  1083. *
  1084. * @param chn
  1085. * Channel handle.
  1086. * @param slave_pub_key
  1087. * Identity of channel slave to remove.
  1088. * @param announced_at
  1089. * ID of the message that announced the membership change.
  1090. * @param result_cb
  1091. * Function to call with the result of the operation.
  1092. * The @e result_code argument is #GNUNET_OK on success, or
  1093. * #GNUNET_SYSERR on error. In case of an error, the @e data argument
  1094. * can contain an optional error message.
  1095. * @param cls
  1096. * Closure for @a result_cb.
  1097. */
  1098. void
  1099. GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn,
  1100. const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
  1101. uint64_t announced_at,
  1102. GNUNET_ResultCallback result_cb,
  1103. void *cls)
  1104. {
  1105. struct ChannelMembershipStoreRequest *req;
  1106. struct GNUNET_MQ_Envelope *
  1107. env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
  1108. req->slave_pub_key = *slave_pub_key;
  1109. req->announced_at = GNUNET_htonll (announced_at);
  1110. req->did_join = GNUNET_NO;
  1111. req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
  1112. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  1113. "GNUNET_PSYC_channel_slave_remove, OP ID: %" PRIu64 "\n",
  1114. GNUNET_ntohll (req->op_id));
  1115. GNUNET_MQ_send (chn->mq, env);
  1116. }
  1117. static struct GNUNET_PSYC_HistoryRequest *
  1118. channel_history_replay (struct GNUNET_PSYC_Channel *chn,
  1119. uint64_t start_message_id,
  1120. uint64_t end_message_id,
  1121. uint64_t message_limit,
  1122. const char *method_prefix,
  1123. uint32_t flags,
  1124. GNUNET_PSYC_MessageCallback message_cb,
  1125. GNUNET_PSYC_MessagePartCallback message_part_cb,
  1126. GNUNET_ResultCallback result_cb,
  1127. void *cls)
  1128. {
  1129. struct GNUNET_PSYC_HistoryRequestMessage *req;
  1130. struct GNUNET_PSYC_HistoryRequest *hist = GNUNET_malloc (sizeof (*hist));
  1131. hist->chn = chn;
  1132. hist->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
  1133. hist->result_cb = result_cb;
  1134. hist->cls = cls;
  1135. hist->op_id = GNUNET_OP_add (chn->op, op_recv_history_result, hist, NULL);
  1136. GNUNET_assert (NULL != method_prefix);
  1137. uint16_t method_size = strnlen (method_prefix,
  1138. GNUNET_MAX_MESSAGE_SIZE
  1139. - sizeof (*req)) + 1;
  1140. GNUNET_assert ('\0' == method_prefix[method_size - 1]);
  1141. struct GNUNET_MQ_Envelope *
  1142. env = GNUNET_MQ_msg_extra (req, method_size,
  1143. GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY);
  1144. req->start_message_id = GNUNET_htonll (start_message_id);
  1145. req->end_message_id = GNUNET_htonll (end_message_id);
  1146. req->message_limit = GNUNET_htonll (message_limit);
  1147. req->flags = htonl (flags);
  1148. req->op_id = GNUNET_htonll (hist->op_id);
  1149. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  1150. "channel_history_replay, OP ID: %" PRIu64 "\n",
  1151. GNUNET_ntohll (req->op_id));
  1152. GNUNET_memcpy (&req[1], method_prefix, method_size);
  1153. GNUNET_MQ_send (chn->mq, env);
  1154. return hist;
  1155. }
  1156. /**
  1157. * Request to replay a part of the message history of the channel.
  1158. *
  1159. * Historic messages (but NOT the state at the time) will be replayed and given
  1160. * to the normal method handlers with a #GNUNET_PSYC_MESSAGE_HISTORIC flag set.
  1161. *
  1162. * Messages are retrieved from the local PSYCstore if available,
  1163. * otherwise requested from the network.
  1164. *
  1165. * @param channel
  1166. * Which channel should be replayed?
  1167. * @param start_message_id
  1168. * Earliest interesting point in history.
  1169. * @param end_message_id
  1170. * Last (inclusive) interesting point in history.
  1171. * @param method_prefix
  1172. * Retrieve only messages with a matching method prefix.
  1173. * @param flags
  1174. * OR'ed enum GNUNET_PSYC_HistoryReplayFlags
  1175. * @param result_cb
  1176. * Function to call when the requested history has been fully replayed.
  1177. * @param cls
  1178. * Closure for the callbacks.
  1179. *
  1180. * @return Handle to cancel history replay operation.
  1181. */
  1182. struct GNUNET_PSYC_HistoryRequest *
  1183. GNUNET_PSYC_channel_history_replay (struct GNUNET_PSYC_Channel *chn,
  1184. uint64_t start_message_id,
  1185. uint64_t end_message_id,
  1186. const char *method_prefix,
  1187. uint32_t flags,
  1188. GNUNET_PSYC_MessageCallback message_cb,
  1189. GNUNET_PSYC_MessagePartCallback message_part_cb,
  1190. GNUNET_ResultCallback result_cb,
  1191. void *cls)
  1192. {
  1193. return channel_history_replay (chn, start_message_id, end_message_id, 0,
  1194. method_prefix, flags,
  1195. message_cb, message_part_cb, result_cb, cls);
  1196. }
  1197. /**
  1198. * Request to replay the latest messages from the message history of the channel.
  1199. *
  1200. * Historic messages (but NOT the state at the time) will be replayed (given to
  1201. * the normal method handlers) if available and if access is permitted.
  1202. *
  1203. * @param channel
  1204. * Which channel should be replayed?
  1205. * @param message_limit
  1206. * Maximum number of messages to replay.
  1207. * @param method_prefix
  1208. * Retrieve only messages with a matching method prefix.
  1209. * Use NULL or "" to retrieve all.
  1210. * @param flags
  1211. * OR'ed enum GNUNET_PSYC_HistoryReplayFlags
  1212. * @param result_cb
  1213. * Function to call when the requested history has been fully replayed.
  1214. * @param cls
  1215. * Closure for the callbacks.
  1216. *
  1217. * @return Handle to cancel history replay operation.
  1218. */
  1219. struct GNUNET_PSYC_HistoryRequest *
  1220. GNUNET_PSYC_channel_history_replay_latest (struct GNUNET_PSYC_Channel *chn,
  1221. uint64_t message_limit,
  1222. const char *method_prefix,
  1223. uint32_t flags,
  1224. GNUNET_PSYC_MessageCallback message_cb,
  1225. GNUNET_PSYC_MessagePartCallback message_part_cb,
  1226. GNUNET_ResultCallback result_cb,
  1227. void *cls)
  1228. {
  1229. return channel_history_replay (chn, 0, 0, message_limit, method_prefix, flags,
  1230. message_cb, message_part_cb, result_cb, cls);
  1231. }
  1232. void
  1233. GNUNET_PSYC_channel_history_replay_cancel (struct GNUNET_PSYC_Channel *channel,
  1234. struct GNUNET_PSYC_HistoryRequest *hist)
  1235. {
  1236. GNUNET_PSYC_receive_destroy (hist->recv);
  1237. GNUNET_OP_remove (hist->chn->op, hist->op_id);
  1238. GNUNET_free (hist);
  1239. }
  1240. /**
  1241. * Retrieve the best matching channel state variable.
  1242. *
  1243. * If the requested variable name is not present in the state, the nearest
  1244. * less-specific name is matched; for example, requesting "_a_b" will match "_a"
  1245. * if "_a_b" does not exist.
  1246. *
  1247. * @param channel
  1248. * Channel handle.
  1249. * @param full_name
  1250. * Full name of the requested variable.
  1251. * The actual variable returned might have a shorter name.
  1252. * @param var_cb
  1253. * Function called once when a matching state variable is found.
  1254. * Not called if there's no matching state variable.
  1255. * @param result_cb
  1256. * Function called after the operation finished.
  1257. * (i.e. all state variables have been returned via @a state_cb)
  1258. * @param cls
  1259. * Closure for the callbacks.
  1260. */
  1261. static struct GNUNET_PSYC_StateRequest *
  1262. channel_state_get (struct GNUNET_PSYC_Channel *chn,
  1263. uint16_t type, const char *name,
  1264. GNUNET_PSYC_StateVarCallback var_cb,
  1265. GNUNET_ResultCallback result_cb, void *cls)
  1266. {
  1267. struct StateRequest *req;
  1268. struct GNUNET_PSYC_StateRequest *sr = GNUNET_malloc (sizeof (*sr));
  1269. sr->chn = chn;
  1270. sr->var_cb = var_cb;
  1271. sr->result_cb = result_cb;
  1272. sr->cls = cls;
  1273. sr->op_id = GNUNET_OP_add (chn->op, op_recv_state_result, sr, NULL);
  1274. GNUNET_assert (NULL != name);
  1275. size_t name_size = strnlen (name, GNUNET_MAX_MESSAGE_SIZE
  1276. - sizeof (*req)) + 1;
  1277. struct GNUNET_MQ_Envelope *
  1278. env = GNUNET_MQ_msg_extra (req, name_size, type);
  1279. req->op_id = GNUNET_htonll (sr->op_id);
  1280. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  1281. "channel_state_get, OP ID: %" PRIu64 "\n",
  1282. GNUNET_ntohll (req->op_id));
  1283. GNUNET_memcpy (&req[1], name, name_size);
  1284. GNUNET_MQ_send (chn->mq, env);
  1285. return sr;
  1286. }
  1287. /**
  1288. * Retrieve the best matching channel state variable.
  1289. *
  1290. * If the requested variable name is not present in the state, the nearest
  1291. * less-specific name is matched; for example, requesting "_a_b" will match "_a"
  1292. * if "_a_b" does not exist.
  1293. *
  1294. * @param channel
  1295. * Channel handle.
  1296. * @param full_name
  1297. * Full name of the requested variable.
  1298. * The actual variable returned might have a shorter name.
  1299. * @param var_cb
  1300. * Function called once when a matching state variable is found.
  1301. * Not called if there's no matching state variable.
  1302. * @param result_cb
  1303. * Function called after the operation finished.
  1304. * (i.e. all state variables have been returned via @a state_cb)
  1305. * @param cls
  1306. * Closure for the callbacks.
  1307. */
  1308. struct GNUNET_PSYC_StateRequest *
  1309. GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *chn,
  1310. const char *full_name,
  1311. GNUNET_PSYC_StateVarCallback var_cb,
  1312. GNUNET_ResultCallback result_cb,
  1313. void *cls)
  1314. {
  1315. return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET,
  1316. full_name, var_cb, result_cb, cls);
  1317. }
  1318. /**
  1319. * Return all channel state variables whose name matches a given prefix.
  1320. *
  1321. * A name matches if it starts with the given @a name_prefix, thus requesting
  1322. * the empty prefix ("") will match all values; requesting "_a_b" will also
  1323. * return values stored under "_a_b_c".
  1324. *
  1325. * The @a state_cb is invoked on all matching state variables asynchronously, as
  1326. * the state is stored in and retrieved from the PSYCstore,
  1327. *
  1328. * @param channel
  1329. * Channel handle.
  1330. * @param name_prefix
  1331. * Prefix of the state variable name to match.
  1332. * @param var_cb
  1333. * Function called once when a matching state variable is found.
  1334. * Not called if there's no matching state variable.
  1335. * @param result_cb
  1336. * Function called after the operation finished.
  1337. * (i.e. all state variables have been returned via @a state_cb)
  1338. * @param cls
  1339. * Closure for the callbacks.
  1340. */
  1341. struct GNUNET_PSYC_StateRequest *
  1342. GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *chn,
  1343. const char *name_prefix,
  1344. GNUNET_PSYC_StateVarCallback var_cb,
  1345. GNUNET_ResultCallback result_cb,
  1346. void *cls)
  1347. {
  1348. return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX,
  1349. name_prefix, var_cb, result_cb, cls);
  1350. }
  1351. /**
  1352. * Cancel a state request operation.
  1353. *
  1354. * @param sr
  1355. * Handle for the operation to cancel.
  1356. */
  1357. void
  1358. GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateRequest *sr)
  1359. {
  1360. GNUNET_OP_remove (sr->chn->op, sr->op_id);
  1361. GNUNET_free (sr);
  1362. }
  1363. /* end of psyc_api.c */