src/psycstore/psycstore_api.c

   1 /*
   2  * This file is part of GNUnet
   3  * Copyright (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., 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 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, size - sizeof (*opres));
 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.
 969  *        If not NULL, a membership test is performed first
 970  *        and the message is only returned if the slave has access to it.
 971  * @param first_message_id
 972  *        First message ID to retrieve.
 973  * @param last_message_id
 974  *        Last consecutive message ID to retrieve.
 975  * @param method_prefix
 976  *        Retrieve only messages with a matching method prefix.
 977  * @todo Implement method_prefix query.
 978  * @param fragment_cb
 979  *        Callback to call with the retrieved fragments.
 980  * @param result_cb
 981  *        Callback to call with the result of the operation.
 982  * @param cls
 983  *        Closure for the callbacks.
 984  *
 985  * @return Handle that can be used to cancel the operation.
 986  */
 987 struct GNUNET_PSYCSTORE_OperationHandle *
 988 GNUNET_PSYCSTORE_message_get (struct GNUNET_PSYCSTORE_Handle *h,
 989                               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
 990                               const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
 991                               uint64_t first_message_id,
 992                               uint64_t last_message_id,
 993                               const char *method_prefix,
 994                               GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
 995                               GNUNET_PSYCSTORE_ResultCallback rcb,
 996                               void *cls)
 997 {
 998   struct MessageGetRequest *req;
 999   if (NULL == method_prefix)
1000     method_prefix = "";
1001   uint16_t method_size = strnlen (method_prefix,
1002                                   GNUNET_SERVER_MAX_MESSAGE_SIZE
1003                                   - sizeof (*req)) + 1;
1004
1005   struct GNUNET_PSYCSTORE_OperationHandle *
1006     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1007   op->h = h;
1008   op->data_cb = (DataCallback) fragment_cb;
1009   op->res_cb = rcb;
1010   op->cls = cls;
1011
1012   req = (struct MessageGetRequest *) &op[1];
1013   op->msg = (struct GNUNET_MessageHeader *) req;
1014   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1015   req->header.size = htons (sizeof (*req) + method_size);
1016   req->channel_key = *channel_key;
1017   req->first_message_id = GNUNET_htonll (first_message_id);
1018   req->last_message_id = GNUNET_htonll (last_message_id);
1019   if (NULL != slave_key)
1020   {
1021     req->slave_key = *slave_key;
1022     req->do_membership_test = GNUNET_YES;
1023   }
1024   memcpy (&req[1], method_prefix, method_size);
1025   ((char *) &req[1])[method_size - 1] = '\0';
1026
1027   op->op_id = get_next_op_id (h);
1028   req->op_id = GNUNET_htonll (op->op_id);
1029
1030   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1031   transmit_next (h);
1032
1033   return op;
1034 }
1035
1036
1037 /**
1038  * Retrieve all fragments of the latest messages.
1039  *
1040  * @param h
1041  *        Handle for the PSYCstore.
1042  * @param channel_key
1043  *        The channel we are interested in.
1044  * @param slave_key
1045  *        The slave requesting the message.
1046  *        If not NULL, a membership test is performed first
1047  *        and the message is only returned if the slave has access to it.
1048  * @param message_limit
1049  *        Maximum number of messages to retrieve.
1050  * @param method_prefix
1051  *        Retrieve only messages with a matching method prefix.
1052  * @todo Implement method_prefix query.
1053  * @param fragment_cb
1054  *        Callback to call with the retrieved fragments.
1055  * @param result_cb
1056  *        Callback to call with the result of the operation.
1057  * @param cls
1058  *        Closure for the callbacks.
1059  *
1060  * @return Handle that can be used to cancel the operation.
1061  */
1062 struct GNUNET_PSYCSTORE_OperationHandle *
1063 GNUNET_PSYCSTORE_message_get_latest (struct GNUNET_PSYCSTORE_Handle *h,
1064                                      const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1065                                      const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1066                                      uint64_t message_limit,
1067                                      const char *method_prefix,
1068                                      GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1069                                      GNUNET_PSYCSTORE_ResultCallback rcb,
1070                                      void *cls)
1071 {
1072   struct MessageGetRequest *req;
1073
1074   if (NULL == method_prefix)
1075     method_prefix = "";
1076   uint16_t method_size = strnlen (method_prefix,
1077                                   GNUNET_SERVER_MAX_MESSAGE_SIZE
1078                                   - sizeof (*req)) + 1;
1079   GNUNET_assert ('\0' == method_prefix[method_size - 1]);
1080
1081   struct GNUNET_PSYCSTORE_OperationHandle *
1082     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + method_size);
1083   op->h = h;
1084   op->data_cb = (DataCallback) fragment_cb;
1085   op->res_cb = rcb;
1086   op->cls = cls;
1087
1088   req = (struct MessageGetRequest *) &op[1];
1089   op->msg = (struct GNUNET_MessageHeader *) req;
1090   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET);
1091   req->header.size = htons (sizeof (*req) + method_size);
1092   req->channel_key = *channel_key;
1093   req->message_limit = GNUNET_ntohll (message_limit);
1094   if (NULL != slave_key)
1095   {
1096     req->slave_key = *slave_key;
1097     req->do_membership_test = GNUNET_YES;
1098   }
1099
1100   op->op_id = get_next_op_id (h);
1101   req->op_id = GNUNET_htonll (op->op_id);
1102   memcpy (&req[1], method_prefix, method_size);
1103
1104   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1105   transmit_next (h);
1106
1107   return op;
1108 }
1109
1110
1111 /**
1112  * Retrieve a fragment of message specified by its message ID and fragment
1113  * offset.
1114  *
1115  * @param h
1116  *        Handle for the PSYCstore.
1117  * @param channel_key
1118  *        The channel we are interested in.
1119  * @param slave_key
1120  *        The slave requesting the message fragment.  If not NULL, a membership
1121  *        test is performed first and the message fragment is only returned
1122  *        if the slave has access to it.
1123  * @param message_id
1124  *        Message ID to retrieve.  Use 0 to get the latest message.
1125  * @param fragment_offset
1126  *        Offset of the fragment to retrieve.
1127  * @param fragment_cb
1128  *        Callback to call with the retrieved fragments.
1129  * @param result_cb
1130  *        Callback to call with the result of the operation.
1131  * @param cls
1132  *        Closure for the callbacks.
1133  *
1134  * @return Handle that can be used to cancel the operation.
1135  */
1136 struct GNUNET_PSYCSTORE_OperationHandle *
1137 GNUNET_PSYCSTORE_message_get_fragment (struct GNUNET_PSYCSTORE_Handle *h,
1138                                        const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1139                                        const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
1140                                        uint64_t message_id,
1141                                        uint64_t fragment_offset,
1142                                        GNUNET_PSYCSTORE_FragmentCallback fragment_cb,
1143                                        GNUNET_PSYCSTORE_ResultCallback rcb,
1144                                        void *cls)
1145 {
1146   struct MessageGetFragmentRequest *req;
1147   struct GNUNET_PSYCSTORE_OperationHandle *
1148     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1149   op->h = h;
1150   op->data_cb = (DataCallback) fragment_cb;
1151   op->res_cb = rcb;
1152   op->cls = cls;
1153
1154   req = (struct MessageGetFragmentRequest *) &op[1];
1155   op->msg = (struct GNUNET_MessageHeader *) req;
1156   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET_FRAGMENT);
1157   req->header.size = htons (sizeof (*req));
1158   req->channel_key = *channel_key;
1159   req->message_id = GNUNET_htonll (message_id);
1160   req->fragment_offset = GNUNET_htonll (fragment_offset);
1161   if (NULL != slave_key)
1162   {
1163     req->slave_key = *slave_key;
1164     req->do_membership_test = GNUNET_YES;
1165   }
1166
1167   op->op_id = get_next_op_id (h);
1168   req->op_id = GNUNET_htonll (op->op_id);
1169
1170   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1171   transmit_next (h);
1172
1173   return op;
1174 }
1175
1176
1177 /**
1178  * Retrieve latest values of counters for a channel master.
1179  *
1180  * The current value of counters are needed when a channel master is restarted,
1181  * so that it can continue incrementing the counters from their last value.
1182  *
1183  * @param h Handle for the PSYCstore.
1184  * @param channel_key Public key that identifies the channel.
1185  * @param ccb Callback to call with the result.
1186  * @param ccb_cls Closure for the @a ccb callback.
1187  *
1188  * @return Handle that can be used to cancel the operation.
1189  */
1190 struct GNUNET_PSYCSTORE_OperationHandle *
1191 GNUNET_PSYCSTORE_counters_get (struct GNUNET_PSYCSTORE_Handle *h,
1192                                struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1193                                GNUNET_PSYCSTORE_CountersCallback ccb,
1194                                void *ccb_cls)
1195 {
1196   struct OperationRequest *req;
1197   struct GNUNET_PSYCSTORE_OperationHandle *
1198     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1199   op->h = h;
1200   op->data_cb = ccb;
1201   op->cls = ccb_cls;
1202
1203   req = (struct OperationRequest *) &op[1];
1204   op->msg = (struct GNUNET_MessageHeader *) req;
1205   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET);
1206   req->header.size = htons (sizeof (*req));
1207   req->channel_key = *channel_key;
1208
1209   op->op_id = get_next_op_id (h);
1210   req->op_id = GNUNET_htonll (op->op_id);
1211
1212   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1213   transmit_next (h);
1214
1215   return op;
1216 }
1217
1218
1219 /**
1220  * Apply modifiers of a message to the current channel state.
1221  *
1222  * An error is returned if there are missing messages containing state
1223  * operations before the current one.
1224  *
1225  * @param h Handle for the PSYCstore.
1226  * @param channel_key The channel we are interested in.
1227  * @param message_id ID of the message that contains the @a modifiers.
1228  * @param state_delta Value of the _state_delta PSYC header variable of the message.
1229  * @param modifier_count Number of elements in the @a modifiers array.
1230  * @param modifiers List of modifiers to apply.
1231  * @param rcb Callback to call with the result of the operation.
1232  * @param rcb_cls Closure for the @a rcb callback.
1233  *
1234  * @return Handle that can be used to cancel the operation.
1235  */
1236 struct GNUNET_PSYCSTORE_OperationHandle *
1237 GNUNET_PSYCSTORE_state_modify (struct GNUNET_PSYCSTORE_Handle *h,
1238                                const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1239                                uint64_t message_id,
1240                                uint64_t state_delta,
1241                                size_t modifier_count,
1242                                const struct GNUNET_ENV_Modifier *modifiers,
1243                                GNUNET_PSYCSTORE_ResultCallback rcb,
1244                                void *rcb_cls)
1245 {
1246   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1247   size_t i;
1248
1249   for (i = 0; i < modifier_count; i++) {
1250     struct StateModifyRequest *req;
1251     uint16_t name_size = strlen (modifiers[i].name) + 1;
1252
1253     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size +
1254                         modifiers[i].value_size);
1255     op->h = h;
1256     op->res_cb = rcb;
1257     op->cls = rcb_cls;
1258
1259     req = (struct StateModifyRequest *) &op[1];
1260     op->msg = (struct GNUNET_MessageHeader *) req;
1261     req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY);
1262     req->header.size = htons (sizeof (*req) + name_size
1263                               + modifiers[i].value_size);
1264     req->channel_key = *channel_key;
1265     req->message_id = GNUNET_htonll (message_id);
1266     req->state_delta = GNUNET_htonll (state_delta);
1267     req->oper = modifiers[i].oper;
1268     req->name_size = htons (name_size);
1269     req->flags
1270       = 0 == i
1271       ? STATE_OP_FIRST
1272       : modifier_count - 1 == i
1273       ? STATE_OP_LAST
1274       : 0;
1275
1276     memcpy (&req[1], modifiers[i].name, name_size);
1277     memcpy ((char *) &req[1] + name_size, modifiers[i].value, modifiers[i].value_size);
1278
1279     op->op_id = get_next_op_id (h);
1280     req->op_id = GNUNET_htonll (op->op_id);
1281
1282     GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1283     transmit_next (h);
1284   }
1285   return op;
1286   /* FIXME: only the last operation is returned,
1287    *        operation_cancel() should be able to cancel all of them.
1288    */
1289 }
1290
1291
1292 /**
1293  * Store synchronized state.
1294  *
1295  * @param h Handle for the PSYCstore.
1296  * @param channel_key The channel we are interested in.
1297  * @param message_id ID of the message that contains the state_hash PSYC header variable.
1298  * @param modifier_count Number of elements in the @a modifiers array.
1299  * @param modifiers Full state to store.
1300  * @param rcb Callback to call with the result of the operation.
1301  * @param rcb_cls Closure for the callback.
1302  *
1303  * @return Handle that can be used to cancel the operation.
1304  */
1305 struct GNUNET_PSYCSTORE_OperationHandle *
1306 GNUNET_PSYCSTORE_state_sync (struct GNUNET_PSYCSTORE_Handle *h,
1307                              const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1308                              uint64_t message_id,
1309                              size_t modifier_count,
1310                              const struct GNUNET_ENV_Modifier *modifiers,
1311                              GNUNET_PSYCSTORE_ResultCallback rcb,
1312                              void *rcb_cls)
1313 {
1314   struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
1315   size_t i;
1316
1317   for (i = 0; i < modifier_count; i++) {
1318     struct StateSyncRequest *req;
1319     uint16_t name_size = strlen (modifiers[i].name) + 1;
1320
1321     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size +
1322                         modifiers[i].value_size);
1323     op->h = h;
1324     op->res_cb = rcb;
1325     op->cls = rcb_cls;
1326
1327     req = (struct StateSyncRequest *) &op[1];
1328     op->msg = (struct GNUNET_MessageHeader *) req;
1329     req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC);
1330     req->header.size = htons (sizeof (*req) + name_size
1331                               + modifiers[i].value_size);
1332     req->channel_key = *channel_key;
1333     req->message_id = GNUNET_htonll (message_id);
1334     req->name_size = htons (name_size);
1335     req->flags
1336       = (0 == i)
1337       ? STATE_OP_FIRST
1338       : (modifier_count - 1 == i)
1339       ? STATE_OP_LAST
1340       : 0;
1341
1342     memcpy (&req[1], modifiers[i].name, name_size);
1343     memcpy ((char *) &req[1] + name_size, modifiers[i].value, modifiers[i].value_size);
1344
1345     op->op_id = get_next_op_id (h);
1346     req->op_id = GNUNET_htonll (op->op_id);
1347
1348     GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1349     transmit_next (h);
1350   }
1351   return op;
1352 }
1353
1354
1355 /**
1356  * Reset the state of a channel.
1357  *
1358  * Delete all state variables stored for the given channel.
1359  *
1360  * @param h Handle for the PSYCstore.
1361  * @param channel_key The channel we are interested in.
1362  * @param rcb Callback to call with the result of the operation.
1363  * @param rcb_cls Closure for the callback.
1364  *
1365  * @return Handle that can be used to cancel the operation.
1366  */
1367 struct GNUNET_PSYCSTORE_OperationHandle *
1368 GNUNET_PSYCSTORE_state_reset (struct GNUNET_PSYCSTORE_Handle *h,
1369                               const struct GNUNET_CRYPTO_EddsaPublicKey
1370                               *channel_key,
1371                               GNUNET_PSYCSTORE_ResultCallback rcb,
1372                               void *rcb_cls)
1373 {
1374   struct OperationRequest *req;
1375   struct GNUNET_PSYCSTORE_OperationHandle *
1376     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1377   op->h = h;
1378   op->res_cb = rcb;
1379   op->cls = rcb_cls;
1380
1381   req = (struct OperationRequest *) &op[1];
1382   op->msg = (struct GNUNET_MessageHeader *) req;
1383   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1384   req->header.size = htons (sizeof (*req));
1385   req->channel_key = *channel_key;
1386
1387   op->op_id = get_next_op_id (h);
1388   req->op_id = GNUNET_htonll (op->op_id);
1389
1390   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1391   transmit_next (h);
1392
1393   return op;
1394 }
1395
1396
1397
1398 /**
1399  * Update signed values of state variables in the state store.
1400  *
1401  * @param h Handle for the PSYCstore.
1402  * @param channel_key The channel we are interested in.
1403  * @param message_id Message ID that contained the state @a hash.
1404  * @param hash Hash of the serialized full state.
1405  * @param rcb Callback to call with the result of the operation.
1406  * @param rcb_cls Closure for the callback.
1407  *
1408  */
1409 struct GNUNET_PSYCSTORE_OperationHandle *
1410 GNUNET_PSYCSTORE_state_hash_update (struct GNUNET_PSYCSTORE_Handle *h,
1411                                     const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1412                                     uint64_t message_id,
1413                                     const struct GNUNET_HashCode *hash,
1414                                     GNUNET_PSYCSTORE_ResultCallback rcb,
1415                                     void *rcb_cls)
1416 {
1417   struct StateHashUpdateRequest *req;
1418   struct GNUNET_PSYCSTORE_OperationHandle *
1419     op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
1420   op->h = h;
1421   op->res_cb = rcb;
1422   op->cls = rcb_cls;
1423
1424   req = (struct StateHashUpdateRequest *) &op[1];
1425   op->msg = (struct GNUNET_MessageHeader *) req;
1426   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
1427   req->header.size = htons (sizeof (*req));
1428   req->channel_key = *channel_key;
1429   req->hash = *hash;
1430
1431   op->op_id = get_next_op_id (h);
1432   req->op_id = GNUNET_htonll (op->op_id);
1433
1434   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1435   transmit_next (h);
1436
1437   return op;
1438 }
1439
1440
1441 /**
1442  * Retrieve the best matching state variable.
1443  *
1444  * @param h Handle for the PSYCstore.
1445  * @param channel_key The channel we are interested in.
1446  * @param name Name of variable to match, the returned variable might be less specific.
1447  * @param scb Callback to return the matching state variable.
1448  * @param rcb Callback to call with the result of the operation.
1449  * @param cls Closure for the callbacks.
1450  *
1451  * @return Handle that can be used to cancel the operation.
1452  */
1453 struct GNUNET_PSYCSTORE_OperationHandle *
1454 GNUNET_PSYCSTORE_state_get (struct GNUNET_PSYCSTORE_Handle *h,
1455                             const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1456                             const char *name,
1457                             GNUNET_PSYCSTORE_StateCallback scb,
1458                             GNUNET_PSYCSTORE_ResultCallback rcb,
1459                             void *cls)
1460 {
1461   size_t name_size = strlen (name) + 1;
1462   struct OperationRequest *req;
1463   struct GNUNET_PSYCSTORE_OperationHandle *
1464     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1465   op->h = h;
1466   op->data_cb = (DataCallback) scb;
1467   op->res_cb = rcb;
1468   op->cls = cls;
1469
1470   req = (struct OperationRequest *) &op[1];
1471   op->msg = (struct GNUNET_MessageHeader *) req;
1472   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET);
1473   req->header.size = htons (sizeof (*req) + name_size);
1474   req->channel_key = *channel_key;
1475   memcpy (&req[1], name, name_size);
1476
1477   op->op_id = get_next_op_id (h);
1478   req->op_id = GNUNET_htonll (op->op_id);
1479
1480   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1481   transmit_next (h);
1482
1483   return op;
1484 }
1485
1486
1487
1488 /**
1489  * Retrieve all state variables for a channel with the given prefix.
1490  *
1491  * @param h Handle for the PSYCstore.
1492  * @param channel_key The channel we are interested in.
1493  * @param name_prefix Prefix of state variable names to match.
1494  * @param scb Callback to return matching state variables.
1495  * @param rcb Callback to call with the result of the operation.
1496  * @param cls Closure for the callbacks.
1497  *
1498  * @return Handle that can be used to cancel the operation.
1499  */
1500 struct GNUNET_PSYCSTORE_OperationHandle *
1501 GNUNET_PSYCSTORE_state_get_prefix (struct GNUNET_PSYCSTORE_Handle *h,
1502                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
1503                                    const char *name_prefix,
1504                                    GNUNET_PSYCSTORE_StateCallback scb,
1505                                    GNUNET_PSYCSTORE_ResultCallback rcb,
1506                                    void *cls)
1507 {
1508   size_t name_size = strlen (name_prefix) + 1;
1509   struct OperationRequest *req;
1510   struct GNUNET_PSYCSTORE_OperationHandle *
1511     op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
1512   op->h = h;
1513   op->data_cb = (DataCallback) scb;
1514   op->res_cb = rcb;
1515   op->cls = cls;
1516
1517   req = (struct OperationRequest *) &op[1];
1518   op->msg = (struct GNUNET_MessageHeader *) req;
1519   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX);
1520   req->header.size = htons (sizeof (*req) + name_size);
1521   req->channel_key = *channel_key;
1522   memcpy (&req[1], name_prefix, name_size);
1523
1524   op->op_id = get_next_op_id (h);
1525   req->op_id = GNUNET_htonll (op->op_id);
1526
1527   GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
1528   transmit_next (h);
1529
1530   return op;
1531 }
1532
1533 /* end of psycstore_api.c */