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