src/psycstore/psycstore_api.c

   1 /*
   2  * This file is part of GNUnet
   3  * (C) 2013 Christian Grothoff (and other contributing authors)
   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., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, 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 = NULL;
 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 StateModifyRequest *smreq;
 306         const struct StateSyncRequest *ssreq;
 307         switch (ntohs (op->msg->type))
 308         {
 309         case GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY:
 310           smreq = (const struct StateModifyRequest *) op->msg;
 311           if (!(smreq->flags & STATE_OP_LAST
 312                 || GNUNET_OK != result_code))
 313             op->res_cb = NULL;
 314           break;
 315         case GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC:
 316           ssreq = (const struct StateSyncRequest *) op->msg;
 317           if (!(ssreq->flags & STATE_OP_LAST
 318                 || GNUNET_OK != result_code))
 319             op->res_cb = NULL;
 320           break;
 321         }
 322       }
 323       if (NULL != op->res_cb)
 324         op->res_cb (op->cls, result_code, str);
 325       GNUNET_free (op);
 326     }
 327     break;
 328
 329   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_COUNTERS:
 330     if (size != sizeof (struct CountersResult))
 331     {
 332       LOG (GNUNET_ERROR_TYPE_ERROR,
 333            "Received message of type %d with length %lu bytes. "
 334            "Expected %lu\n",
 335            type, size, sizeof (struct CountersResult));
 336       GNUNET_break (0);
 337       reschedule_connect (h);
 338       return;
 339     }
 340
 341     cres = (const struct CountersResult *) msg;
 342
 343     op = find_op_by_id (h, GNUNET_ntohll (cres->op_id));
 344     if (NULL == op)
 345     {
 346       LOG (GNUNET_ERROR_TYPE_DEBUG,
 347            "No callback registered for operation with ID %" PRIu64 ".\n",
 348            type, GNUNET_ntohll (cres->op_id));
 349     }
 350     else
 351     {
 352       GNUNET_CONTAINER_DLL_remove (h->op_head, h->op_tail, op);
 353       if (NULL != op->data_cb)
 354         ((GNUNET_PSYCSTORE_CountersCallback)
 355          op->data_cb) (op->cls,
 356                        ntohl (cres->result_code) + INT32_MIN,
 357                        GNUNET_ntohll (cres->max_fragment_id),
 358                        GNUNET_ntohll (cres->max_message_id),
 359                        GNUNET_ntohll (cres->max_group_generation),
 360                        GNUNET_ntohll (cres->max_state_message_id));
 361       GNUNET_free (op);
 362     }
 363     break;
 364
 365   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_FRAGMENT:
 366     if (size < sizeof (struct FragmentResult))
 367     {
 368       LOG (GNUNET_ERROR_TYPE_ERROR,
 369            "Received message of type %d with length %lu bytes. "
 370            "Expected >= %lu\n",
 371            type, size, sizeof (struct FragmentResult));
 372       GNUNET_break (0);
 373       reschedule_connect (h);
 374       return;
 375     }
 376
 377     fres = (const struct FragmentResult *) msg;
 378     struct GNUNET_MULTICAST_MessageHeader *mmsg =
 379       (struct GNUNET_MULTICAST_MessageHeader *) &fres[1];
 380     if (size != sizeof (struct FragmentResult) + ntohs (mmsg->header.size))
 381     {
 382       LOG (GNUNET_ERROR_TYPE_ERROR,
 383            "Received message of type %d with length %lu bytes. "
 384            "Expected = %lu\n",
 385            type, size,
 386            sizeof (struct FragmentResult) + ntohs (mmsg->header.size));
 387       GNUNET_break (0);
 388       reschedule_connect (h);
 389       return;
 390     }
 391
 392     op = find_op_by_id (h, GNUNET_ntohll (fres->op_id));
 393     if (NULL == op)
 394     {
 395       LOG (GNUNET_ERROR_TYPE_DEBUG,
 396            "No callback registered for operation with ID %" PRIu64 ".\n",
 397            type, GNUNET_ntohll (fres->op_id));
 398     }
 399     else
 400     {
 401       if (NULL != op->data_cb)
 402         ((GNUNET_PSYCSTORE_FragmentCallback)
 403          op->data_cb) (op->cls, mmsg, ntohl (fres->psycstore_flags));
 404     }
 405     break;
 406
 407   case GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_STATE:
 408     if (size < sizeof (struct StateResult))
 409     {
 410       LOG (GNUNET_ERROR_TYPE_ERROR,
 411            "Received message of type %d with length %lu bytes. "
 412            "Expected >= %lu\n",
 413            type, size, sizeof (struct StateResult));
 414       GNUNET_break (0);
 415       reschedule_connect (h);
 416       return;
 417     }
 418
 419     sres = (const struct StateResult *) msg;
 420     const char *name = (const char *) &sres[1];
 421     uint16_t name_size = ntohs (sres->name_size);
 422
 423     if (name_size <= 2 || '\0' != name[name_size - 1])
 424     {
 425       LOG (GNUNET_ERROR_TYPE_ERROR,
 426            "Received state result message (type %d) with invalid name.\n",
 427            type);
 428       GNUNET_break (0);
 429       reschedule_connect (h);
 430       return;
 431     }
 432
 433     op = find_op_by_id (h, GNUNET_ntohll (sres->op_id));
 434     if (NULL == op)
 435     {
 436       LOG (GNUNET_ERROR_TYPE_DEBUG,
 437            "No callback registered for operation with ID %" PRIu64 ".\n",
 438            type, GNUNET_ntohll (sres->op_id));
 439     }
 440     else
 441     {
 442       if (NULL != op->data_cb)
 443         ((GNUNET_PSYCSTORE_StateCallback)
 444          op->data_cb) (op->cls, name, (char *) &sres[1] + name_size,
 445                        ntohs (sres->header.size) - sizeof (*sres) - name_size);
 446     }
 447     break;
 448
 449   default:
 450     GNUNET_break (0);
 451     reschedule_connect (h);
 452     return;
 453   }
 454
 455   GNUNET_CLIENT_receive (h->client, &message_handler, h,
 456                          GNUNET_TIME_UNIT_FOREVER_REL);
 457 }
 458
 459
 460 /**
 461  * Transmit next message to service.
 462  *
 463  * @param cls The 'struct GNUNET_PSYCSTORE_Handle'.
 464  * @param size Number of bytes available in buf.
 465  * @param buf Where to copy the message.
 466  * @return Number of bytes copied to buf.
 467  */
 468 static size_t
 469 send_next_message (void *cls, size_t size, void *buf)
 470 {
 471   struct GNUNET_PSYCSTORE_Handle *h = cls;
 472   struct GNUNET_PSYCSTORE_OperationHandle *op = h->transmit_head;
 473   size_t ret;
 474
 475   h->th = NULL;
 476   if (NULL == op)
 477     return 0;
 478   ret = ntohs (op->msg->size);
 479   if (ret > size)
 480   {
 481     reschedule_connect (h);
 482     return 0;
 483   }
 484   LOG (GNUNET_ERROR_TYPE_DEBUG,
 485        "Sending message of type %d to PSYCstore service. ID: %" PRIu64 "\n",
 486        ntohs (op->msg->type), op->op_id);
 487   memcpy (buf, op->msg, ret);
 488
 489   GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 490
 491   if (NULL == op->res_cb && NULL == op->data_cb)
 492   {
 493     GNUNET_free (op);
 494   }
 495   else
 496   {
 497     GNUNET_CONTAINER_DLL_insert_tail (h->op_head, h->op_tail, op);
 498   }
 499
 500   if (NULL != h->transmit_head)
 501     transmit_next (h);
 502
 503   if (GNUNET_NO == h->in_receive)
 504   {
 505     h->in_receive = GNUNET_YES;
 506     GNUNET_CLIENT_receive (h->client, &message_handler, h,
 507                            GNUNET_TIME_UNIT_FOREVER_REL);
 508   }
 509   return ret;
 510 }
 511
 512
 513 /**
 514  * Schedule transmission of the next message from our queue.
 515  *
 516  * @param h PSYCstore handle.
 517  */
 518 static void
 519 transmit_next (struct GNUNET_PSYCSTORE_Handle *h)
 520 {
 521   if (NULL != h->th || NULL == h->client)
 522     return;
 523
 524   struct GNUNET_PSYCSTORE_OperationHandle *op = h->transmit_head;
 525   if (NULL == op)
 526     return;
 527
 528   h->th = GNUNET_CLIENT_notify_transmit_ready (h->client,
 529                                                ntohs (op->msg->size),
 530                                                GNUNET_TIME_UNIT_FOREVER_REL,
 531                                                GNUNET_NO,
 532                                                &send_next_message,
 533                                                h);
 534 }
 535
 536
 537 /**
 538  * Try again to connect to the PSYCstore service.
 539  *
 540  * @param cls Handle to the PSYCstore service.
 541  * @param tc Scheduler context.
 542  */
 543 static void
 544 reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 545 {
 546   struct GNUNET_PSYCSTORE_Handle *h = cls;
 547
 548   h->reconnect_task = NULL;
 549   LOG (GNUNET_ERROR_TYPE_DEBUG,
 550        "Connecting to PSYCstore service.\n");
 551   GNUNET_assert (NULL == h->client);
 552   h->client = GNUNET_CLIENT_connect ("psycstore", h->cfg);
 553   GNUNET_assert (NULL != h->client);
 554   transmit_next (h);
 555 }
 556
 557
 558 /**
 559  * Connect to the PSYCstore service.
 560  *
 561  * @param cfg The configuration to use
 562  * @return Handle to use
 563  */
 564 struct GNUNET_PSYCSTORE_Handle *
 565 GNUNET_PSYCSTORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
 566 {
 567   struct GNUNET_PSYCSTORE_Handle *h
 568     = GNUNET_new (struct GNUNET_PSYCSTORE_Handle);
 569   h->cfg = cfg;
 570   h->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
 571   h->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, h);
 572   return h;
 573 }
 574
 575
 576 /**
 577  * Disconnect from PSYCstore service
 578  *
 579  * @param h Handle to destroy
 580  */
 581 void
 582 GNUNET_PSYCSTORE_disconnect (struct GNUNET_PSYCSTORE_Handle *h)
 583 {
 584   GNUNET_assert (NULL != h);
 585   if (h->reconnect_task != NULL)
 586   {
 587     GNUNET_SCHEDULER_cancel (h->reconnect_task);
 588     h->reconnect_task = NULL;
 589   }
 590   if (NULL != h->th)
 591   {
 592     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 593     h->th = NULL;
 594   }
 595   if (NULL != h->client)
 596   {
 597     GNUNET_CLIENT_disconnect (h->client);
 598     h->client = NULL;
 599   }
 600   GNUNET_free (h);
 601 }
 602
 603
 604 /**
 605  * Cancel a PSYCstore operation. Note that the operation MAY still
 606  * be executed; this merely cancels the continuation; if the request
 607  * was already transmitted, the service may still choose to complete
 608  * the operation.
 609  *
 610  * @param op Operation to cancel.
 611  */
 612 void
 613 GNUNET_PSYCSTORE_operation_cancel (struct GNUNET_PSYCSTORE_OperationHandle *op)
 614 {
 615   struct GNUNET_PSYCSTORE_Handle *h = op->h;
 616
 617   if (h->transmit_head != NULL && (h->transmit_head != op || NULL == h->client))
 618   {
 619     /* request not active, can simply remove */
 620     GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 621     GNUNET_free (op);
 622     return;
 623   }
 624   if (NULL != h->th)
 625   {
 626     /* request active but not yet with service, can still abort */
 627     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 628     h->th = NULL;
 629     GNUNET_CONTAINER_DLL_remove (h->transmit_head, h->transmit_tail, op);
 630     GNUNET_free (op);
 631     transmit_next (h);
 632     return;
 633   }
 634   /* request active with service, simply ensure continuations are not called */
 635   op->res_cb = NULL;
 636   op->data_cb = NULL;
 637 }
 638
 639
 640 /**
 641  * Store join/leave events for a PSYC channel in order to be able to answer
 642  * membership test queries later.
 643  *
 644  * @param h
 645  *        Handle for the PSYCstore.
 646  * @param channel_key
 647  *        The channel where the event happened.
 648  * @param slave_key
 649  *        Public key of joining/leaving slave.
 650  * @param did_join
 651  *        #GNUNET_YES on join, #GNUNET_NO on part.
 652  * @param announced_at
 653  *        ID of the message that announced the membership change.
 654  * @param effective_since
 655  *        Message ID this membership change is in effect since.
 656  *        For joins it is <= announced_at, for parts it is always 0.
 657  * @param group_generation
 658  *        In case of a part, the last group generation the slave has access to.
 659  *        It has relevance when a larger message have fragments with different
 660  *        group generations.
 661  * @param rcb
 662  *        Callback to call with the result of the storage operation.
 663  * @param rcb_cls
 664  *        Closure for the callback.
 665  *
 666  * @return Operation handle that can be used to cancel the operation.
 667  */
 668 struct GNUNET_PSYCSTORE_OperationHandle *
 669 GNUNET_PSYCSTORE_membership_store (struct GNUNET_PSYCSTORE_Handle *h,
 670                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 671                                    const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 672                                    int did_join,
 673                                    uint64_t announced_at,
 674                                    uint64_t effective_since,
 675                                    uint64_t group_generation,
 676                                    GNUNET_PSYCSTORE_ResultCallback rcb,
 677                                    void *rcb_cls)
 678 {
 679   GNUNET_assert (NULL != h);
 680   GNUNET_assert (NULL != channel_key);
 681   GNUNET_assert (NULL != slave_key);
 682   GNUNET_assert (GNUNET_YES == did_join || GNUNET_NO == did_join);
 683   GNUNET_assert (did_join
 684                  ? effective_since <= announced_at
 685                  : effective_since == 0);
 686
 687   struct MembershipStoreRequest *req;
 688   struct GNUNET_PSYCSTORE_OperationHandle *
 689     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 690   op->h = h;
 691   op->res_cb = rcb;
 692   op->cls = rcb_cls;
 693
 694   req = (struct MembershipStoreRequest *) &op[1];
 695   op->msg = (struct GNUNET_MessageHeader *) req;
 696   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_STORE);
 697   req->header.size = htons (sizeof (*req));
 698   req->channel_key = *channel_key;
 699   req->slave_key = *slave_key;
 700   req->did_join = did_join;
 701   req->announced_at = GNUNET_htonll (announced_at);
 702   req->effective_since = GNUNET_htonll (effective_since);
 703   req->group_generation = GNUNET_htonll (group_generation);
 704
 705   op->op_id = get_next_op_id (h);
 706   req->op_id = GNUNET_htonll (op->op_id);
 707
 708   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 709   transmit_next (h);
 710
 711   return op;
 712 }
 713
 714
 715 /**
 716  * Test if a member was admitted to the channel at the given message ID.
 717  *
 718  * This is useful when relaying and replaying messages to check if a particular
 719  * slave has access to the message fragment with a given group generation.  It
 720  * is also used when handling join requests to determine whether the slave is
 721  * currently admitted to the channel.
 722  *
 723  * @param h
 724  *        Handle for the PSYCstore.
 725  * @param channel_key
 726  *        The channel we are interested in.
 727  * @param slave_key
 728  *        Public key of slave whose membership to check.
 729  * @param message_id
 730  *        Message ID for which to do the membership test.
 731  * @param group_generation
 732  *        Group generation of the fragment of the message to test.
 733  *        It has relevance if the message consists of multiple fragments with
 734  *        different group generations.
 735  * @param rcb
 736  *        Callback to call with the test result.
 737  * @param rcb_cls
 738  *        Closure for the callback.
 739  *
 740  * @return Operation handle that can be used to cancel the operation.
 741  */
 742 struct GNUNET_PSYCSTORE_OperationHandle *
 743 GNUNET_PSYCSTORE_membership_test (struct GNUNET_PSYCSTORE_Handle *h,
 744                                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 745                                   const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 746                                   uint64_t message_id,
 747                                   uint64_t group_generation,
 748                                   GNUNET_PSYCSTORE_ResultCallback rcb,
 749                                   void *rcb_cls)
 750 {
 751   struct MembershipTestRequest *req;
 752   struct GNUNET_PSYCSTORE_OperationHandle *
 753     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 754   op->h = h;
 755   op->res_cb = rcb;
 756   op->cls = rcb_cls;
 757
 758   req = (struct MembershipTestRequest *) &op[1];
 759   op->msg = (struct GNUNET_MessageHeader *) req;
 760   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_TEST);
 761   req->header.size = htons (sizeof (*req));
 762   req->channel_key = *channel_key;
 763   req->slave_key = *slave_key;
 764   req->message_id = GNUNET_htonll (message_id);
 765   req->group_generation = GNUNET_htonll (group_generation);
 766
 767   op->op_id = get_next_op_id (h);
 768   req->op_id = GNUNET_htonll (op->op_id);
 769
 770   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 771   transmit_next (h);
 772
 773   return op;
 774 }
 775
 776
 777 /**
 778  * Store a message fragment sent to a channel.
 779  *
 780  * @param h Handle for the PSYCstore.
 781  * @param channel_key The channel the message belongs to.
 782  * @param message Message to store.
 783  * @param psycstore_flags Flags indicating whether the PSYC message contains
 784  *        state modifiers.
 785  * @param rcb Callback to call with the result of the operation.
 786  * @param rcb_cls Closure for the callback.
 787  *
 788  * @return Handle that can be used to cancel the operation.
 789  */
 790 struct GNUNET_PSYCSTORE_OperationHandle *
 791 GNUNET_PSYCSTORE_fragment_store (struct GNUNET_PSYCSTORE_Handle *h,
 792                                  const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 793                                  const struct GNUNET_MULTICAST_MessageHeader *msg,
 794                                  enum GNUNET_PSYCSTORE_MessageFlags psycstore_flags,
 795                                  GNUNET_PSYCSTORE_ResultCallback rcb,
 796                                  void *rcb_cls)
 797 {
 798   uint16_t size = ntohs (msg->header.size);
 799   struct FragmentStoreRequest *req;
 800   struct GNUNET_PSYCSTORE_OperationHandle *
 801     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + size);
 802   op->h = h;
 803   op->res_cb = rcb;
 804   op->cls = rcb_cls;
 805
 806   req = (struct FragmentStoreRequest *) &op[1];
 807   op->msg = (struct GNUNET_MessageHeader *) req;
 808   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_STORE);
 809   req->header.size = htons (sizeof (*req) + size);
 810   req->channel_key = *channel_key;
 811   req->psycstore_flags = htonl (psycstore_flags);
 812   memcpy (&req[1], msg, size);
 813
 814   op->op_id = get_next_op_id (h);
 815   req->op_id = GNUNET_htonll (op->op_id);
 816
 817   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 818   transmit_next (h);
 819
 820   return op;
 821 }
 822
 823
 824 /**
 825  * Retrieve message fragments by fragment ID range.
 826  *
 827  * @param h
 828  *        Handle for the PSYCstore.
 829  * @param channel_key
 830  *        The channel we are interested in.
 831  * @param slave_key
 832  *        The slave requesting the fragment.  If not NULL, a membership test is
 833  *        performed first and the fragment is only returned if the slave has
 834  *        access to it.
 835  * @param first_fragment_id
 836  *        First fragment ID to retrieve.
 837  *        Use 0 to get the latest message fragment.
 838  * @param last_fragment_id
 839  *        Last consecutive fragment ID to retrieve.
 840  *        Use 0 to get the latest message fragment.
 841  * @param fragment_limit
 842  *        Maximum number of fragments to retrieve.
 843  * @param fragment_cb
 844  *        Callback to call with the retrieved fragments.
 845  * @param rcb
 846  *        Callback to call with the result of the operation.
 847  * @param cls
 848  *        Closure for the callbacks.
 849  *
 850  * @return Handle that can be used to cancel the operation.
 851  */
 852 struct GNUNET_PSYCSTORE_OperationHandle *
 853 GNUNET_PSYCSTORE_fragment_get (struct GNUNET_PSYCSTORE_Handle *h,
 854                                const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 855                                const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 856                                uint64_t first_fragment_id,
 857                                uint64_t last_fragment_id,
 858                                GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 859                                GNUNET_PSYCSTORE_ResultCallback rcb,
 860                                void *cls)
 861 {
 862   struct FragmentGetRequest *req;
 863   struct GNUNET_PSYCSTORE_OperationHandle *
 864     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 865   op->h = h;
 866   op->data_cb = (DataCallback) fragment_cb;
 867   op->res_cb = rcb;
 868   op->cls = cls;
 869
 870   req = (struct FragmentGetRequest *) &op[1];
 871   op->msg = (struct GNUNET_MessageHeader *) req;
 872   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET);
 873   req->header.size = htons (sizeof (*req));
 874   req->channel_key = *channel_key;
 875   req->first_fragment_id = GNUNET_htonll (first_fragment_id);
 876   req->last_fragment_id = GNUNET_htonll (last_fragment_id);
 877   if (NULL != slave_key)
 878   {
 879     req->slave_key = *slave_key;
 880     req->do_membership_test = GNUNET_YES;
 881   }
 882
 883   op->op_id = get_next_op_id (h);
 884   req->op_id = GNUNET_htonll (op->op_id);
 885
 886   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 887   transmit_next (h);
 888
 889   return op;
 890 }
 891
 892
 893 /**
 894  * Retrieve latest message fragments.
 895  *
 896  * @param h
 897  *        Handle for the PSYCstore.
 898  * @param channel_key
 899  *        The channel we are interested in.
 900  * @param slave_key
 901  *        The slave requesting the fragment.  If not NULL, a membership test is
 902  *        performed first and the fragment is only returned if the slave has
 903  *        access to it.
 904  * @param first_fragment_id
 905  *        First fragment ID to retrieve.
 906  *        Use 0 to get the latest message fragment.
 907  * @param last_fragment_id
 908  *        Last consecutive fragment ID to retrieve.
 909  *        Use 0 to get the latest message fragment.
 910  * @param fragment_limit
 911  *        Maximum number of fragments to retrieve.
 912  * @param fragment_cb
 913  *        Callback to call with the retrieved fragments.
 914  * @param rcb
 915  *        Callback to call with the result of the operation.
 916  * @param cls
 917  *        Closure for the callbacks.
 918  *
 919  * @return Handle that can be used to cancel the operation.
 920  */
 921 struct GNUNET_PSYCSTORE_OperationHandle *
 922 GNUNET_PSYCSTORE_fragment_get_latest (struct GNUNET_PSYCSTORE_Handle *h,
 923                                       const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 924                                       const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 925                                       uint64_t fragment_limit,
 926                                       GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 927                                       GNUNET_PSYCSTORE_ResultCallback rcb,
 928                                       void *cls)
 929 {
 930   struct FragmentGetRequest *req;
 931   struct GNUNET_PSYCSTORE_OperationHandle *
 932     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 933   op->h = h;
 934   op->data_cb = (DataCallback) fragment_cb;
 935   op->res_cb = rcb;
 936   op->cls = cls;
 937
 938   req = (struct FragmentGetRequest *) &op[1];
 939   op->msg = (struct GNUNET_MessageHeader *) req;
 940   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET);
 941   req->header.size = htons (sizeof (*req));
 942   req->channel_key = *channel_key;
 943   req->fragment_limit = GNUNET_ntohll (fragment_limit);
 944   if (NULL != slave_key)
 945   {
 946     req->slave_key = *slave_key;
 947     req->do_membership_test = GNUNET_YES;
 948   }
 949
 950   op->op_id = get_next_op_id (h);
 951   req->op_id = GNUNET_htonll (op->op_id);
 952
 953   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
 954   transmit_next (h);
 955
 956   return op;
 957 }
 958
 959
 960 /**
 961  * Retrieve all fragments of messages in a message ID range.
 962  *
 963  * @param h
 964  *        Handle for the PSYCstore.
 965  * @param channel_key
 966  *        The channel we are interested in.
 967  * @param slave_key
 968  *        The slave requesting the message.  If not NULL, a membership test is
 969  *        performed first and the message is only returned if the slave has
 970  *        access to it.
 971  * @param first_message_id
 972  *        First message ID to retrieve.
 973  *        Use 0 to get the latest message.
 974  * @param last_message_id
 975  *        Last consecutive message ID to retrieve.
 976  *        Use 0 to get the latest message.
 977  * @param fragment_cb
 978  *        Callback to call with the retrieved fragments.
 979  * @param rcb
 980  *        Callback to call with the result of the operation.
 981  * @param cls
 982  *        Closure for the callbacks.
 983  *
 984  * @return Handle that can be used to cancel the operation.
 985  */
 986 struct GNUNET_PSYCSTORE_OperationHandle *
 987 GNUNET_PSYCSTORE_message_get (struct GNUNET_PSYCSTORE_Handle *h,
 988                               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 989                               const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 990                               uint64_t first_message_id,
 991                               uint64_t last_message_id,
 992                               GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 993                               GNUNET_PSYCSTORE_ResultCallback rcb,
 994                               void *cls)
 995 {
 996   struct MessageGetRequest *req;
 997   struct GNUNET_PSYCSTORE_OperationHandle *
 998     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
 999   op->h = h;
1000   op->data_cb = (DataCallback) fragment_cb;
1001   op->res_cb = rcb;
1002   op->cls = cls;
1003
1004   req = (struct MessageGetRequest *) &op[1];
1005   op->msg = (struct GNUNET_MessageHeader *) req;
1006   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1007   req->header.size = htons (sizeof (*req));
1008   req->channel_key = *channel_key;
1009   req->first_message_id = GNUNET_htonll (first_message_id);
1010   req->last_message_id = GNUNET_htonll (last_message_id);
1011   if (NULL != slave_key)
1012   {
1013     req->slave_key = *slave_key;
1014     req->do_membership_test = GNUNET_YES;
1015   }
1016
1017   op->op_id = get_next_op_id (h);
1018   req->op_id = GNUNET_htonll (op->op_id);
1019
1020   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1021   transmit_next (h);
1022
1023   return op;
1024 }
1025
1026
1027 /**
1028  * Retrieve all fragments of the latest messages.
1029  *
1030  * @param h
1031  *        Handle for the PSYCstore.
1032  * @param channel_key
1033  *        The channel we are interested in.
1034  * @param slave_key
1035  *        The slave requesting the message.  If not NULL, a membership test is
1036  *        performed first and the message is only returned if the slave has
1037  *        access to it.
1038  * @param message_limit
1039  *        Maximum number of messages to retrieve.
1040  * @param fragment_cb
1041  *        Callback to call with the retrieved fragments.
1042  * @param rcb
1043  *        Callback to call with the result of the operation.
1044  * @param cls
1045  *        Closure for the callbacks.
1046  *
1047  * @return Handle that can be used to cancel the operation.
1048  */
1049 struct GNUNET_PSYCSTORE_OperationHandle *
1050 GNUNET_PSYCSTORE_message_get_latest (struct GNUNET_PSYCSTORE_Handle *h,
1051                                      const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1052                                      const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1053                                      uint64_t message_limit,
1054                                      GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1055                                      GNUNET_PSYCSTORE_ResultCallback rcb,
1056                                      void *cls)
1057 {
1058   struct MessageGetRequest *req;
1059   struct GNUNET_PSYCSTORE_OperationHandle *
1060     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1061   op->h = h;
1062   op->data_cb = (DataCallback) fragment_cb;
1063   op->res_cb = rcb;
1064   op->cls = cls;
1065
1066   req = (struct MessageGetRequest *) &op[1];
1067   op->msg = (struct GNUNET_MessageHeader *) req;
1068   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1069   req->header.size = htons (sizeof (*req));
1070   req->channel_key = *channel_key;
1071   req->message_limit = GNUNET_ntohll (message_limit);
1072   if (NULL != slave_key)
1073   {
1074     req->slave_key = *slave_key;
1075     req->do_membership_test = GNUNET_YES;
1076   }
1077
1078   op->op_id = get_next_op_id (h);
1079   req->op_id = GNUNET_htonll (op->op_id);
1080
1081   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1082   transmit_next (h);
1083
1084   return op;
1085 }
1086
1087
1088 /**
1089  * Retrieve a fragment of message specified by its message ID and fragment
1090  * offset.
1091  *
1092  * @param h
1093  *        Handle for the PSYCstore.
1094  * @param channel_key
1095  *        The channel we are interested in.
1096  * @param slave_key
1097  *        The slave requesting the message fragment.  If not NULL, a membership
1098  *        test is performed first and the message fragment is only returned
1099  *        if the slave has access to it.
1100  * @param message_id
1101  *        Message ID to retrieve.  Use 0 to get the latest message.
1102  * @param fragment_offset
1103  *        Offset of the fragment to retrieve.
1104  * @param fragment_cb
1105  *        Callback to call with the retrieved fragments.
1106  * @param result_cb
1107  *        Callback to call with the result of the operation.
1108  * @param cls
1109  *        Closure for the callbacks.
1110  *
1111  * @return Handle that can be used to cancel the operation.
1112  */
1113 struct GNUNET_PSYCSTORE_OperationHandle *
1114 GNUNET_PSYCSTORE_message_get_fragment (struct GNUNET_PSYCSTORE_Handle *h,
1115                                        const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1116                                        const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1117                                        uint64_t message_id,
1118                                        uint64_t fragment_offset,
1119                                        GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1120                                        GNUNET_PSYCSTORE_ResultCallback rcb,
1121                                        void *cls)
1122 {
1123   struct MessageGetFragmentRequest *req;
1124   struct GNUNET_PSYCSTORE_OperationHandle *
1125     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1126   op->h = h;
1127   op->data_cb = (DataCallback) fragment_cb;
1128   op->res_cb = rcb;
1129   op->cls = cls;
1130
1131   req = (struct MessageGetFragmentRequest *) &op[1];
1132   op->msg = (struct GNUNET_MessageHeader *) req;
1133   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET_FRAGMENT);
1134   req->header.size = htons (sizeof (*req));
1135   req->channel_key = *channel_key;
1136   req->message_id = GNUNET_htonll (message_id);
1137   req->fragment_offset = GNUNET_htonll (fragment_offset);
1138   if (NULL != slave_key)
1139   {
1140     req->slave_key = *slave_key;
1141     req->do_membership_test = GNUNET_YES;
1142   }
1143
1144   op->op_id = get_next_op_id (h);
1145   req->op_id = GNUNET_htonll (op->op_id);
1146
1147   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1148   transmit_next (h);
1149
1150   return op;
1151 }
1152
1153
1154 /**
1155  * Retrieve latest values of counters for a channel master.
1156  *
1157  * The current value of counters are needed when a channel master is restarted,
1158  * so that it can continue incrementing the counters from their last value.
1159  *
1160  * @param h Handle for the PSYCstore.
1161  * @param channel_key Public key that identifies the channel.
1162  * @param ccb Callback to call with the result.
1163  * @param ccb_cls Closure for the @a ccb callback.
1164  *
1165  * @return Handle that can be used to cancel the operation.
1166  */
1167 struct GNUNET_PSYCSTORE_OperationHandle *
1168 GNUNET_PSYCSTORE_counters_get (struct GNUNET_PSYCSTORE_Handle *h,
1169                                struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1170                                GNUNET_PSYCSTORE_CountersCallback ccb,
1171                                void *ccb_cls)
1172 {
1173   struct OperationRequest *req;
1174   struct GNUNET_PSYCSTORE_OperationHandle *
1175     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1176   op->h = h;
1177   op->data_cb = ccb;
1178   op->cls = ccb_cls;
1179
1180   req = (struct OperationRequest *) &op[1];
1181   op->msg = (struct GNUNET_MessageHeader *) req;
1182   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET);
1183   req->header.size = htons (sizeof (*req));
1184   req->channel_key = *channel_key;
1185
1186   op->op_id = get_next_op_id (h);
1187   req->op_id = GNUNET_htonll (op->op_id);
1188
1189   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1190   transmit_next (h);
1191
1192   return op;
1193 }
1194
1195
1196 /**
1197  * Apply modifiers of a message to the current channel state.
1198  *
1199  * An error is returned if there are missing messages containing state
1200  * operations before the current one.
1201  *
1202  * @param h Handle for the PSYCstore.
1203  * @param channel_key The channel we are interested in.
1204  * @param message_id ID of the message that contains the @a modifiers.
1205  * @param state_delta Value of the _state_delta PSYC header variable of the message.
1206  * @param modifier_count Number of elements in the @a modifiers array.
1207  * @param modifiers List of modifiers to apply.
1208  * @param rcb Callback to call with the result of the operation.
1209  * @param rcb_cls Closure for the @a rcb callback.
1210  *
1211  * @return Handle that can be used to cancel the operation.
1212  */
1213 struct GNUNET_PSYCSTORE_OperationHandle *
1214 GNUNET_PSYCSTORE_state_modify (struct GNUNET_PSYCSTORE_Handle *h,
1215                                const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1216                                uint64_t message_id,
1217                                uint64_t state_delta,
1218                                size_t modifier_count,
1219                                const struct GNUNET_ENV_Modifier *modifiers,
1220                                GNUNET_PSYCSTORE_ResultCallback rcb,
1221                                void *rcb_cls)
1222 {
1223   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1224   size_t i;
1225
1226   for (i = 0; i < modifier_count; i++) {
1227     struct StateModifyRequest *req;
1228     uint16_t name_size = strlen (modifiers[i].name) + 1;
1229
1230     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size +
1231                         modifiers[i].value_size);
1232     op->h = h;
1233     op->res_cb = rcb;
1234     op->cls = rcb_cls;
1235
1236     req = (struct StateModifyRequest *) &op[1];
1237     op->msg = (struct GNUNET_MessageHeader *) req;
1238     req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY);
1239     req->header.size = htons (sizeof (*req) + name_size
1240                               + modifiers[i].value_size);
1241     req->channel_key = *channel_key;
1242     req->message_id = GNUNET_htonll (message_id);
1243     req->state_delta = GNUNET_htonll (state_delta);
1244     req->oper = modifiers[i].oper;
1245     req->name_size = htons (name_size);
1246     req->flags
1247       = 0 == i
1248       ? STATE_OP_FIRST
1249       : modifier_count - 1 == i
1250       ? STATE_OP_LAST
1251       : 0;
1252
1253     memcpy (&req[1], modifiers[i].name, name_size);
1254     memcpy ((char *) &req[1] + name_size, modifiers[i].value, modifiers[i].value_size);
1255
1256     op->op_id = get_next_op_id (h);
1257     req->op_id = GNUNET_htonll (op->op_id);
1258
1259     GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1260     transmit_next (h);
1261   }
1262   return op;
1263   /* FIXME: only the last operation is returned,
1264    *        operation_cancel() should be able to cancel all of them.
1265    */
1266 }
1267
1268
1269 /**
1270  * Store synchronized state.
1271  *
1272  * @param h Handle for the PSYCstore.
1273  * @param channel_key The channel we are interested in.
1274  * @param message_id ID of the message that contains the state_hash PSYC header variable.
1275  * @param modifier_count Number of elements in the @a modifiers array.
1276  * @param modifiers Full state to store.
1277  * @param rcb Callback to call with the result of the operation.
1278  * @param rcb_cls Closure for the callback.
1279  *
1280  * @return Handle that can be used to cancel the operation.
1281  */
1282 struct GNUNET_PSYCSTORE_OperationHandle *
1283 GNUNET_PSYCSTORE_state_sync (struct GNUNET_PSYCSTORE_Handle *h,
1284                              const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1285                              uint64_t message_id,
1286                              size_t modifier_count,
1287                              const struct GNUNET_ENV_Modifier *modifiers,
1288                              GNUNET_PSYCSTORE_ResultCallback rcb,
1289                              void *rcb_cls)
1290 {
1291   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1292   size_t i;
1293
1294   for (i = 0; i < modifier_count; i++) {
1295     struct StateSyncRequest *req;
1296     uint16_t name_size = strlen (modifiers[i].name) + 1;
1297
1298     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size +
1299                         modifiers[i].value_size);
1300     op->h = h;
1301     op->res_cb = rcb;
1302     op->cls = rcb_cls;
1303
1304     req = (struct StateSyncRequest *) &op[1];
1305     op->msg = (struct GNUNET_MessageHeader *) req;
1306     req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC);
1307     req->header.size = htons (sizeof (*req) + name_size
1308                               + modifiers[i].value_size);
1309     req->channel_key = *channel_key;
1310     req->message_id = GNUNET_htonll (message_id);
1311     req->name_size = htons (name_size);
1312     req->flags
1313       = (0 == i)
1314       ? STATE_OP_FIRST
1315       : (modifier_count - 1 == i)
1316       ? STATE_OP_LAST
1317       : 0;
1318
1319     memcpy (&req[1], modifiers[i].name, name_size);
1320     memcpy ((char *) &req[1] + name_size, modifiers[i].value, modifiers[i].value_size);
1321
1322     op->op_id = get_next_op_id (h);
1323     req->op_id = GNUNET_htonll (op->op_id);
1324
1325     GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1326     transmit_next (h);
1327   }
1328   return op;
1329 }
1330
1331
1332 /**
1333  * Reset the state of a channel.
1334  *
1335  * Delete all state variables stored for the given channel.
1336  *
1337  * @param h Handle for the PSYCstore.
1338  * @param channel_key The channel we are interested in.
1339  * @param rcb Callback to call with the result of the operation.
1340  * @param rcb_cls Closure for the callback.
1341  *
1342  * @return Handle that can be used to cancel the operation.
1343  */
1344 struct GNUNET_PSYCSTORE_OperationHandle *
1345 GNUNET_PSYCSTORE_state_reset (struct GNUNET_PSYCSTORE_Handle *h,
1346                               const struct GNUNET_CRYPTO_EddsaPublicKey
1347                               *channel_key,
1348                               GNUNET_PSYCSTORE_ResultCallback rcb,
1349                               void *rcb_cls)
1350 {
1351   struct OperationRequest *req;
1352   struct GNUNET_PSYCSTORE_OperationHandle *
1353     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1354   op->h = h;
1355   op->res_cb = rcb;
1356   op->cls = rcb_cls;
1357
1358   req = (struct OperationRequest *) &op[1];
1359   op->msg = (struct GNUNET_MessageHeader *) req;
1360   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1361   req->header.size = htons (sizeof (*req));
1362   req->channel_key = *channel_key;
1363
1364   op->op_id = get_next_op_id (h);
1365   req->op_id = GNUNET_htonll (op->op_id);
1366
1367   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1368   transmit_next (h);
1369
1370   return op;
1371 }
1372
1373
1374
1375 /**
1376  * Update signed values of state variables in the state store.
1377  *
1378  * @param h Handle for the PSYCstore.
1379  * @param channel_key The channel we are interested in.
1380  * @param message_id Message ID that contained the state @a hash.
1381  * @param hash Hash of the serialized full state.
1382  * @param rcb Callback to call with the result of the operation.
1383  * @param rcb_cls Closure for the callback.
1384  *
1385  */
1386 struct GNUNET_PSYCSTORE_OperationHandle *
1387 GNUNET_PSYCSTORE_state_hash_update (struct GNUNET_PSYCSTORE_Handle *h,
1388                                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1389                                     uint64_t message_id,
1390                                     const struct GNUNET_HashCode *hash,
1391                                     GNUNET_PSYCSTORE_ResultCallback rcb,
1392                                     void *rcb_cls)
1393 {
1394   struct StateHashUpdateRequest *req;
1395   struct GNUNET_PSYCSTORE_OperationHandle *
1396     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1397   op->h = h;
1398   op->res_cb = rcb;
1399   op->cls = rcb_cls;
1400
1401   req = (struct StateHashUpdateRequest *) &op[1];
1402   op->msg = (struct GNUNET_MessageHeader *) req;
1403   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1404   req->header.size = htons (sizeof (*req));
1405   req->channel_key = *channel_key;
1406   req->hash = *hash;
1407
1408   op->op_id = get_next_op_id (h);
1409   req->op_id = GNUNET_htonll (op->op_id);
1410
1411   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1412   transmit_next (h);
1413
1414   return op;
1415 }
1416
1417
1418 /**
1419  * Retrieve the best matching state variable.
1420  *
1421  * @param h Handle for the PSYCstore.
1422  * @param channel_key The channel we are interested in.
1423  * @param name Name of variable to match, the returned variable might be less specific.
1424  * @param scb Callback to return the matching state variable.
1425  * @param rcb Callback to call with the result of the operation.
1426  * @param cls Closure for the callbacks.
1427  *
1428  * @return Handle that can be used to cancel the operation.
1429  */
1430 struct GNUNET_PSYCSTORE_OperationHandle *
1431 GNUNET_PSYCSTORE_state_get (struct GNUNET_PSYCSTORE_Handle *h,
1432                             const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1433                             const char *name,
1434                             GNUNET_PSYCSTORE_StateCallback scb,
1435                             GNUNET_PSYCSTORE_ResultCallback rcb,
1436                             void *cls)
1437 {
1438   size_t name_size = strlen (name) + 1;
1439   struct OperationRequest *req;
1440   struct GNUNET_PSYCSTORE_OperationHandle *
1441     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1442   op->h = h;
1443   op->data_cb = (DataCallback) scb;
1444   op->res_cb = rcb;
1445   op->cls = cls;
1446
1447   req = (struct OperationRequest *) &op[1];
1448   op->msg = (struct GNUNET_MessageHeader *) req;
1449   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET);
1450   req->header.size = htons (sizeof (*req) + name_size);
1451   req->channel_key = *channel_key;
1452   memcpy (&req[1], name, name_size);
1453
1454   op->op_id = get_next_op_id (h);
1455   req->op_id = GNUNET_htonll (op->op_id);
1456
1457   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1458   transmit_next (h);
1459
1460   return op;
1461 }
1462
1463
1464
1465 /**
1466  * Retrieve all state variables for a channel with the given prefix.
1467  *
1468  * @param h Handle for the PSYCstore.
1469  * @param channel_key The channel we are interested in.
1470  * @param name_prefix Prefix of state variable names to match.
1471  * @param scb Callback to return matching state variables.
1472  * @param rcb Callback to call with the result of the operation.
1473  * @param cls Closure for the callbacks.
1474  *
1475  * @return Handle that can be used to cancel the operation.
1476  */
1477 struct GNUNET_PSYCSTORE_OperationHandle *
1478 GNUNET_PSYCSTORE_state_get_prefix (struct GNUNET_PSYCSTORE_Handle *h,
1479                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1480                                    const char *name_prefix,
1481                                    GNUNET_PSYCSTORE_StateCallback scb,
1482                                    GNUNET_PSYCSTORE_ResultCallback rcb,
1483                                    void *cls)
1484 {
1485   size_t name_size = strlen (name_prefix) + 1;
1486   struct OperationRequest *req;
1487   struct GNUNET_PSYCSTORE_OperationHandle *
1488     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1489   op->h = h;
1490   op->data_cb = (DataCallback) scb;
1491   op->res_cb = rcb;
1492   op->cls = cls;
1493
1494   req = (struct OperationRequest *) &op[1];
1495   op->msg = (struct GNUNET_MessageHeader *) req;
1496   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX);
1497   req->header.size = htons (sizeof (*req) + name_size);
1498   req->channel_key = *channel_key;
1499   memcpy (&req[1], name_prefix, name_size);
1500
1501   op->op_id = get_next_op_id (h);
1502   req->op_id = GNUNET_htonll (op->op_id);
1503
1504   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1505   transmit_next (h);
1506
1507   return op;
1508 }
1509
1510 /* end of psycstore_api.c */