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