src/psyc/psyc_api.c

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