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