src/psycstore/psycstore_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 psycstore/psycstore_api.c
  23  * @brief API to interact with the PSYCstore service
  24  * @author Gabor X Toth
  25  * @author Christian Grothoff
  26  */
  27
  28 #include <inttypes.h>
  29
  30 #include "platform.h"
  31 #include "gnunet_util_lib.h"
  32 #include "gnunet_constants.h"
  33 #include "gnunet_protocols.h"
  34 #include "gnunet_psycstore_service.h"
  35 #include "gnunet_multicast_service.h"
  36 #include "psycstore.h"
  37
  38 #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-api",__VA_ARGS__)
  39
  40 typedef void (*DataCallback) ();
  41
  42 /**
  43  * Handle for an operation with the PSYCstore service.
  44  */
  45 struct GNUNET_PSYCSTORE_OperationHandle
  46 {
  47
  48   /**
  49    * Main PSYCstore handle.
  50    */
  51   struct GNUNET_PSYCSTORE_Handle *h;
  52
  53   /**
  54    * We keep operations in a DLL.
  55    */
  56   struct GNUNET_PSYCSTORE_OperationHandle *next;
  57
  58   /**
  59    * We keep operations in a DLL.
  60    */
  61   struct GNUNET_PSYCSTORE_OperationHandle *prev;
  62
  63   /**
  64    * Continuation to invoke with the result of an operation.
  65    */
  66   GNUNET_PSYCSTORE_ResultCallback res_cb;
  67
  68   /**
  69    * Continuation to invoke with the result of an operation returning data.
  70    */
  71   DataCallback data_cb;
  72
  73   /**
  74    * Closure for the callbacks.
  75    */
  76   void *cls;
  77
  78   /**
  79    * Operation ID.
  80    */
  81   uint64_t op_id;
  82
  83   /**
  84    * Message to send to the PSYCstore service.
  85    * Allocated at the end of this struct.
  86    */
  87   const struct GNUNET_MessageHeader *msg;
  88 };
  89
  90
  91 /**
  92  * Handle for the service.
  93  */
  94 struct GNUNET_PSYCSTORE_Handle
  95 {
  96   /**
  97    * Configuration to use.
  98    */
  99   const struct GNUNET_CONFIGURATION_Handle *cfg;
 100
 101   /**
 102    * Socket (if available).
 103    */
 104   struct GNUNET_CLIENT_Connection *client;
 105
 106   /**
 107    * Head of operations to transmit.
 108    */
 109   struct GNUNET_PSYCSTORE_OperationHandle *transmit_head;
 110
 111   /**
 112    * Tail of operations to transmit.
 113    */
 114   struct GNUNET_PSYCSTORE_OperationHandle *transmit_tail;
 115
 116   /**
 117    * Head of active operations waiting for response.
 118    */
 119   struct GNUNET_PSYCSTORE_OperationHandle *op_head;
 120
 121   /**
 122    * Tail of active operations waiting for response.
 123    */
 124   struct GNUNET_PSYCSTORE_OperationHandle *op_tail;
 125
 126   /**
 127    * Currently pending transmission request, or NULL for none.
 128    */
 129   struct GNUNET_CLIENT_TransmitHandle *th;
 130
 131   /**
 132    * Task doing exponential back-off trying to reconnect.
 133    */
 134   struct GNUNET_SCHEDULER_Task * reconnect_task;
 135
 136   /**
 137    * Time for next connect retry.
 138    */
 139   struct GNUNET_TIME_Relative reconnect_delay;
 140
 141   /**
 142    * Last operation ID used.
 143    */
 144   uint64_t last_op_id;
 145
 146   /**
 147    * Are we polling for incoming messages right now?
 148    */
 149   uint8_t in_receive;
 150 };
 151
 152
 153 /**
 154  * Get a fresh operation ID to distinguish between PSYCstore requests.
 155  *
 156  * @param h Handle to the PSYCstore service.
 157  * @return next operation id to use
 158  */
 159 static uint64_t
 160 get_next_op_id (struct GNUNET_PSYCSTORE_Handle *h)
 161 {
 162   return h->last_op_id++;
 163 }
 164
 165
 166 /**
 167  * Find operation by ID.
 168  *
 169  * @return OperationHandle if found, or NULL otherwise.
 170  */
 171 static struct GNUNET_PSYCSTORE_OperationHandle *
 172 find_op_by_id (struct GNUNET_PSYCSTORE_Handle *h, uint64_t op_id)
 173 {
 174   struct GNUNET_PSYCSTORE_OperationHandle *op = h->op_head;
 175   while (NULL != op)
 176   {
 177     if (op->op_id == op_id)
 178       return op;
 179     op = op->next;
 180   }
 181   return NULL;
 182 }
 183
 184
 185 /**
 186  * Try again to connect to the PSYCstore service.
 187  *
 188  * @param cls handle to the PSYCstore service.
 189  * @param tc scheduler context
 190  */
 191 static void
 192 reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 193
 194
 195 /**
 196  * Reschedule a connect attempt to the service.
 197  *
 198  * @param h transport service to reconnect
 199  */
 200 static void
 201 reschedule_connect (struct GNUNET_PSYCSTORE_Handle *h)
 202 {
 203   GNUNET_assert (h->reconnect_task == NULL);
 204
 205   if (NULL != h->th)
 206   {
 207     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 208     h->th = NULL;
 209   }
 210   if (NULL != h->client)
 211   {
 212     GNUNET_CLIENT_disconnect (h->client);
 213     h->client = NULL;
 214   }
 215   h->in_receive = GNUNET_NO;
 216   LOG (GNUNET_ERROR_TYPE_DEBUG,
 217        "Scheduling task to reconnect to PSYCstore service in %s.\n",
 218        GNUNET_STRINGS_relative_time_to_string (h->reconnect_delay, GNUNET_YES));
 219   h->reconnect_task =
 220       GNUNET_SCHEDULER_add_delayed (h->reconnect_delay, &reconnect, h);
 221   h->reconnect_delay = GNUNET_TIME_STD_BACKOFF (h->reconnect_delay);
 222 }
 223
 224
 225 /**
 226  * Schedule transmission of the next message from our queue.
 227  *
 228  * @param h PSYCstore handle
 229  */
 230 static void
 231 transmit_next (struct GNUNET_PSYCSTORE_Handle *h);
 232
 233
 234 /**
 235  * Type of a function to call when we receive a message
 236  * from the service.
 237  *
 238  * @param cls closure
 239  * @param msg message received, NULL on timeout or fatal error
 240  */
 241 static void
 242 message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
 243 {
 244   struct GNUNET_PSYCSTORE_Handle *h = cls;
 245   struct GNUNET_PSYCSTORE_OperationHandle *op;
 246   const struct OperationResult *opres;
 247   const struct CountersResult *cres;
 248   const struct FragmentResult *fres;
 249   const struct StateResult *sres;
 250   const char *str;
 251
 252   if (NULL == msg)
 253   {
 254     reschedule_connect (h);
 255     return;
 256   }
 257   LOG (GNUNET_ERROR_TYPE_DEBUG,
 258        "Received message of type %d from PSYCstore service.\n",
 259        ntohs (msg->type));
 260   uint16_t size = ntohs (msg->size);
 261   uint16_t type = ntohs (msg->type);
 262   switch (type)
 263   {
 264   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_CODE:
 265     if (size < sizeof (struct OperationResult))
 266     {
 267       LOG (GNUNET_ERROR_TYPE_ERROR,
 268            "Received message of type %d with length %lu bytes. "
 269            "Expected >= %lu\n",
 270            type, size, sizeof (struct OperationResult));
 271       GNUNET_break (0);
 272       reschedule_connect (h);
 273       return;
 274     }
 275
 276     opres = (const struct OperationResult *) msg;
 277     str = (const char *) &opres[1];
 278     if ( (size > sizeof (struct OperationResult)) &&
 279          ('\0' != str[size - sizeof (struct OperationResult) - 1]) )
 280     {
 281       GNUNET_break (0);
 282       reschedule_connect (h);
 283       return;
 284     }
 285     if (size == sizeof (struct OperationResult))
 286       str = "";
 287
 288     op = find_op_by_id (h, GNUNET_ntohll (opres->op_id));
 289     if (NULL == op)
 290     {
 291       LOG (GNUNET_ERROR_TYPE_DEBUG,
 292            "No callback registered for operation with ID %" PRIu64 ".\n",
 293            type, GNUNET_ntohll (opres->op_id));
 294     }
 295     else
 296     {
 297       LOG (GNUNET_ERROR_TYPE_DEBUG,
 298            "Received result message (type %d) with operation ID: %" PRIu64 "\n",
 299            type, op->op_id);
 300
 301       int64_t result_code = GNUNET_ntohll (opres->result_code) + INT64_MIN;
 302       GNUNET_CONTAINER_DLL_remove (h->op_head, h->op_tail, op);
 303       if (NULL != op->res_cb)
 304       {
 305         const struct StateSyncRequest *ssreq;
 306         switch (ntohs (op->msg->type))
 307         {
 308         case GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC:
 309           ssreq = (const struct StateSyncRequest *) op->msg;
 310           if (!(ssreq->flags & STATE_OP_LAST
 311                 || GNUNET_OK != result_code))
 312             op->res_cb = NULL;
 313           break;
 314         }
 315       }
 316       if (NULL != op->res_cb)
 317         op->res_cb (op->cls, result_code, str, size - sizeof (*opres));
 318       GNUNET_free (op);
 319     }
 320     break;
 321
 322   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_COUNTERS:
 323     if (size != sizeof (struct CountersResult))
 324     {
 325       LOG (GNUNET_ERROR_TYPE_ERROR,
 326            "Received message of type %d with length %lu bytes. "
 327            "Expected %lu\n",
 328            type, size, sizeof (struct CountersResult));
 329       GNUNET_break (0);
 330       reschedule_connect (h);
 331       return;
 332     }
 333
 334     cres = (const struct CountersResult *) msg;
 335
 336     op = find_op_by_id (h, GNUNET_ntohll (cres->op_id));
 337     if (NULL == op)
 338     {
 339       LOG (GNUNET_ERROR_TYPE_DEBUG,
 340            "No callback registered for operation with ID %" PRIu64 ".\n",
 341            type, GNUNET_ntohll (cres->op_id));
 342     }
 343     else
 344     {
 345       GNUNET_CONTAINER_DLL_remove (h->op_head, h->op_tail, op);
 346       if (NULL != op->data_cb)
 347         ((GNUNET_PSYCSTORE_CountersCallback)
 348          op->data_cb) (op->cls,
 349                        ntohl (cres->result_code),
 350                        GNUNET_ntohll (cres->max_fragment_id),
 351                        GNUNET_ntohll (cres->max_message_id),
 352                        GNUNET_ntohll (cres->max_group_generation),
 353                        GNUNET_ntohll (cres->max_state_message_id));
 354       GNUNET_free (op);
 355     }
 356     break;
 357
 358   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_FRAGMENT:
 359     if (size < sizeof (struct FragmentResult))
 360     {
 361       LOG (GNUNET_ERROR_TYPE_ERROR,
 362            "Received message of type %d with length %lu bytes. "
 363            "Expected >= %lu\n",
 364            type, size, sizeof (struct FragmentResult));
 365       GNUNET_break (0);
 366       reschedule_connect (h);
 367       return;
 368     }
 369
 370     fres = (const struct FragmentResult *) msg;
 371     struct GNUNET_MULTICAST_MessageHeader *mmsg =
 372       (struct GNUNET_MULTICAST_MessageHeader *) &fres[1];
 373     if (size != sizeof (struct FragmentResult) + ntohs (mmsg->header.size))
 374     {
 375       LOG (GNUNET_ERROR_TYPE_ERROR,
 376            "Received message of type %d with length %lu bytes. "
 377            "Expected = %lu\n",
 378            type, size,
 379            sizeof (struct FragmentResult) + ntohs (mmsg->header.size));
 380       GNUNET_break (0);
 381       reschedule_connect (h);
 382       return;
 383     }
 384
 385     op = find_op_by_id (h, GNUNET_ntohll (fres->op_id));
 386     if (NULL == op)
 387     {
 388       LOG (GNUNET_ERROR_TYPE_DEBUG,
 389            "No callback registered for operation with ID %" PRIu64 ".\n",
 390            type, GNUNET_ntohll (fres->op_id));
 391     }
 392     else
 393     {
 394       if (NULL != op->data_cb)
 395         ((GNUNET_PSYCSTORE_FragmentCallback)
 396          op->data_cb) (op->cls, mmsg, ntohl (fres->psycstore_flags));
 397     }
 398     break;
 399
 400   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_STATE:
 401     if (size < sizeof (struct StateResult))
 402     {
 403       LOG (GNUNET_ERROR_TYPE_ERROR,
 404            "Received message of type %d with length %lu bytes. "
 405            "Expected >= %lu\n",
 406            type, size, sizeof (struct StateResult));
 407       GNUNET_break (0);
 408       reschedule_connect (h);
 409       return;
 410     }
 411
 412     sres = (const struct StateResult *) msg;
 413     const char *name = (const char *) &sres[1];
 414     uint16_t name_size = ntohs (sres->name_size);
 415
 416     if (name_size <= 2 || '\0' != name[name_size - 1])
 417     {
 418       LOG (GNUNET_ERROR_TYPE_ERROR,
 419            "Received state result message (type %d) with invalid name.\n",
 420            type);
 421       GNUNET_break (0);
 422       reschedule_connect (h);
 423       return;
 424     }
 425
 426     op = find_op_by_id (h, GNUNET_ntohll (sres->op_id));
 427     if (NULL == op)
 428     {
 429       LOG (GNUNET_ERROR_TYPE_DEBUG,
 430            "No callback registered for operation with ID %" PRIu64 ".\n",
 431            type, GNUNET_ntohll (sres->op_id));
 432     }
 433     else
 434     {
 435       if (NULL != op->data_cb)
 436         ((GNUNET_PSYCSTORE_StateCallback)
 437          op->data_cb) (op->cls, name, (char *) &sres[1] + name_size,
 438                        ntohs (sres->header.size) - sizeof (*sres) - name_size);
 439     }
 440     break;
 441
 442   default:
 443     GNUNET_break (0);
 444     reschedule_connect (h);
 445     return;
 446   }
 447
 448   GNUNET_CLIENT_receive (h->client, &message_handler, h,
 449                          GNUNET_TIME_UNIT_FOREVER_REL);
 450 }
 451
 452
 453 /**
 454  * Transmit next message to service.
 455  *
 456  * @param cls The 'struct GNUNET_PSYCSTORE_Handle'.
 457  * @param size Number of bytes available in buf.
 458  * @param buf Where to copy the message.
 459  * @return Number of bytes copied to buf.
 460  */
 461 static size_t
 462 send_next_message (void *cls, size_t size, void *buf)
 463 {
 464   struct GNUNET_PSYCSTORE_Handle *h = cls;
 465   struct GNUNET_PSYCSTORE_OperationHandle *op = h->transmit_head;
 466   size_t ret;
 467
 468   h->th = NULL;
 469   if (NULL == op)
 470     return 0;
 471   ret = ntohs (op->msg->size);
 472   if (ret > size)
 473   {
 474     reschedule_connect (h);
 475     return 0;
 476   }
 477   LOG (GNUNET_ERROR_TYPE_DEBUG,
 478        "Sending message of type %d to PSYCstore service. ID: %" PRIu64 "\n",
 479        ntohs (op->msg->type), op->op_id);
 480   memcpy (buf, op->msg, ret);
 481
 482   GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 483
 484   if (NULL == op->res_cb && NULL == op->data_cb)
 485   {
 486     GNUNET_free (op);
 487   }
 488   else
 489   {
 490     GNUNET_CONTAINER_DLL_insert_tail (h->op_head, h->op_tail, op);
 491   }
 492
 493   if (NULL != h->transmit_head)
 494     transmit_next (h);
 495
 496   if (GNUNET_NO == h->in_receive)
 497   {
 498     h->in_receive = GNUNET_YES;
 499     GNUNET_CLIENT_receive (h->client, &message_handler, h,
 500                            GNUNET_TIME_UNIT_FOREVER_REL);
 501   }
 502   return ret;
 503 }
 504
 505
 506 /**
 507  * Schedule transmission of the next message from our queue.
 508  *
 509  * @param h PSYCstore handle.
 510  */
 511 static void
 512 transmit_next (struct GNUNET_PSYCSTORE_Handle *h)
 513 {
 514   if (NULL != h->th || NULL == h->client)
 515     return;
 516
 517   struct GNUNET_PSYCSTORE_OperationHandle *op = h->transmit_head;
 518   if (NULL == op)
 519     return;
 520
 521   h->th = GNUNET_CLIENT_notify_transmit_ready (h->client,
 522                                                ntohs (op->msg->size),
 523                                                GNUNET_TIME_UNIT_FOREVER_REL,
 524                                                GNUNET_NO,
 525                                                &send_next_message,
 526                                                h);
 527 }
 528
 529
 530 /**
 531  * Try again to connect to the PSYCstore service.
 532  *
 533  * @param cls Handle to the PSYCstore service.
 534  * @param tc Scheduler context.
 535  */
 536 static void
 537 reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 538 {
 539   struct GNUNET_PSYCSTORE_Handle *h = cls;
 540
 541   h->reconnect_task = NULL;
 542   LOG (GNUNET_ERROR_TYPE_DEBUG,
 543        "Connecting to PSYCstore service.\n");
 544   GNUNET_assert (NULL == h->client);
 545   h->client = GNUNET_CLIENT_connect ("psycstore", h->cfg);
 546   GNUNET_assert (NULL != h->client);
 547   transmit_next (h);
 548 }
 549
 550
 551 /**
 552  * Connect to the PSYCstore service.
 553  *
 554  * @param cfg The configuration to use
 555  * @return Handle to use
 556  */
 557 struct GNUNET_PSYCSTORE_Handle *
 558 GNUNET_PSYCSTORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
 559 {
 560   struct GNUNET_PSYCSTORE_Handle *h
 561     = GNUNET_new (struct GNUNET_PSYCSTORE_Handle);
 562   h->cfg = cfg;
 563   h->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
 564   h->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, h);
 565   return h;
 566 }
 567
 568
 569 /**
 570  * Disconnect from PSYCstore service
 571  *
 572  * @param h Handle to destroy
 573  */
 574 void
 575 GNUNET_PSYCSTORE_disconnect (struct GNUNET_PSYCSTORE_Handle *h)
 576 {
 577   GNUNET_assert (NULL != h);
 578   if (h->reconnect_task != NULL)
 579   {
 580     GNUNET_SCHEDULER_cancel (h->reconnect_task);
 581     h->reconnect_task = NULL;
 582   }
 583   if (NULL != h->th)
 584   {
 585     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 586     h->th = NULL;
 587   }
 588   if (NULL != h->client)
 589   {
 590     GNUNET_CLIENT_disconnect (h->client);
 591     h->client = NULL;
 592   }
 593   GNUNET_free (h);
 594 }
 595
 596
 597 /**
 598  * Cancel a PSYCstore operation. Note that the operation MAY still
 599  * be executed; this merely cancels the continuation; if the request
 600  * was already transmitted, the service may still choose to complete
 601  * the operation.
 602  *
 603  * @param op Operation to cancel.
 604  */
 605 void
 606 GNUNET_PSYCSTORE_operation_cancel (struct GNUNET_PSYCSTORE_OperationHandle *op)
 607 {
 608   struct GNUNET_PSYCSTORE_Handle *h = op->h;
 609
 610   if (h->transmit_head != NULL && (h->transmit_head != op || NULL == h->client))
 611   {
 612     /* request not active, can simply remove */
 613     GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 614     GNUNET_free (op);
 615     return;
 616   }
 617   if (NULL != h->th)
 618   {
 619     /* request active but not yet with service, can still abort */
 620     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 621     h->th = NULL;
 622     GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 623     GNUNET_free (op);
 624     transmit_next (h);
 625     return;
 626   }
 627   /* request active with service, simply ensure continuations are not called */
 628   op->res_cb = NULL;
 629   op->data_cb = NULL;
 630 }
 631
 632
 633 /**
 634  * Store join/leave events for a PSYC channel in order to be able to answer
 635  * membership test queries later.
 636  *
 637  * @param h
 638  *        Handle for the PSYCstore.
 639  * @param channel_key
 640  *        The channel where the event happened.
 641  * @param slave_key
 642  *        Public key of joining/leaving slave.
 643  * @param did_join
 644  *        #GNUNET_YES on join, #GNUNET_NO on part.
 645  * @param announced_at
 646  *        ID of the message that announced the membership change.
 647  * @param effective_since
 648  *        Message ID this membership change is in effect since.
 649  *        For joins it is <= announced_at, for parts it is always 0.
 650  * @param group_generation
 651  *        In case of a part, the last group generation the slave has access to.
 652  *        It has relevance when a larger message have fragments with different
 653  *        group generations.
 654  * @param rcb
 655  *        Callback to call with the result of the storage operation.
 656  * @param rcb_cls
 657  *        Closure for the callback.
 658  *
 659  * @return Operation handle that can be used to cancel the operation.
 660  */
 661 struct GNUNET_PSYCSTORE_OperationHandle *
 662 GNUNET_PSYCSTORE_membership_store (struct GNUNET_PSYCSTORE_Handle *h,
 663                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 664                                    const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 665                                    int did_join,
 666                                    uint64_t announced_at,
 667                                    uint64_t effective_since,
 668                                    uint64_t group_generation,
 669                                    GNUNET_PSYCSTORE_ResultCallback rcb,
 670                                    void *rcb_cls)
 671 {
 672   GNUNET_assert (NULL != h);
 673   GNUNET_assert (NULL != channel_key);
 674   GNUNET_assert (NULL != slave_key);
 675   GNUNET_assert (GNUNET_YES == did_join || GNUNET_NO == did_join);
 676   GNUNET_assert (did_join
 677                  ? effective_since <= announced_at
 678                  : effective_since == 0);
 679
 680   struct MembershipStoreRequest *req;
 681   struct GNUNET_PSYCSTORE_OperationHandle *
 682     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 683   op->h = h;
 684   op->res_cb = rcb;
 685   op->cls = rcb_cls;
 686
 687   req = (struct MembershipStoreRequest *) &op[1];
 688   op->msg = (struct GNUNET_MessageHeader *) req;
 689   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_STORE);
 690   req->header.size = htons (sizeof (*req));
 691   req->channel_key = *channel_key;
 692   req->slave_key = *slave_key;
 693   req->did_join = did_join;
 694   req->announced_at = GNUNET_htonll (announced_at);
 695   req->effective_since = GNUNET_htonll (effective_since);
 696   req->group_generation = GNUNET_htonll (group_generation);
 697
 698   op->op_id = get_next_op_id (h);
 699   req->op_id = GNUNET_htonll (op->op_id);
 700
 701   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 702   transmit_next (h);
 703
 704   return op;
 705 }
 706
 707
 708 /**
 709  * Test if a member was admitted to the channel at the given message ID.
 710  *
 711  * This is useful when relaying and replaying messages to check if a particular
 712  * slave has access to the message fragment with a given group generation.  It
 713  * is also used when handling join requests to determine whether the slave is
 714  * currently admitted to the channel.
 715  *
 716  * @param h
 717  *        Handle for the PSYCstore.
 718  * @param channel_key
 719  *        The channel we are interested in.
 720  * @param slave_key
 721  *        Public key of slave whose membership to check.
 722  * @param message_id
 723  *        Message ID for which to do the membership test.
 724  * @param group_generation
 725  *        Group generation of the fragment of the message to test.
 726  *        It has relevance if the message consists of multiple fragments with
 727  *        different group generations.
 728  * @param rcb
 729  *        Callback to call with the test result.
 730  * @param rcb_cls
 731  *        Closure for the callback.
 732  *
 733  * @return Operation handle that can be used to cancel the operation.
 734  */
 735 struct GNUNET_PSYCSTORE_OperationHandle *
 736 GNUNET_PSYCSTORE_membership_test (struct GNUNET_PSYCSTORE_Handle *h,
 737                                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 738                                   const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 739                                   uint64_t message_id,
 740                                   uint64_t group_generation,
 741                                   GNUNET_PSYCSTORE_ResultCallback rcb,
 742                                   void *rcb_cls)
 743 {
 744   struct MembershipTestRequest *req;
 745   struct GNUNET_PSYCSTORE_OperationHandle *
 746     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 747   op->h = h;
 748   op->res_cb = rcb;
 749   op->cls = rcb_cls;
 750
 751   req = (struct MembershipTestRequest *) &op[1];
 752   op->msg = (struct GNUNET_MessageHeader *) req;
 753   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_TEST);
 754   req->header.size = htons (sizeof (*req));
 755   req->channel_key = *channel_key;
 756   req->slave_key = *slave_key;
 757   req->message_id = GNUNET_htonll (message_id);
 758   req->group_generation = GNUNET_htonll (group_generation);
 759
 760   op->op_id = get_next_op_id (h);
 761   req->op_id = GNUNET_htonll (op->op_id);
 762
 763   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 764   transmit_next (h);
 765
 766   return op;
 767 }
 768
 769
 770 /**
 771  * Store a message fragment sent to a channel.
 772  *
 773  * @param h Handle for the PSYCstore.
 774  * @param channel_key The channel the message belongs to.
 775  * @param message Message to store.
 776  * @param psycstore_flags Flags indicating whether the PSYC message contains
 777  *        state modifiers.
 778  * @param rcb Callback to call with the result of the operation.
 779  * @param rcb_cls Closure for the callback.
 780  *
 781  * @return Handle that can be used to cancel the operation.
 782  */
 783 struct GNUNET_PSYCSTORE_OperationHandle *
 784 GNUNET_PSYCSTORE_fragment_store (struct GNUNET_PSYCSTORE_Handle *h,
 785                                  const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 786                                  const struct GNUNET_MULTICAST_MessageHeader *msg,
 787                                  enum GNUNET_PSYCSTORE_MessageFlags psycstore_flags,
 788                                  GNUNET_PSYCSTORE_ResultCallback rcb,
 789                                  void *rcb_cls)
 790 {
 791   uint16_t size = ntohs (msg->header.size);
 792   struct FragmentStoreRequest *req;
 793   struct GNUNET_PSYCSTORE_OperationHandle *
 794     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + size);
 795   op->h = h;
 796   op->res_cb = rcb;
 797   op->cls = rcb_cls;
 798
 799   req = (struct FragmentStoreRequest *) &op[1];
 800   op->msg = (struct GNUNET_MessageHeader *) req;
 801   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_STORE);
 802   req->header.size = htons (sizeof (*req) + size);
 803   req->channel_key = *channel_key;
 804   req->psycstore_flags = htonl (psycstore_flags);
 805   memcpy (&req[1], msg, size);
 806
 807   op->op_id = get_next_op_id (h);
 808   req->op_id = GNUNET_htonll (op->op_id);
 809
 810   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 811   transmit_next (h);
 812
 813   return op;
 814 }
 815
 816
 817 /**
 818  * Retrieve message fragments by fragment ID range.
 819  *
 820  * @param h
 821  *        Handle for the PSYCstore.
 822  * @param channel_key
 823  *        The channel we are interested in.
 824  * @param slave_key
 825  *        The slave requesting the fragment.  If not NULL, a membership test is
 826  *        performed first and the fragment is only returned if the slave has
 827  *        access to it.
 828  * @param first_fragment_id
 829  *        First fragment ID to retrieve.
 830  *        Use 0 to get the latest message fragment.
 831  * @param last_fragment_id
 832  *        Last consecutive fragment ID to retrieve.
 833  *        Use 0 to get the latest message fragment.
 834  * @param fragment_limit
 835  *        Maximum number of fragments to retrieve.
 836  * @param fragment_cb
 837  *        Callback to call with the retrieved fragments.
 838  * @param rcb
 839  *        Callback to call with the result of the operation.
 840  * @param cls
 841  *        Closure for the callbacks.
 842  *
 843  * @return Handle that can be used to cancel the operation.
 844  */
 845 struct GNUNET_PSYCSTORE_OperationHandle *
 846 GNUNET_PSYCSTORE_fragment_get (struct GNUNET_PSYCSTORE_Handle *h,
 847                                const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 848                                const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 849                                uint64_t first_fragment_id,
 850                                uint64_t last_fragment_id,
 851                                GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 852                                GNUNET_PSYCSTORE_ResultCallback rcb,
 853                                void *cls)
 854 {
 855   struct FragmentGetRequest *req;
 856   struct GNUNET_PSYCSTORE_OperationHandle *
 857     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 858   op->h = h;
 859   op->data_cb = (DataCallback) fragment_cb;
 860   op->res_cb = rcb;
 861   op->cls = cls;
 862
 863   req = (struct FragmentGetRequest *) &op[1];
 864   op->msg = (struct GNUNET_MessageHeader *) req;
 865   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET);
 866   req->header.size = htons (sizeof (*req));
 867   req->channel_key = *channel_key;
 868   req->first_fragment_id = GNUNET_htonll (first_fragment_id);
 869   req->last_fragment_id = GNUNET_htonll (last_fragment_id);
 870   if (NULL != slave_key)
 871   {
 872     req->slave_key = *slave_key;
 873     req->do_membership_test = GNUNET_YES;
 874   }
 875
 876   op->op_id = get_next_op_id (h);
 877   req->op_id = GNUNET_htonll (op->op_id);
 878
 879   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 880   transmit_next (h);
 881
 882   return op;
 883 }
 884
 885
 886 /**
 887  * Retrieve latest message fragments.
 888  *
 889  * @param h
 890  *        Handle for the PSYCstore.
 891  * @param channel_key
 892  *        The channel we are interested in.
 893  * @param slave_key
 894  *        The slave requesting the fragment.  If not NULL, a membership test is
 895  *        performed first and the fragment is only returned if the slave has
 896  *        access to it.
 897  * @param first_fragment_id
 898  *        First fragment ID to retrieve.
 899  *        Use 0 to get the latest message fragment.
 900  * @param last_fragment_id
 901  *        Last consecutive fragment ID to retrieve.
 902  *        Use 0 to get the latest message fragment.
 903  * @param fragment_limit
 904  *        Maximum number of fragments to retrieve.
 905  * @param fragment_cb
 906  *        Callback to call with the retrieved fragments.
 907  * @param rcb
 908  *        Callback to call with the result of the operation.
 909  * @param cls
 910  *        Closure for the callbacks.
 911  *
 912  * @return Handle that can be used to cancel the operation.
 913  */
 914 struct GNUNET_PSYCSTORE_OperationHandle *
 915 GNUNET_PSYCSTORE_fragment_get_latest (struct GNUNET_PSYCSTORE_Handle *h,
 916                                       const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 917                                       const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 918                                       uint64_t fragment_limit,
 919                                       GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 920                                       GNUNET_PSYCSTORE_ResultCallback rcb,
 921                                       void *cls)
 922 {
 923   struct FragmentGetRequest *req;
 924   struct GNUNET_PSYCSTORE_OperationHandle *
 925     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 926   op->h = h;
 927   op->data_cb = (DataCallback) fragment_cb;
 928   op->res_cb = rcb;
 929   op->cls = cls;
 930
 931   req = (struct FragmentGetRequest *) &op[1];
 932   op->msg = (struct GNUNET_MessageHeader *) req;
 933   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET);
 934   req->header.size = htons (sizeof (*req));
 935   req->channel_key = *channel_key;
 936   req->fragment_limit = GNUNET_ntohll (fragment_limit);
 937   if (NULL != slave_key)
 938   {
 939     req->slave_key = *slave_key;
 940     req->do_membership_test = GNUNET_YES;
 941   }
 942
 943   op->op_id = get_next_op_id (h);
 944   req->op_id = GNUNET_htonll (op->op_id);
 945
 946   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 947   transmit_next (h);
 948
 949   return op;
 950 }
 951
 952
 953 /**
 954  * Retrieve all fragments of messages in a message ID range.
 955  *
 956  * @param h
 957  *        Handle for the PSYCstore.
 958  * @param channel_key
 959  *        The channel we are interested in.
 960  * @param slave_key
 961  *        The slave requesting the message.
 962  *        If not NULL, a membership test is performed first
 963  *        and the message is only returned if the slave has access to it.
 964  * @param first_message_id
 965  *        First message ID to retrieve.
 966  * @param last_message_id
 967  *        Last consecutive message ID to retrieve.
 968  * @param fragment_limit
 969  *        Maximum number of fragments to retrieve.
 970  * @param method_prefix
 971  *        Retrieve only messages with a matching method prefix.
 972  * @todo Implement method_prefix query.
 973  * @param fragment_cb
 974  *        Callback to call with the retrieved fragments.
 975  * @param result_cb
 976  *        Callback to call with the result of the operation.
 977  * @param cls
 978  *        Closure for the callbacks.
 979  *
 980  * @return Handle that can be used to cancel the operation.
 981  */
 982 struct GNUNET_PSYCSTORE_OperationHandle *
 983 GNUNET_PSYCSTORE_message_get (struct GNUNET_PSYCSTORE_Handle *h,
 984                               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 985                               const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 986                               uint64_t first_message_id,
 987                               uint64_t last_message_id,
 988                               uint64_t fragment_limit,
 989                               const char *method_prefix,
 990                               GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 991                               GNUNET_PSYCSTORE_ResultCallback rcb,
 992                               void *cls)
 993 {
 994   struct MessageGetRequest *req;
 995   if (NULL == method_prefix)
 996     method_prefix = "";
 997   uint16_t method_size = strnlen (method_prefix,
 998                                   GNUNET_SERVER_MAX_MESSAGE_SIZE
 999                                   - sizeof (*req)) + 1;
1000
1001   struct GNUNET_PSYCSTORE_OperationHandle *
1002     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1003   op->h = h;
1004   op->data_cb = (DataCallback) fragment_cb;
1005   op->res_cb = rcb;
1006   op->cls = cls;
1007
1008   req = (struct MessageGetRequest *) &op[1];
1009   op->msg = (struct GNUNET_MessageHeader *) req;
1010   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1011   req->header.size = htons (sizeof (*req) + method_size);
1012   req->channel_key = *channel_key;
1013   req->first_message_id = GNUNET_htonll (first_message_id);
1014   req->last_message_id = GNUNET_htonll (last_message_id);
1015   req->fragment_limit = GNUNET_htonll (fragment_limit);
1016   if (NULL != slave_key)
1017   {
1018     req->slave_key = *slave_key;
1019     req->do_membership_test = GNUNET_YES;
1020   }
1021   memcpy (&req[1], method_prefix, method_size);
1022   ((char *) &req[1])[method_size - 1] = '\0';
1023
1024   op->op_id = get_next_op_id (h);
1025   req->op_id = GNUNET_htonll (op->op_id);
1026
1027   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1028   transmit_next (h);
1029
1030   return op;
1031 }
1032
1033
1034 /**
1035  * Retrieve all fragments of the latest messages.
1036  *
1037  * @param h
1038  *        Handle for the PSYCstore.
1039  * @param channel_key
1040  *        The channel we are interested in.
1041  * @param slave_key
1042  *        The slave requesting the message.
1043  *        If not NULL, a membership test is performed first
1044  *        and the message is only returned if the slave has access to it.
1045  * @param message_limit
1046  *        Maximum number of messages to retrieve.
1047  * @param method_prefix
1048  *        Retrieve only messages with a matching method prefix.
1049  * @todo Implement method_prefix query.
1050  * @param fragment_cb
1051  *        Callback to call with the retrieved fragments.
1052  * @param result_cb
1053  *        Callback to call with the result of the operation.
1054  * @param cls
1055  *        Closure for the callbacks.
1056  *
1057  * @return Handle that can be used to cancel the operation.
1058  */
1059 struct GNUNET_PSYCSTORE_OperationHandle *
1060 GNUNET_PSYCSTORE_message_get_latest (struct GNUNET_PSYCSTORE_Handle *h,
1061                                      const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1062                                      const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1063                                      uint64_t message_limit,
1064                                      const char *method_prefix,
1065                                      GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1066                                      GNUNET_PSYCSTORE_ResultCallback rcb,
1067                                      void *cls)
1068 {
1069   struct MessageGetRequest *req;
1070
1071   if (NULL == method_prefix)
1072     method_prefix = "";
1073   uint16_t method_size = strnlen (method_prefix,
1074                                   GNUNET_SERVER_MAX_MESSAGE_SIZE
1075                                   - sizeof (*req)) + 1;
1076   GNUNET_assert ('\0' == method_prefix[method_size - 1]);
1077
1078   struct GNUNET_PSYCSTORE_OperationHandle *
1079     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + method_size);
1080   op->h = h;
1081   op->data_cb = (DataCallback) fragment_cb;
1082   op->res_cb = rcb;
1083   op->cls = cls;
1084
1085   req = (struct MessageGetRequest *) &op[1];
1086   op->msg = (struct GNUNET_MessageHeader *) req;
1087   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1088   req->header.size = htons (sizeof (*req) + method_size);
1089   req->channel_key = *channel_key;
1090   req->message_limit = GNUNET_ntohll (message_limit);
1091   if (NULL != slave_key)
1092   {
1093     req->slave_key = *slave_key;
1094     req->do_membership_test = GNUNET_YES;
1095   }
1096
1097   op->op_id = get_next_op_id (h);
1098   req->op_id = GNUNET_htonll (op->op_id);
1099   memcpy (&req[1], method_prefix, method_size);
1100
1101   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1102   transmit_next (h);
1103
1104   return op;
1105 }
1106
1107
1108 /**
1109  * Retrieve a fragment of message specified by its message ID and fragment
1110  * offset.
1111  *
1112  * @param h
1113  *        Handle for the PSYCstore.
1114  * @param channel_key
1115  *        The channel we are interested in.
1116  * @param slave_key
1117  *        The slave requesting the message fragment.  If not NULL, a membership
1118  *        test is performed first and the message fragment is only returned
1119  *        if the slave has access to it.
1120  * @param message_id
1121  *        Message ID to retrieve.  Use 0 to get the latest message.
1122  * @param fragment_offset
1123  *        Offset of the fragment to retrieve.
1124  * @param fragment_cb
1125  *        Callback to call with the retrieved fragments.
1126  * @param result_cb
1127  *        Callback to call with the result of the operation.
1128  * @param cls
1129  *        Closure for the callbacks.
1130  *
1131  * @return Handle that can be used to cancel the operation.
1132  */
1133 struct GNUNET_PSYCSTORE_OperationHandle *
1134 GNUNET_PSYCSTORE_message_get_fragment (struct GNUNET_PSYCSTORE_Handle *h,
1135                                        const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1136                                        const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1137                                        uint64_t message_id,
1138                                        uint64_t fragment_offset,
1139                                        GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1140                                        GNUNET_PSYCSTORE_ResultCallback rcb,
1141                                        void *cls)
1142 {
1143   struct MessageGetFragmentRequest *req;
1144   struct GNUNET_PSYCSTORE_OperationHandle *
1145     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1146   op->h = h;
1147   op->data_cb = (DataCallback) fragment_cb;
1148   op->res_cb = rcb;
1149   op->cls = cls;
1150
1151   req = (struct MessageGetFragmentRequest *) &op[1];
1152   op->msg = (struct GNUNET_MessageHeader *) req;
1153   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET_FRAGMENT);
1154   req->header.size = htons (sizeof (*req));
1155   req->channel_key = *channel_key;
1156   req->message_id = GNUNET_htonll (message_id);
1157   req->fragment_offset = GNUNET_htonll (fragment_offset);
1158   if (NULL != slave_key)
1159   {
1160     req->slave_key = *slave_key;
1161     req->do_membership_test = GNUNET_YES;
1162   }
1163
1164   op->op_id = get_next_op_id (h);
1165   req->op_id = GNUNET_htonll (op->op_id);
1166
1167   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1168   transmit_next (h);
1169
1170   return op;
1171 }
1172
1173
1174 /**
1175  * Retrieve latest values of counters for a channel master.
1176  *
1177  * The current value of counters are needed when a channel master is restarted,
1178  * so that it can continue incrementing the counters from their last value.
1179  *
1180  * @param h
1181  *        Handle for the PSYCstore.
1182  * @param channel_key
1183  *        Public key that identifies the channel.
1184  * @param ccb
1185  *        Callback to call with the result.
1186  * @param ccb_cls
1187  *        Closure for the @a ccb callback.
1188  *
1189  * @return Handle that can be used to cancel the operation.
1190  */
1191 struct GNUNET_PSYCSTORE_OperationHandle *
1192 GNUNET_PSYCSTORE_counters_get (struct GNUNET_PSYCSTORE_Handle *h,
1193                                struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1194                                GNUNET_PSYCSTORE_CountersCallback ccb,
1195                                void *ccb_cls)
1196 {
1197   struct OperationRequest *req;
1198   struct GNUNET_PSYCSTORE_OperationHandle *
1199     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1200   op->h = h;
1201   op->data_cb = ccb;
1202   op->cls = ccb_cls;
1203
1204   req = (struct OperationRequest *) &op[1];
1205   op->msg = (struct GNUNET_MessageHeader *) req;
1206   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET);
1207   req->header.size = htons (sizeof (*req));
1208   req->channel_key = *channel_key;
1209
1210   op->op_id = get_next_op_id (h);
1211   req->op_id = GNUNET_htonll (op->op_id);
1212
1213   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1214   transmit_next (h);
1215
1216   return op;
1217 }
1218
1219
1220 /**
1221  * Apply modifiers of a message to the current channel state.
1222  *
1223  * An error is returned if there are missing messages containing state
1224  * operations before the current one.
1225  *
1226  * @param h
1227  *        Handle for the PSYCstore.
1228  * @param channel_key
1229  *        The channel we are interested in.
1230  * @param message_id
1231  *        ID of the message that contains the @a modifiers.
1232  * @param state_delta
1233  *        Value of the _state_delta PSYC header variable of the message.
1234  * @param rcb
1235  *        Callback to call with the result of the operation.
1236  * @param rcb_cls
1237  *        Closure for the @a rcb callback.
1238  *
1239  * @return Handle that can be used to cancel the operation.
1240  */
1241 struct GNUNET_PSYCSTORE_OperationHandle *
1242 GNUNET_PSYCSTORE_state_modify (struct GNUNET_PSYCSTORE_Handle *h,
1243                                const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1244                                uint64_t message_id,
1245                                uint64_t state_delta,
1246                                GNUNET_PSYCSTORE_ResultCallback rcb,
1247                                void *rcb_cls)
1248 {
1249   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1250   struct StateModifyRequest *req;
1251
1252   op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1253   op->h = h;
1254   op->res_cb = rcb;
1255   op->cls = rcb_cls;
1256
1257   req = (struct StateModifyRequest *) &op[1];
1258   op->msg = (struct GNUNET_MessageHeader *) req;
1259   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY);
1260   req->header.size = htons (sizeof (*req));
1261   req->channel_key = *channel_key;
1262   req->message_id = GNUNET_htonll (message_id);
1263   req->state_delta = GNUNET_htonll (state_delta);
1264
1265   op->op_id = get_next_op_id (h);
1266   req->op_id = GNUNET_htonll (op->op_id);
1267
1268   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1269   transmit_next (h);
1270
1271   return op;
1272   /* FIXME: only the last operation is returned,
1273    *        operation_cancel() should be able to cancel all of them.
1274    */
1275 }
1276
1277
1278 /**
1279  * Store synchronized state.
1280  *
1281  * @param h
1282  *        Handle for the PSYCstore.
1283  * @param channel_key
1284  *        The channel we are interested in.
1285  * @param max_state_message_id
1286  *        ID of the last stateful message before @a state_hash_message_id.
1287  * @param state_hash_message_id
1288  *        ID of the message that contains the state_hash PSYC header variable.
1289  * @param modifier_count
1290  *        Number of elements in the @a modifiers array.
1291  * @param modifiers
1292  *        Full state to store.
1293  * @param rcb
1294  *        Callback to call with the result of the operation.
1295  * @param rcb_cls
1296  *        Closure for the callback.
1297  *
1298  * @return Handle that can be used to cancel the operation.
1299  */
1300 struct GNUNET_PSYCSTORE_OperationHandle *
1301 GNUNET_PSYCSTORE_state_sync (struct GNUNET_PSYCSTORE_Handle *h,
1302                              const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1303                              uint64_t max_state_message_id,
1304                              uint64_t state_hash_message_id,
1305                              size_t modifier_count,
1306                              const struct GNUNET_PSYC_Modifier *modifiers,
1307                              GNUNET_PSYCSTORE_ResultCallback rcb,
1308                              void *rcb_cls)
1309 {
1310   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1311   size_t i;
1312
1313   for (i = 0; i < modifier_count; i++) {
1314     struct StateSyncRequest *req;
1315     uint16_t name_size = strlen (modifiers[i].name) + 1;
1316
1317     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size +
1318                         modifiers[i].value_size);
1319     op->h = h;
1320     op->res_cb = rcb;
1321     op->cls = rcb_cls;
1322
1323     req = (struct StateSyncRequest *) &op[1];
1324     op->msg = (struct GNUNET_MessageHeader *) req;
1325     req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC);
1326     req->header.size = htons (sizeof (*req) + name_size
1327                               + modifiers[i].value_size);
1328     req->channel_key = *channel_key;
1329     req->max_state_message_id = GNUNET_htonll (max_state_message_id);
1330     req->state_hash_message_id = GNUNET_htonll (state_hash_message_id);
1331     req->name_size = htons (name_size);
1332     req->flags
1333       = (0 == i)
1334       ? STATE_OP_FIRST
1335       : (modifier_count - 1 == i)
1336       ? STATE_OP_LAST
1337       : 0;
1338
1339     memcpy (&req[1], modifiers[i].name, name_size);
1340     memcpy ((char *) &req[1] + name_size, modifiers[i].value, modifiers[i].value_size);
1341
1342     op->op_id = get_next_op_id (h);
1343     req->op_id = GNUNET_htonll (op->op_id);
1344
1345     GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1346     transmit_next (h);
1347   }
1348   return op;
1349 }
1350
1351
1352 /**
1353  * Reset the state of a channel.
1354  *
1355  * Delete all state variables stored for the given channel.
1356  *
1357  * @param h
1358  *        Handle for the PSYCstore.
1359  * @param channel_key
1360  *        The channel we are interested in.
1361  * @param rcb
1362  *        Callback to call with the result of the operation.
1363  * @param rcb_cls
1364  *        Closure for the callback.
1365  *
1366  * @return Handle that can be used to cancel the operation.
1367  */
1368 struct GNUNET_PSYCSTORE_OperationHandle *
1369 GNUNET_PSYCSTORE_state_reset (struct GNUNET_PSYCSTORE_Handle *h,
1370                               const struct GNUNET_CRYPTO_EddsaPublicKey
1371                               *channel_key,
1372                               GNUNET_PSYCSTORE_ResultCallback rcb,
1373                               void *rcb_cls)
1374 {
1375   struct OperationRequest *req;
1376   struct GNUNET_PSYCSTORE_OperationHandle *
1377     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1378   op->h = h;
1379   op->res_cb = rcb;
1380   op->cls = rcb_cls;
1381
1382   req = (struct OperationRequest *) &op[1];
1383   op->msg = (struct GNUNET_MessageHeader *) req;
1384   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1385   req->header.size = htons (sizeof (*req));
1386   req->channel_key = *channel_key;
1387
1388   op->op_id = get_next_op_id (h);
1389   req->op_id = GNUNET_htonll (op->op_id);
1390
1391   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1392   transmit_next (h);
1393
1394   return op;
1395 }
1396
1397
1398
1399 /**
1400  * Update signed values of state variables in the state store.
1401  *
1402  * @param h
1403  *        Handle for the PSYCstore.
1404  * @param channel_key
1405  *        The channel we are interested in.
1406  * @param message_id
1407  *        Message ID that contained the state @a hash.
1408  * @param hash
1409  *        Hash of the serialized full state.
1410  * @param rcb
1411  *        Callback to call with the result of the operation.
1412  * @param rcb_cls
1413  *        Closure for the callback.
1414  */
1415 struct GNUNET_PSYCSTORE_OperationHandle *
1416 GNUNET_PSYCSTORE_state_hash_update (struct GNUNET_PSYCSTORE_Handle *h,
1417                                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1418                                     uint64_t message_id,
1419                                     const struct GNUNET_HashCode *hash,
1420                                     GNUNET_PSYCSTORE_ResultCallback rcb,
1421                                     void *rcb_cls)
1422 {
1423   struct StateHashUpdateRequest *req;
1424   struct GNUNET_PSYCSTORE_OperationHandle *
1425     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1426   op->h = h;
1427   op->res_cb = rcb;
1428   op->cls = rcb_cls;
1429
1430   req = (struct StateHashUpdateRequest *) &op[1];
1431   op->msg = (struct GNUNET_MessageHeader *) req;
1432   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1433   req->header.size = htons (sizeof (*req));
1434   req->channel_key = *channel_key;
1435   req->hash = *hash;
1436
1437   op->op_id = get_next_op_id (h);
1438   req->op_id = GNUNET_htonll (op->op_id);
1439
1440   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1441   transmit_next (h);
1442
1443   return op;
1444 }
1445
1446
1447 /**
1448  * Retrieve the best matching state variable.
1449  *
1450  * @param h
1451  *        Handle for the PSYCstore.
1452  * @param channel_key
1453  *        The channel we are interested in.
1454  * @param name
1455  *        Name of variable to match, the returned variable might be less specific.
1456  * @param scb
1457  *        Callback to return the matching state variable.
1458  * @param rcb
1459  *        Callback to call with the result of the operation.
1460  * @param cls
1461  *        Closure for the callbacks.
1462  *
1463  * @return Handle that can be used to cancel the operation.
1464  */
1465 struct GNUNET_PSYCSTORE_OperationHandle *
1466 GNUNET_PSYCSTORE_state_get (struct GNUNET_PSYCSTORE_Handle *h,
1467                             const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1468                             const char *name,
1469                             GNUNET_PSYCSTORE_StateCallback scb,
1470                             GNUNET_PSYCSTORE_ResultCallback rcb,
1471                             void *cls)
1472 {
1473   size_t name_size = strlen (name) + 1;
1474   struct OperationRequest *req;
1475   struct GNUNET_PSYCSTORE_OperationHandle *
1476     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1477   op->h = h;
1478   op->data_cb = (DataCallback) scb;
1479   op->res_cb = rcb;
1480   op->cls = cls;
1481
1482   req = (struct OperationRequest *) &op[1];
1483   op->msg = (struct GNUNET_MessageHeader *) req;
1484   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET);
1485   req->header.size = htons (sizeof (*req) + name_size);
1486   req->channel_key = *channel_key;
1487   memcpy (&req[1], name, name_size);
1488
1489   op->op_id = get_next_op_id (h);
1490   req->op_id = GNUNET_htonll (op->op_id);
1491
1492   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1493   transmit_next (h);
1494
1495   return op;
1496 }
1497
1498
1499
1500 /**
1501  * Retrieve all state variables for a channel with the given prefix.
1502  *
1503  * @param h
1504  *        Handle for the PSYCstore.
1505  * @param channel_key
1506  *        The channel we are interested in.
1507  * @param name_prefix
1508  *        Prefix of state variable names to match.
1509  * @param scb
1510  *        Callback to return matching state variables.
1511  * @param rcb
1512  *        Callback to call with the result of the operation.
1513  * @param cls
1514  *        Closure for the callbacks.
1515  *
1516  * @return Handle that can be used to cancel the operation.
1517  */
1518 struct GNUNET_PSYCSTORE_OperationHandle *
1519 GNUNET_PSYCSTORE_state_get_prefix (struct GNUNET_PSYCSTORE_Handle *h,
1520                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1521                                    const char *name_prefix,
1522                                    GNUNET_PSYCSTORE_StateCallback scb,
1523                                    GNUNET_PSYCSTORE_ResultCallback rcb,
1524                                    void *cls)
1525 {
1526   size_t name_size = strlen (name_prefix) + 1;
1527   struct OperationRequest *req;
1528   struct GNUNET_PSYCSTORE_OperationHandle *
1529     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1530   op->h = h;
1531   op->data_cb = (DataCallback) scb;
1532   op->res_cb = rcb;
1533   op->cls = cls;
1534
1535   req = (struct OperationRequest *) &op[1];
1536   op->msg = (struct GNUNET_MessageHeader *) req;
1537   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX);
1538   req->header.size = htons (sizeof (*req) + name_size);
1539   req->channel_key = *channel_key;
1540   memcpy (&req[1], name_prefix, name_size);
1541
1542   op->op_id = get_next_op_id (h);
1543   req->op_id = GNUNET_htonll (op->op_id);
1544
1545   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1546   transmit_next (h);
1547
1548   return op;
1549 }
1550
1551 /* end of psycstore_api.c */