src/psycstore/gnunet-service-psycstore.c

   1 /**
   2  * This file is part of GNUnet
   3  * Copyright (C) 2013 GNUnet e.V.
   4  *
   5  * GNUnet is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published
   7  * by the Free Software Foundation; either version 3, or (at your
   8  * option) any later version.
   9  *
  10  * GNUnet is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with GNUnet; see the file COPYING.  If not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20
  21 /**
  22  * @file psycstore/gnunet-service-psycstore.c
  23  * @brief 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_statistics_service.h"
  35 #include "gnunet_psyc_util_lib.h"
  36 #include "gnunet_psycstore_service.h"
  37 #include "gnunet_psycstore_plugin.h"
  38 #include "psycstore.h"
  39
  40
  41 /**
  42  * Handle to our current configuration.
  43  */
  44 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  45
  46 /**
  47  * Service handle.
  48  */
  49 static struct GNUNET_SERVICE_Handle *service;
  50
  51 /**
  52  * Handle to the statistics service.
  53  */
  54 static struct GNUNET_STATISTICS_Handle *stats;
  55
  56 /**
  57  * Database handle
  58  */
  59 static struct GNUNET_PSYCSTORE_PluginFunctions *db;
  60
  61 /**
  62  * Name of the database plugin
  63  */
  64 static char *db_lib_name;
  65
  66
  67 /**
  68  * Task run during shutdown.
  69  *
  70  * @param cls unused
  71  */
  72 static void
  73 shutdown_task (void *cls)
  74 {
  75   if (NULL != stats)
  76   {
  77     GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
  78     stats = NULL;
  79   }
  80   GNUNET_break (NULL == GNUNET_PLUGIN_unload (db_lib_name, db));
  81   GNUNET_free (db_lib_name);
  82   db_lib_name = NULL;
  83 }
  84
  85
  86 /**
  87  * Send a result code back to the client.
  88  *
  89  * @param client
  90  *        Client that should receive the result code.
  91  * @param result_code
  92  *        Code to transmit.
  93  * @param op_id
  94  *        Operation ID in network byte order.
  95  * @param err_msg
  96  *        Error message to include (or NULL for none).
  97  */
  98 static void
  99 send_result_code (struct GNUNET_SERVICE_Client *client,
 100                   uint64_t op_id,
 101                   int64_t result_code,
 102                   const char *err_msg)
 103 {
 104   struct OperationResult *res;
 105   size_t err_size = 0;
 106
 107   if (NULL != err_msg)
 108     err_size = strnlen (err_msg,
 109                         GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (*res) - 1) + 1;
 110   struct GNUNET_MQ_Envelope *
 111     env = GNUNET_MQ_msg_extra (res, err_size,
 112                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_CODE);
 113   res->result_code = GNUNET_htonll (result_code - INT64_MIN);
 114   res->op_id = op_id;
 115   if (0 < err_size)
 116   {
 117     GNUNET_memcpy (&res[1], err_msg, err_size);
 118     ((char *) &res[1])[err_size - 1] = '\0';
 119   }
 120
 121   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 122               "Sending result to client: %" PRId64 " (%s)\n",
 123               result_code, err_msg);
 124   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
 125 }
 126
 127
 128 enum
 129 {
 130   MEMBERSHIP_TEST_NOT_NEEDED = 0,
 131   MEMBERSHIP_TEST_NEEDED = 1,
 132   MEMBERSHIP_TEST_DONE = 2,
 133 } MessageMembershipTest;
 134
 135
 136 struct SendClosure
 137 {
 138   struct GNUNET_SERVICE_Client *client;
 139
 140   /**
 141    * Channel's public key.
 142    */
 143   struct GNUNET_CRYPTO_EddsaPublicKey channel_key;
 144
 145   /**
 146    * Slave's public key.
 147    */
 148   struct GNUNET_CRYPTO_EcdsaPublicKey slave_key;
 149
 150   /**
 151    * Operation ID.
 152    */
 153   uint64_t op_id;
 154
 155   /**
 156    * Membership test result.
 157    */
 158   int membership_test_result;
 159
 160   /**
 161    * Do membership test with @a slave_key before returning fragment?
 162    * @see enum MessageMembershipTest
 163    */
 164   uint8_t membership_test;
 165 };
 166
 167
 168 static int
 169 send_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 170                enum GNUNET_PSYCSTORE_MessageFlags flags)
 171 {
 172   struct SendClosure *sc = cls;
 173   struct FragmentResult *res;
 174
 175   if (MEMBERSHIP_TEST_NEEDED == sc->membership_test)
 176   {
 177     sc->membership_test = MEMBERSHIP_TEST_DONE;
 178     sc->membership_test_result
 179       = db->membership_test (db->cls, &sc->channel_key, &sc->slave_key,
 180                              GNUNET_ntohll (msg->message_id));
 181     switch (sc->membership_test_result)
 182     {
 183     case GNUNET_YES:
 184       break;
 185
 186     case GNUNET_NO:
 187     case GNUNET_SYSERR:
 188       return GNUNET_NO;
 189     }
 190   }
 191
 192   size_t msg_size = ntohs (msg->header.size);
 193
 194   struct GNUNET_MQ_Envelope *
 195     env = GNUNET_MQ_msg_extra (res, msg_size,
 196                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_FRAGMENT);
 197   res->op_id = sc->op_id;
 198   res->psycstore_flags = htonl (flags);
 199   GNUNET_memcpy (&res[1], msg, msg_size);
 200   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 201               "Sending fragment %ld to client\n",
 202               GNUNET_ntohll (msg->fragment_id));
 203   GNUNET_free (msg);
 204
 205   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (sc->client), env);
 206   return GNUNET_YES;
 207 }
 208
 209
 210 static int
 211 send_state_var (void *cls, const char *name,
 212                 const void *value, uint32_t value_size)
 213 {
 214   struct SendClosure *sc = cls;
 215   struct StateResult *res;
 216   size_t name_size = strlen (name) + 1;
 217
 218   /** @todo FIXME: split up value into 64k chunks */
 219
 220   struct GNUNET_MQ_Envelope *
 221     env = GNUNET_MQ_msg_extra (res, name_size + value_size,
 222                                GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_STATE);
 223   res->op_id = sc->op_id;
 224   res->name_size = htons (name_size);
 225   GNUNET_memcpy (&res[1], name, name_size);
 226   GNUNET_memcpy ((char *) &res[1] + name_size, value, value_size);
 227   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 228               "Sending state variable %s to client\n", name);
 229
 230   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (sc->client), env);
 231   return GNUNET_OK;
 232 }
 233
 234
 235 static void
 236 handle_client_membership_store (void *cls,
 237                                 const struct MembershipStoreRequest *req)
 238 {
 239   struct GNUNET_SERVICE_Client *client = cls;
 240
 241   int ret = db->membership_store (db->cls, &req->channel_key, &req->slave_key,
 242                                   req->did_join,
 243                                   GNUNET_ntohll (req->announced_at),
 244                                   GNUNET_ntohll (req->effective_since),
 245                                   GNUNET_ntohll (req->group_generation));
 246
 247   if (ret != GNUNET_OK)
 248     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 249                 _("Failed to store membership information!\n"));
 250
 251   send_result_code (client, req->op_id, ret, NULL);
 252   GNUNET_SERVICE_client_continue (client);
 253 }
 254
 255
 256 static void
 257 handle_client_membership_test (void *cls,
 258                                const struct MembershipTestRequest *req)
 259 {
 260   struct GNUNET_SERVICE_Client *client = cls;
 261
 262   int ret = db->membership_test (db->cls, &req->channel_key, &req->slave_key,
 263                                  GNUNET_ntohll (req->message_id));
 264   switch (ret)
 265   {
 266   case GNUNET_YES:
 267   case GNUNET_NO:
 268     break;
 269   default:
 270     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 271                 _("Failed to test membership!\n"));
 272   }
 273
 274   send_result_code (client, req->op_id, ret, NULL);
 275   GNUNET_SERVICE_client_continue (client);
 276 }
 277
 278
 279 static void
 280 handle_client_fragment_store (void *cls,
 281                               const struct FragmentStoreRequest *req)
 282 {
 283   struct GNUNET_SERVICE_Client *client = cls;
 284
 285   int ret = db->fragment_store (db->cls, &req->channel_key,
 286                                 (const struct GNUNET_MULTICAST_MessageHeader *)
 287                                 &req[1], ntohl (req->psycstore_flags));
 288
 289   if (ret != GNUNET_OK)
 290     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 291                 _("Failed to store fragment!\n"));
 292
 293   send_result_code (client, req->op_id, ret, NULL);
 294   GNUNET_SERVICE_client_continue (client);
 295 }
 296
 297
 298 static void
 299 handle_client_fragment_get (void *cls,
 300                             const struct FragmentGetRequest *req)
 301 {
 302   struct GNUNET_SERVICE_Client *client = cls;
 303
 304   struct SendClosure
 305     sc = { .op_id = req->op_id,
 306            .client = client,
 307            .channel_key = req->channel_key,
 308            .slave_key = req->slave_key,
 309            .membership_test = req->do_membership_test };
 310
 311   int64_t ret;
 312   uint64_t ret_frags = 0;
 313   uint64_t first_fragment_id = GNUNET_ntohll (req->first_fragment_id);
 314   uint64_t last_fragment_id = GNUNET_ntohll (req->last_fragment_id);
 315   uint64_t limit = GNUNET_ntohll (req->fragment_limit);
 316
 317   if (0 == limit)
 318     ret = db->fragment_get (db->cls, &req->channel_key,
 319                             first_fragment_id, last_fragment_id,
 320                             &ret_frags, send_fragment, &sc);
 321   else
 322     ret = db->fragment_get_latest (db->cls, &req->channel_key, limit,
 323                                    &ret_frags, send_fragment, &sc);
 324
 325   switch (ret)
 326   {
 327   case GNUNET_YES:
 328   case GNUNET_NO:
 329     if (MEMBERSHIP_TEST_DONE == sc.membership_test)
 330     {
 331       switch (sc.membership_test_result)
 332       {
 333       case GNUNET_YES:
 334         break;
 335
 336       case GNUNET_NO:
 337         ret = GNUNET_PSYCSTORE_MEMBERSHIP_TEST_FAILED;
 338         break;
 339
 340       case GNUNET_SYSERR:
 341         ret = GNUNET_SYSERR;
 342         break;
 343       }
 344     }
 345     break;
 346   default:
 347     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 348                 _("Failed to get fragment!\n"));
 349   }
 350   send_result_code (client, req->op_id, (ret < 0) ? ret : ret_frags, NULL);
 351   GNUNET_SERVICE_client_continue (client);
 352 }
 353
 354
 355 static void
 356 handle_client_message_get (void *cls,
 357                            const struct MessageGetRequest *req)
 358 {
 359   struct GNUNET_SERVICE_Client *client = cls;
 360
 361   uint16_t size = ntohs (req->header.size);
 362   const char *method_prefix = (const char *) &req[1];
 363
 364   if (size < sizeof (*req) + 1
 365       || '\0' != method_prefix[size - sizeof (*req) - 1])
 366   {
 367     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 368                 "Message get: invalid method prefix. size: %u < %u?\n",
 369                 size,
 370                 (unsigned int) (sizeof (*req) + 1));
 371     GNUNET_break (0);
 372     GNUNET_SERVICE_client_drop (client);
 373     return;
 374   }
 375
 376   struct SendClosure
 377     sc = { .op_id = req->op_id,
 378            .client = client,
 379            .channel_key = req->channel_key,
 380            .slave_key = req->slave_key,
 381            .membership_test = req->do_membership_test };
 382
 383   int64_t ret;
 384   uint64_t ret_frags = 0;
 385   uint64_t first_message_id = GNUNET_ntohll (req->first_message_id);
 386   uint64_t last_message_id = GNUNET_ntohll (req->last_message_id);
 387   uint64_t msg_limit = GNUNET_ntohll (req->message_limit);
 388   uint64_t frag_limit = GNUNET_ntohll (req->fragment_limit);
 389
 390   /** @todo method_prefix */
 391   if (0 == msg_limit)
 392     ret = db->message_get (db->cls, &req->channel_key,
 393                            first_message_id, last_message_id, frag_limit,
 394                            &ret_frags, send_fragment, &sc);
 395   else
 396     ret = db->message_get_latest (db->cls, &req->channel_key, msg_limit,
 397                                   &ret_frags, send_fragment, &sc);
 398
 399   switch (ret)
 400   {
 401   case GNUNET_YES:
 402   case GNUNET_NO:
 403     break;
 404   default:
 405     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 406                 _("Failed to get message!\n"));
 407   }
 408
 409   send_result_code (client, req->op_id, (ret < 0) ? ret : ret_frags, NULL);
 410   GNUNET_SERVICE_client_continue (client);
 411 }
 412
 413
 414 static void
 415 handle_client_message_get_fragment (void *cls,
 416                                     const struct MessageGetFragmentRequest *req)
 417 {
 418   struct GNUNET_SERVICE_Client *client = cls;
 419
 420   struct SendClosure
 421     sc = { .op_id = req->op_id, .client = client,
 422            .channel_key = req->channel_key, .slave_key = req->slave_key,
 423            .membership_test = req->do_membership_test };
 424
 425   int ret = db->message_get_fragment (db->cls, &req->channel_key,
 426                                       GNUNET_ntohll (req->message_id),
 427                                       GNUNET_ntohll (req->fragment_offset),
 428                                       &send_fragment, &sc);
 429   switch (ret)
 430   {
 431   case GNUNET_YES:
 432   case GNUNET_NO:
 433     break;
 434   default:
 435     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 436                 _("Failed to get message fragment!\n"));
 437   }
 438
 439   send_result_code (client, req->op_id, ret, NULL);
 440   GNUNET_SERVICE_client_continue (client);
 441 }
 442
 443
 444 static void
 445 handle_client_counters_get (void *cls,
 446                             const struct OperationRequest *req)
 447 {
 448   struct GNUNET_SERVICE_Client *client = cls;
 449
 450   struct CountersResult *res;
 451   struct GNUNET_MQ_Envelope *
 452     env = GNUNET_MQ_msg (res, GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_COUNTERS);
 453
 454   int ret = db->counters_message_get (db->cls, &req->channel_key,
 455                                       &res->max_fragment_id, &res->max_message_id,
 456                                       &res->max_group_generation);
 457   switch (ret)
 458   {
 459   case GNUNET_OK:
 460     ret = db->counters_state_get (db->cls, &req->channel_key,
 461                                   &res->max_state_message_id);
 462   case GNUNET_NO:
 463     break;
 464   default:
 465     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 466                 _("Failed to get master counters!\n"));
 467   }
 468
 469   res->result_code = htonl (ret);
 470   res->op_id = req->op_id;
 471   res->max_fragment_id = GNUNET_htonll (res->max_fragment_id);
 472   res->max_message_id = GNUNET_htonll (res->max_message_id);
 473   res->max_group_generation = GNUNET_htonll (res->max_group_generation);
 474   res->max_state_message_id = GNUNET_htonll (res->max_state_message_id);
 475
 476   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
 477   GNUNET_SERVICE_client_continue (client);
 478 }
 479
 480
 481 struct StateModifyClosure
 482 {
 483   const struct GNUNET_CRYPTO_EddsaPublicKey channel_key;
 484   struct GNUNET_PSYC_ReceiveHandle *recv;
 485   enum GNUNET_PSYC_MessageState msg_state;
 486   char mod_oper;
 487   char *mod_name;
 488   char *mod_value;
 489   uint32_t mod_value_size;
 490   uint32_t mod_value_remaining;
 491 };
 492
 493
 494 static void
 495 recv_state_message_part (void *cls,
 496                          const struct GNUNET_PSYC_MessageHeader *msg,
 497                          const struct GNUNET_MessageHeader *pmsg)
 498 {
 499   struct StateModifyClosure *scls = cls;
 500   uint16_t psize;
 501
 502   if (NULL == msg)
 503   { // FIXME: error on unknown message
 504     return;
 505   }
 506
 507   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 508               "recv_state_message_part()  message_id: %" PRIu64
 509               ", fragment_offset: %" PRIu64 ", flags: %u\n",
 510               GNUNET_ntohll (msg->message_id),
 511               GNUNET_ntohll (msg->fragment_offset),
 512               ntohl (msg->flags));
 513
 514   if (NULL == pmsg)
 515   {
 516     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 517     return;
 518   }
 519
 520   switch (ntohs (pmsg->type))
 521   {
 522   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
 523   {
 524     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_METHOD;
 525     break;
 526   }
 527
 528   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
 529   {
 530     struct GNUNET_PSYC_MessageModifier *
 531       pmod = (struct GNUNET_PSYC_MessageModifier *) pmsg;
 532     psize = ntohs (pmod->header.size);
 533     uint16_t name_size = ntohs (pmod->name_size);
 534     uint32_t value_size = ntohl (pmod->value_size);
 535
 536     const char *name = (const char *) &pmod[1];
 537     const void *value = name + name_size;
 538
 539     if (GNUNET_PSYC_OP_SET != pmod->oper)
 540     { // Apply non-transient operation.
 541       if (psize == sizeof (*pmod) + name_size + value_size)
 542       {
 543         db->state_modify_op (db->cls, &scls->channel_key,
 544                              pmod->oper, name, value, value_size);
 545       }
 546       else
 547       {
 548         scls->mod_oper = pmod->oper;
 549         scls->mod_name = GNUNET_malloc (name_size);
 550         GNUNET_memcpy (scls->mod_name, name, name_size);
 551
 552         scls->mod_value_size = value_size;
 553         scls->mod_value = GNUNET_malloc (scls->mod_value_size);
 554         scls->mod_value_remaining
 555           = scls->mod_value_size - (psize - sizeof (*pmod) - name_size);
 556         GNUNET_memcpy (scls->mod_value, value, value_size - scls->mod_value_remaining);
 557       }
 558     }
 559     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MODIFIER;
 560     break;
 561   }
 562
 563   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
 564     if (GNUNET_PSYC_OP_SET != scls->mod_oper)
 565     {
 566       if (scls->mod_value_remaining == 0)
 567       {
 568         GNUNET_break_op (0);
 569         scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 570       }
 571       psize = ntohs (pmsg->size);
 572       GNUNET_memcpy (scls->mod_value + (scls->mod_value_size - scls->mod_value_remaining),
 573               &pmsg[1], psize - sizeof (*pmsg));
 574       scls->mod_value_remaining -= psize - sizeof (*pmsg);
 575       if (0 == scls->mod_value_remaining)
 576       {
 577         db->state_modify_op (db->cls, &scls->channel_key,
 578                              scls->mod_oper, scls->mod_name,
 579                              scls->mod_value, scls->mod_value_size);
 580         GNUNET_free (scls->mod_name);
 581         GNUNET_free (scls->mod_value);
 582         scls->mod_oper = 0;
 583         scls->mod_name = NULL;
 584         scls->mod_value = NULL;
 585         scls->mod_value_size = 0;
 586       }
 587     }
 588     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MOD_CONT;
 589     break;
 590
 591   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
 592     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_DATA;
 593     break;
 594
 595   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
 596     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_END;
 597     break;
 598
 599   default:
 600     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
 601   }
 602 }
 603
 604
 605 static int
 606 recv_state_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 607                      enum GNUNET_PSYCSTORE_MessageFlags flags)
 608 {
 609   struct StateModifyClosure *scls = cls;
 610
 611   if (NULL == scls->recv)
 612   {
 613     scls->recv = GNUNET_PSYC_receive_create (NULL, recv_state_message_part,
 614                                              scls);
 615   }
 616
 617   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 618               "recv_state_fragment: %" PRIu64 "\n", GNUNET_ntohll (msg->fragment_id));
 619
 620   struct GNUNET_PSYC_MessageHeader *
 621     pmsg = GNUNET_PSYC_message_header_create (msg, flags);
 622   GNUNET_PSYC_receive_message (scls->recv, pmsg);
 623   GNUNET_free (pmsg);
 624
 625   return GNUNET_YES;
 626 }
 627
 628
 629 static void
 630 handle_client_state_modify (void *cls,
 631                             const struct StateModifyRequest *req)
 632 {
 633   struct GNUNET_SERVICE_Client *client = cls;
 634
 635   uint64_t message_id = GNUNET_ntohll (req->message_id);
 636   uint64_t state_delta = GNUNET_ntohll (req->state_delta);
 637   uint64_t ret_frags = 0;
 638   struct StateModifyClosure
 639     scls = { .channel_key = req->channel_key };
 640
 641   int ret = db->state_modify_begin (db->cls, &req->channel_key,
 642                                     message_id, state_delta);
 643
 644   if (GNUNET_OK != ret)
 645   {
 646     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 647                 _("Failed to begin modifying state: %d\n"), ret);
 648   }
 649   else
 650   {
 651     ret = db->message_get (db->cls, &req->channel_key,
 652                            message_id, message_id, 0,
 653                            &ret_frags, recv_state_fragment, &scls);
 654     if (GNUNET_OK != ret)
 655     {
 656       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 657                   _("Failed to modify state: %d\n"), ret);
 658       GNUNET_break (0);
 659     }
 660     else
 661     {
 662       if (GNUNET_OK != db->state_modify_end (db->cls, &req->channel_key, message_id))
 663       {
 664         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 665                     _("Failed to end modifying state!\n"));
 666         GNUNET_break (0);
 667       }
 668     }
 669     if (NULL != scls.recv)
 670     {
 671       GNUNET_PSYC_receive_destroy (scls.recv);
 672     }
 673   }
 674
 675   send_result_code (client, req->op_id, ret, NULL);
 676   GNUNET_SERVICE_client_continue (client);
 677 }
 678
 679
 680 static int
 681 check_client_state_sync (void *cls,
 682                          const struct StateSyncRequest *req)
 683 {
 684   return GNUNET_OK;
 685 }
 686
 687
 688 /** @todo FIXME: stop processing further state sync messages after an error */
 689 static void
 690 handle_client_state_sync (void *cls,
 691                           const struct StateSyncRequest *req)
 692 {
 693   struct GNUNET_SERVICE_Client *client = cls;
 694
 695   int ret = GNUNET_SYSERR;
 696   const char *name = (const char *) &req[1];
 697   uint16_t name_size = ntohs (req->name_size);
 698
 699   if (name_size <= 2 || '\0' != name[name_size - 1])
 700   {
 701     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 702                 _("Tried to set invalid state variable name!\n"));
 703     GNUNET_break_op (0);
 704   }
 705   else
 706   {
 707     ret = GNUNET_OK;
 708
 709     if (req->flags & STATE_OP_FIRST)
 710     {
 711       ret = db->state_sync_begin (db->cls, &req->channel_key);
 712     }
 713     if (ret != GNUNET_OK)
 714     {
 715       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 716                   _("Failed to begin synchronizing state!\n"));
 717     }
 718     else
 719     {
 720       ret = db->state_sync_assign (db->cls, &req->channel_key, name,
 721                                    name + ntohs (req->name_size),
 722                                    ntohs (req->header.size) - sizeof (*req)
 723                                    - ntohs (req->name_size));
 724     }
 725
 726     if (GNUNET_OK == ret && req->flags & STATE_OP_LAST)
 727     {
 728       ret = db->state_sync_end (db->cls, &req->channel_key,
 729                                 GNUNET_ntohll (req->max_state_message_id),
 730                                 GNUNET_ntohll (req->state_hash_message_id));
 731       if (ret != GNUNET_OK)
 732         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 733                     _("Failed to end synchronizing state!\n"));
 734     }
 735   }
 736   send_result_code (client, req->op_id, ret, NULL);
 737   GNUNET_SERVICE_client_continue (client);
 738 }
 739
 740
 741 static void
 742 handle_client_state_reset (void *cls,
 743                            const struct OperationRequest *req)
 744 {
 745   struct GNUNET_SERVICE_Client *client = cls;
 746
 747   int ret = db->state_reset (db->cls, &req->channel_key);
 748
 749   if (ret != GNUNET_OK)
 750     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 751                 _("Failed to reset state!\n"));
 752
 753   send_result_code (client, req->op_id, ret, NULL);
 754   GNUNET_SERVICE_client_continue (client);
 755 }
 756
 757
 758 static void
 759 handle_client_state_hash_update (void *cls,
 760                                  const struct StateHashUpdateRequest *req)
 761 {
 762   struct GNUNET_SERVICE_Client *client = cls;
 763
 764   int ret = db->state_reset (db->cls, &req->channel_key);
 765   if (ret != GNUNET_OK)
 766     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 767                 _("Failed to reset state!\n"));
 768
 769   send_result_code (client, req->op_id, ret, NULL);
 770   GNUNET_SERVICE_client_continue (client);
 771 }
 772
 773
 774 static int
 775 check_client_state_get (void *cls,
 776                         const struct OperationRequest *req)
 777 {
 778   return GNUNET_OK;
 779 }
 780
 781
 782 static void
 783 handle_client_state_get (void *cls,
 784                          const struct OperationRequest *req)
 785 {
 786   struct GNUNET_SERVICE_Client *client = cls;
 787
 788   struct SendClosure sc = { .op_id = req->op_id, .client = client };
 789   int64_t ret = GNUNET_SYSERR;
 790   const char *name = (const char *) &req[1];
 791   uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
 792
 793   if (name_size <= 2 || '\0' != name[name_size - 1])
 794   {
 795     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 796                 _("Tried to get invalid state variable name!\n"));
 797     GNUNET_break (0);
 798   }
 799   else
 800   {
 801     ret = db->state_get (db->cls, &req->channel_key, name,
 802                          &send_state_var, &sc);
 803     if (GNUNET_NO == ret && name_size >= 5) /* min: _a_b\0 */
 804     {
 805       char *p, *n = GNUNET_malloc (name_size);
 806       GNUNET_memcpy (n, name, name_size);
 807       while (&n[1] < (p = strrchr (n, '_')) && GNUNET_NO == ret)
 808       {
 809         *p = '\0';
 810         ret = db->state_get (db->cls, &req->channel_key, n,
 811                              &send_state_var, &sc);
 812       }
 813       GNUNET_free (n);
 814     }
 815   }
 816   switch (ret)
 817   {
 818   case GNUNET_OK:
 819   case GNUNET_NO:
 820     break;
 821   default:
 822     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 823                 _("Failed to get state variable!\n"));
 824   }
 825
 826   send_result_code (client, req->op_id, ret, NULL);
 827   GNUNET_SERVICE_client_continue (client);
 828 }
 829
 830
 831 static int
 832 check_client_state_get_prefix (void *cls,
 833                                const struct OperationRequest *req)
 834 {
 835   return GNUNET_OK;
 836 }
 837
 838
 839 static void
 840 handle_client_state_get_prefix (void *cls,
 841                                 const struct OperationRequest *req)
 842 {
 843   struct GNUNET_SERVICE_Client *client = cls;
 844
 845   struct SendClosure sc = { .op_id = req->op_id, .client = client };
 846   int64_t ret = GNUNET_SYSERR;
 847   const char *name = (const char *) &req[1];
 848   uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
 849
 850   if (name_size <= 1 || '\0' != name[name_size - 1])
 851   {
 852     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 853                 _("Tried to get invalid state variable name!\n"));
 854     GNUNET_break (0);
 855   }
 856   else
 857   {
 858     ret = db->state_get_prefix (db->cls, &req->channel_key, name,
 859                                 &send_state_var, &sc);
 860   }
 861   switch (ret)
 862   {
 863   case GNUNET_OK:
 864   case GNUNET_NO:
 865     break;
 866   default:
 867     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 868                 _("Failed to get state variable!\n"));
 869   }
 870
 871   send_result_code (client, req->op_id, ret, NULL);
 872   GNUNET_SERVICE_client_continue (client);
 873 }
 874
 875
 876 /**
 877  * A new client connected.
 878  *
 879  * @param cls NULL
 880  * @param client client to add
 881  * @param mq message queue for @a client
 882  * @return @a client
 883  */
 884 static void *
 885 client_notify_connect (void *cls,
 886                        struct GNUNET_SERVICE_Client *client,
 887                        struct GNUNET_MQ_Handle *mq)
 888 {
 889   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
 890
 891   return client;
 892 }
 893
 894
 895 /**
 896  * Called whenever a client is disconnected.
 897  * Frees our resources associated with that client.
 898  *
 899  * @param cls closure
 900  * @param client identification of the client
 901  * @param app_ctx must match @a client
 902  */
 903 static void
 904 client_notify_disconnect (void *cls,
 905                           struct GNUNET_SERVICE_Client *client,
 906                           void *app_ctx)
 907 {
 908 }
 909
 910
 911 /**
 912  * Initialize the PSYCstore service.
 913  *
 914  * @param cls Closure.
 915  * @param server The initialized server.
 916  * @param c Configuration to use.
 917  */
 918 static void
 919 run (void *cls,
 920      const struct GNUNET_CONFIGURATION_Handle *c,
 921      struct GNUNET_SERVICE_Handle *svc)
 922 {
 923   cfg = c;
 924   service = svc;
 925
 926   /* Loading database plugin */
 927   char *database;
 928   if (GNUNET_OK !=
 929       GNUNET_CONFIGURATION_get_value_string (cfg, "psycstore", "database",
 930                                              &database))
 931   {
 932     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
 933                                "psycstore",
 934                                "database");
 935   }
 936   else
 937   {
 938     GNUNET_asprintf (&db_lib_name,
 939                      "libgnunet_plugin_psycstore_%s",
 940                      database);
 941     db = GNUNET_PLUGIN_load (db_lib_name, (void *) cfg);
 942     GNUNET_free (database);
 943   }
 944   if (NULL == db)
 945   {
 946     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 947                 "Could not load database backend `%s'\n",
 948                 db_lib_name);
 949     GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
 950     return;
 951   }
 952
 953   stats = GNUNET_STATISTICS_create ("psycstore", cfg);
 954   GNUNET_SCHEDULER_add_shutdown (shutdown_task,
 955                                  NULL);
 956 }
 957
 958 /**
 959  * Define "main" method using service macro.
 960  */
 961 GNUNET_SERVICE_MAIN
 962 ("psycstore",
 963  GNUNET_SERVICE_OPTION_NONE,
 964  run,
 965  client_notify_connect,
 966  client_notify_disconnect,
 967  NULL,
 968  GNUNET_MQ_hd_fixed_size (client_membership_store,
 969                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_STORE,
 970                           struct MembershipStoreRequest,
 971                           NULL),
 972  GNUNET_MQ_hd_fixed_size (client_membership_test,
 973                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MEMBERSHIP_TEST,
 974                           struct MembershipTestRequest,
 975                           NULL),
 976  GNUNET_MQ_hd_fixed_size (client_fragment_store,
 977                         GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_STORE,
 978                         struct FragmentStoreRequest,
 979                         NULL),
 980  GNUNET_MQ_hd_fixed_size (client_fragment_get,
 981                           GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_GET,
 982                           struct FragmentGetRequest,
 983                           NULL),
 984  GNUNET_MQ_hd_fixed_size (client_message_get,
 985                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET,
 986                           struct MessageGetRequest,
 987                           NULL),
 988  GNUNET_MQ_hd_fixed_size (client_message_get_fragment,
 989                           GNUNET_MESSAGE_TYPE_PSYCSTORE_MESSAGE_GET_FRAGMENT,
 990                           struct MessageGetFragmentRequest,
 991                           NULL),
 992  GNUNET_MQ_hd_fixed_size (client_counters_get,
 993                           GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET,
 994                           struct OperationRequest,
 995                           NULL),
 996  GNUNET_MQ_hd_fixed_size (client_state_modify,
 997                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY,
 998                         struct StateModifyRequest,
 999                         NULL),
1000  GNUNET_MQ_hd_var_size (client_state_sync,
1001                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_SYNC,
1002                         struct StateSyncRequest,
1003                         NULL),
1004  GNUNET_MQ_hd_fixed_size (client_state_reset,
1005                           GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET,
1006                           struct OperationRequest,
1007                           NULL),
1008  GNUNET_MQ_hd_fixed_size (client_state_hash_update,
1009                           GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_HASH_UPDATE,
1010                           struct StateHashUpdateRequest,
1011                           NULL),
1012  GNUNET_MQ_hd_var_size (client_state_get,
1013                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET,
1014                         struct OperationRequest,
1015                         NULL),
1016  GNUNET_MQ_hd_var_size (client_state_get_prefix,
1017                         GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX,
1018                         struct OperationRequest,
1019                         NULL));
1020
1021 /* end of gnunet-service-psycstore.c */