src/peerstore/peerstore_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file peerstore/peerstore_api.c
  23  * @brief API for peerstore
  24  * @author Omar Tarabai
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "peerstore.h"
  29 #include "peerstore_common.h"
  30
  31 #define LOG(kind,...) GNUNET_log_from (kind, "peerstore-api",__VA_ARGS__)
  32
  33 /******************************************************************************/
  34 /************************      DATA STRUCTURES     ****************************/
  35 /******************************************************************************/
  36
  37 /**
  38  * Handle to the PEERSTORE service.
  39  */
  40 struct GNUNET_PEERSTORE_Handle
  41 {
  42
  43   /**
  44    * Our configuration.
  45    */
  46   const struct GNUNET_CONFIGURATION_Handle *cfg;
  47
  48   /**
  49    * Connection to the service.
  50    */
  51   struct GNUNET_CLIENT_Connection *client;
  52
  53   /**
  54    * Message queue
  55    */
  56   struct GNUNET_MQ_Handle *mq;
  57
  58   /**
  59    * Head of active STORE requests.
  60    */
  61   struct GNUNET_PEERSTORE_StoreContext *store_head;
  62
  63   /**
  64    * Tail of active STORE requests.
  65    */
  66   struct GNUNET_PEERSTORE_StoreContext *store_tail;
  67
  68   /**
  69    * Head of active ITERATE requests.
  70    */
  71   struct GNUNET_PEERSTORE_IterateContext *iterate_head;
  72
  73   /**
  74    * Tail of active ITERATE requests.
  75    */
  76   struct GNUNET_PEERSTORE_IterateContext *iterate_tail;
  77
  78   /**
  79    * Hashmap of watch requests
  80    */
  81   struct GNUNET_CONTAINER_MultiHashMap *watches;
  82
  83 };
  84
  85 /**
  86  * Context for a store request
  87  */
  88 struct GNUNET_PEERSTORE_StoreContext
  89 {
  90   /**
  91    * Kept in a DLL.
  92    */
  93   struct GNUNET_PEERSTORE_StoreContext *next;
  94
  95   /**
  96    * Kept in a DLL.
  97    */
  98   struct GNUNET_PEERSTORE_StoreContext *prev;
  99
 100   /**
 101    * Handle to the PEERSTORE service.
 102    */
 103   struct GNUNET_PEERSTORE_Handle *h;
 104
 105   /**
 106    * MQ Envelope with store request message
 107    */
 108   struct GNUNET_MQ_Envelope *ev;
 109
 110   /**
 111    * Continuation called with service response
 112    */
 113   GNUNET_PEERSTORE_Continuation cont;
 114
 115   /**
 116    * Closure for 'cont'
 117    */
 118   void *cont_cls;
 119
 120 };
 121
 122 /**
 123  * Context for a iterate request
 124  */
 125 struct GNUNET_PEERSTORE_IterateContext
 126 {
 127   /**
 128    * Kept in a DLL.
 129    */
 130   struct GNUNET_PEERSTORE_IterateContext *next;
 131
 132   /**
 133    * Kept in a DLL.
 134    */
 135   struct GNUNET_PEERSTORE_IterateContext *prev;
 136
 137   /**
 138    * Handle to the PEERSTORE service.
 139    */
 140   struct GNUNET_PEERSTORE_Handle *h;
 141
 142   /**
 143    * MQ Envelope with iterate request message
 144    */
 145   struct GNUNET_MQ_Envelope *ev;
 146
 147   /**
 148    * Callback with each matching record
 149    */
 150   GNUNET_PEERSTORE_Processor callback;
 151
 152   /**
 153    * Closure for 'callback'
 154    */
 155   void *callback_cls;
 156
 157   /**
 158    * #GNUNET_YES / #GNUNET_NO
 159    * if sent, cannot be canceled
 160    */
 161   int request_sent;
 162
 163   /**
 164    * Task identifier for the function called
 165    * on iterate request timeout
 166    */
 167   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 168
 169 };
 170
 171 /**
 172  * Context for a watch request
 173  */
 174 struct GNUNET_PEERSTORE_WatchContext
 175 {
 176   /**
 177    * Kept in a DLL.
 178    */
 179   struct GNUNET_PEERSTORE_WatchContext *next;
 180
 181   /**
 182    * Kept in a DLL.
 183    */
 184   struct GNUNET_PEERSTORE_WatchContext *prev;
 185
 186   /**
 187    * Handle to the PEERSTORE service.
 188    */
 189   struct GNUNET_PEERSTORE_Handle *h;
 190
 191   /**
 192    * MQ Envelope with watch request message
 193    */
 194   struct GNUNET_MQ_Envelope *ev;
 195
 196   /**
 197    * Callback with each record received
 198    */
 199   GNUNET_PEERSTORE_Processor callback;
 200
 201   /**
 202    * Closure for 'callback'
 203    */
 204   void *callback_cls;
 205
 206   /**
 207    * Hash of the combined key
 208    */
 209   struct GNUNET_HashCode keyhash;
 210
 211   /**
 212    * #GNUNET_YES / #GNUNET_NO
 213    * if sent, cannot be canceled
 214    */
 215   int request_sent;
 216
 217 };
 218
 219 /******************************************************************************/
 220 /*******************             DECLARATIONS             *********************/
 221 /******************************************************************************/
 222
 223 /**
 224  * When a response for iterate request is received
 225  *
 226  * @param cls a 'struct GNUNET_PEERSTORE_Handle *'
 227  * @param msg message received, NULL on timeout or fatal error
 228  */
 229 void handle_iterate_result (void *cls, const struct GNUNET_MessageHeader *msg);
 230
 231 /**
 232  * When a watch record is received
 233  *
 234  * @param cls a 'struct GNUNET_PEERSTORE_Handle *'
 235  * @param msg message received, NULL on timeout or fatal error
 236  */
 237 void handle_watch_result (void *cls, const struct GNUNET_MessageHeader *msg);
 238
 239 /**
 240  * Close the existing connection to PEERSTORE and reconnect.
 241  *
 242  * @param h handle to the service
 243  */
 244 static void
 245 reconnect (struct GNUNET_PEERSTORE_Handle *h);
 246
 247 /**
 248  * MQ message handlers
 249  */
 250 static const struct GNUNET_MQ_MessageHandler mq_handlers[] = {
 251     {&handle_iterate_result, GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_RECORD, 0},
 252     {&handle_iterate_result, GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_END, sizeof(struct GNUNET_MessageHeader)},
 253     {&handle_watch_result, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_RECORD, 0},
 254     GNUNET_MQ_HANDLERS_END
 255 };
 256
 257 /******************************************************************************/
 258 /*******************         CONNECTION FUNCTIONS         *********************/
 259 /******************************************************************************/
 260
 261 static void
 262 handle_client_error (void *cls, enum GNUNET_MQ_Error error)
 263 {
 264   struct GNUNET_PEERSTORE_Handle *h = cls;
 265
 266   LOG(GNUNET_ERROR_TYPE_ERROR, "Received an error notification from MQ of type: %d\n", error);
 267   reconnect(h);
 268 }
 269
 270 /**
 271  * Close the existing connection to PEERSTORE and reconnect.
 272  *
 273  * @param h handle to the service
 274  */
 275 static void
 276 reconnect (struct GNUNET_PEERSTORE_Handle *h)
 277 {
 278   LOG(GNUNET_ERROR_TYPE_DEBUG, "Reconnecting...\n");
 279   if (NULL != h->mq)
 280   {
 281     GNUNET_MQ_destroy(h->mq);
 282     h->mq = NULL;
 283   }
 284   if (NULL != h->client)
 285   {
 286     GNUNET_CLIENT_disconnect (h->client);
 287     h->client = NULL;
 288   }
 289   h->client = GNUNET_CLIENT_connect ("peerstore", h->cfg);
 290   //FIXME: retry connecting if fails again (client == NULL)
 291   h->mq = GNUNET_MQ_queue_for_connection_client(h->client,
 292       mq_handlers,
 293       &handle_client_error,
 294       h);
 295   //FIXME: resend pending requests after reconnecting
 296
 297 }
 298
 299 /**
 300  * Connect to the PEERSTORE service.
 301  *
 302  * @return NULL on error
 303  */
 304 struct GNUNET_PEERSTORE_Handle *
 305 GNUNET_PEERSTORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
 306 {
 307   struct GNUNET_PEERSTORE_Handle *h;
 308
 309   h = GNUNET_new (struct GNUNET_PEERSTORE_Handle);
 310   h->client = GNUNET_CLIENT_connect ("peerstore", cfg);
 311   if(NULL == h->client)
 312   {
 313     GNUNET_free(h);
 314     return NULL;
 315   }
 316   h->cfg = cfg;
 317   h->mq = GNUNET_MQ_queue_for_connection_client(h->client,
 318       mq_handlers,
 319       &handle_client_error,
 320       h);
 321   if(NULL == h->mq)
 322   {
 323     GNUNET_free(h);
 324     return NULL;
 325   }
 326   LOG(GNUNET_ERROR_TYPE_DEBUG, "New connection created\n");
 327   return h;
 328 }
 329
 330 /**
 331  * Disconnect from the PEERSTORE service
 332  * Do not call in case of pending requests
 333  *
 334  * @param h handle to disconnect
 335  */
 336 void
 337 GNUNET_PEERSTORE_disconnect(struct GNUNET_PEERSTORE_Handle *h)
 338 {
 339   if(NULL != h->watches)
 340   {
 341     GNUNET_CONTAINER_multihashmap_destroy(h->watches);
 342     h->watches = NULL;
 343   }
 344   if(NULL != h->mq)
 345   {
 346     GNUNET_MQ_destroy(h->mq);
 347     h->mq = NULL;
 348   }
 349   if (NULL != h->client)
 350   {
 351     GNUNET_CLIENT_disconnect (h->client);
 352     h->client = NULL;
 353   }
 354   GNUNET_free(h);
 355   LOG(GNUNET_ERROR_TYPE_DEBUG, "Disconnected, BYE!\n");
 356 }
 357
 358
 359 /******************************************************************************/
 360 /*******************            STORE FUNCTIONS           *********************/
 361 /******************************************************************************/
 362
 363 /**
 364  * Callback after MQ envelope is sent
 365  *
 366  * @param cls a 'struct GNUNET_PEERSTORE_StoreContext *'
 367  */
 368 void store_request_sent (void *cls)
 369 {
 370   struct GNUNET_PEERSTORE_StoreContext *sc = cls;
 371   GNUNET_PEERSTORE_Continuation cont;
 372   void *cont_cls;
 373
 374   sc->ev = NULL;
 375   cont = sc->cont;
 376   cont_cls = sc->cont_cls;
 377   GNUNET_PEERSTORE_store_cancel(sc);
 378   if(NULL != cont)
 379     cont(cont_cls, GNUNET_OK);
 380 }
 381
 382 /**
 383  * Cancel a store request
 384  *
 385  * @param sc Store request context
 386  */
 387 void
 388 GNUNET_PEERSTORE_store_cancel (struct GNUNET_PEERSTORE_StoreContext *sc)
 389 {
 390   LOG(GNUNET_ERROR_TYPE_DEBUG,
 391           "Canceling store request.\n");
 392   if(NULL != sc->ev)
 393   {
 394     GNUNET_MQ_send_cancel(sc->ev);
 395     sc->ev = NULL;
 396   }
 397   GNUNET_CONTAINER_DLL_remove(sc->h->store_head, sc->h->store_tail, sc);
 398   GNUNET_free(sc);
 399 }
 400
 401 /**
 402  * Store a new entry in the PEERSTORE
 403  *
 404  * @param h Handle to the PEERSTORE service
 405  * @param sub_system name of the sub system
 406  * @param peer Peer Identity
 407  * @param key entry key
 408  * @param value entry value BLOB
 409  * @param size size of 'value'
 410  * @param expiry absolute time after which the entry is (possibly) deleted
 411  * @param cont Continuation function after the store request is processed
 412  * @param cont_cls Closure for 'cont'
 413  */
 414 struct GNUNET_PEERSTORE_StoreContext *
 415 GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
 416     const char *sub_system,
 417     const struct GNUNET_PeerIdentity *peer,
 418     const char *key,
 419     const void *value,
 420     size_t size,
 421     struct GNUNET_TIME_Absolute expiry,
 422     enum GNUNET_PEERSTORE_StoreOption options,
 423     GNUNET_PEERSTORE_Continuation cont,
 424     void *cont_cls)
 425 {
 426   struct GNUNET_MQ_Envelope *ev;
 427   struct GNUNET_PEERSTORE_StoreContext *sc;
 428
 429   LOG (GNUNET_ERROR_TYPE_DEBUG,
 430       "Storing value (size: %lu) for subsytem `%s', peer `%s', key `%s'\n",
 431       size, sub_system, GNUNET_i2s (peer), key);
 432   ev = PEERSTORE_create_record_mq_envelope(sub_system,
 433       peer,
 434       key,
 435       value,
 436       size,
 437       &expiry,
 438       options,
 439       GNUNET_MESSAGE_TYPE_PEERSTORE_STORE);
 440   sc = GNUNET_new(struct GNUNET_PEERSTORE_StoreContext);
 441   sc->ev = ev;
 442   sc->cont = cont;
 443   sc->cont_cls = cont_cls;
 444   sc->h = h;
 445   GNUNET_CONTAINER_DLL_insert(h->store_head, h->store_tail, sc);
 446   GNUNET_MQ_notify_sent(ev, &store_request_sent, sc);
 447   GNUNET_MQ_send(h->mq, ev);
 448   return sc;
 449
 450 }
 451
 452 /******************************************************************************/
 453 /*******************           ITERATE FUNCTIONS          *********************/
 454 /******************************************************************************/
 455
 456 /**
 457  * When a response for iterate request is received
 458  *
 459  * @param cls a 'struct GNUNET_PEERSTORE_Handle *'
 460  * @param msg message received, NULL on timeout or fatal error
 461  */
 462 void handle_iterate_result (void *cls, const struct GNUNET_MessageHeader *msg)
 463 {
 464   struct GNUNET_PEERSTORE_Handle *h = cls;
 465   struct GNUNET_PEERSTORE_IterateContext *ic;
 466   GNUNET_PEERSTORE_Processor callback;
 467   void *callback_cls;
 468   uint16_t msg_type;
 469   struct GNUNET_PEERSTORE_Record *record;
 470   int continue_iter;
 471
 472   ic = h->iterate_head;
 473   if(NULL == ic)
 474   {
 475     LOG(GNUNET_ERROR_TYPE_ERROR, "Unexpected iteration response, this should not happen.\n");
 476     reconnect(h);
 477     return;
 478   }
 479   callback = ic->callback;
 480   callback_cls = ic->callback_cls;
 481   if(NULL == msg) /* Connection error */
 482   {
 483
 484     if(NULL != callback)
 485       callback(callback_cls, NULL,
 486           _("Error communicating with `PEERSTORE' service."));
 487     reconnect(h);
 488     return;
 489   }
 490   msg_type = ntohs(msg->type);
 491   if(GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_END == msg_type)
 492   {
 493     ic->request_sent = GNUNET_NO;
 494     GNUNET_PEERSTORE_iterate_cancel(ic);
 495     if(NULL != callback)
 496       callback(callback_cls, NULL, NULL);
 497     return;
 498   }
 499   if(NULL != callback)
 500   {
 501     record = PEERSTORE_parse_record_message(msg);
 502     if(NULL == record)
 503       continue_iter = callback(callback_cls, NULL, _("Received a malformed response from service."));
 504     else
 505     {
 506       continue_iter = callback(callback_cls, record, NULL);
 507       PEERSTORE_destroy_record(record);
 508     }
 509     if(GNUNET_NO == continue_iter)
 510       ic->callback = NULL;
 511   }
 512
 513 }
 514
 515 /**
 516  * Callback after MQ envelope is sent
 517  *
 518  * @param cls a 'struct GNUNET_PEERSTORE_IterateContext *'
 519  */
 520 void iterate_request_sent (void *cls)
 521 {
 522   struct GNUNET_PEERSTORE_IterateContext *ic = cls;
 523
 524   LOG(GNUNET_ERROR_TYPE_DEBUG, "Iterate request sent to service.\n");
 525   ic->request_sent = GNUNET_YES;
 526   ic->ev = NULL;
 527 }
 528
 529 /**
 530  * Called when the iterate request is timedout
 531  *
 532  * @param cls a 'struct GNUNET_PEERSTORE_IterateContext *'
 533  */
 534 void iterate_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 535 {
 536   struct GNUNET_PEERSTORE_IterateContext *ic = cls;
 537
 538   ic->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 539   GNUNET_PEERSTORE_iterate_cancel(ic);
 540 }
 541
 542 /**
 543  * Cancel an iterate request
 544  * Please do not call after the iterate request is done
 545  *
 546  * @param ic Iterate request context as returned by GNUNET_PEERSTORE_iterate()
 547  */
 548 void
 549 GNUNET_PEERSTORE_iterate_cancel (struct GNUNET_PEERSTORE_IterateContext *ic)
 550 {
 551   LOG(GNUNET_ERROR_TYPE_DEBUG, "Canceling iterate request.\n");
 552   if(GNUNET_SCHEDULER_NO_TASK != ic->timeout_task)
 553   {
 554     GNUNET_SCHEDULER_cancel(ic->timeout_task);
 555     ic->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 556   }
 557   if(GNUNET_NO == ic->request_sent)
 558   {
 559     if(NULL != ic->ev)
 560     {
 561       GNUNET_MQ_send_cancel(ic->ev);
 562       ic->ev = NULL;
 563     }
 564     GNUNET_CONTAINER_DLL_remove(ic->h->iterate_head, ic->h->iterate_tail, ic);
 565     GNUNET_free(ic);
 566   }
 567   else
 568     ic->callback = NULL;
 569 }
 570
 571 /**
 572  * Iterate over records matching supplied key information
 573  *
 574  * @param h handle to the PEERSTORE service
 575  * @param sub_system name of sub system
 576  * @param peer Peer identity (can be NULL)
 577  * @param key entry key string (can be NULL)
 578  * @param timeout time after which the iterate request is canceled
 579  * @param callback function called with each matching record, all NULL's on end
 580  * @param callback_cls closure for @a callback
 581  * @return Handle to iteration request
 582  */
 583 struct GNUNET_PEERSTORE_IterateContext *
 584 GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h,
 585     const char *sub_system,
 586     const struct GNUNET_PeerIdentity *peer,
 587     const char *key,
 588     struct GNUNET_TIME_Relative timeout,
 589     GNUNET_PEERSTORE_Processor callback, void *callback_cls)
 590 {
 591   struct GNUNET_MQ_Envelope *ev;
 592   struct GNUNET_PEERSTORE_IterateContext *ic;
 593
 594   ev = PEERSTORE_create_record_mq_envelope(sub_system,
 595       peer,
 596       key,
 597       NULL,
 598       0,
 599       NULL,
 600       0,
 601       GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE);
 602   ic = GNUNET_new(struct GNUNET_PEERSTORE_IterateContext);
 603   ic->callback = callback;
 604   ic->callback_cls = callback_cls;
 605   ic->ev = ev;
 606   ic->h = h;
 607   ic->request_sent = GNUNET_NO;
 608   GNUNET_CONTAINER_DLL_insert(h->iterate_head, h->iterate_tail, ic);
 609   LOG(GNUNET_ERROR_TYPE_DEBUG,
 610         "Sending an iterate request for sub system `%s'\n", sub_system);
 611   GNUNET_MQ_notify_sent(ev, &iterate_request_sent, ic);
 612   GNUNET_MQ_send(h->mq, ev);
 613   ic->timeout_task = GNUNET_SCHEDULER_add_delayed(timeout, &iterate_timeout, ic);
 614   return ic;
 615 }
 616
 617 /******************************************************************************/
 618 /*******************            WATCH FUNCTIONS           *********************/
 619 /******************************************************************************/
 620
 621 /**
 622  * When a watch record is received
 623  *
 624  * @param cls a 'struct GNUNET_PEERSTORE_Handle *'
 625  * @param msg message received, NULL on timeout or fatal error
 626  */
 627 void handle_watch_result (void *cls, const struct GNUNET_MessageHeader *msg)
 628 {
 629   struct GNUNET_PEERSTORE_Handle *h = cls;
 630   struct GNUNET_PEERSTORE_Record *record;
 631   struct GNUNET_HashCode keyhash;
 632   struct GNUNET_PEERSTORE_WatchContext *wc;
 633
 634   if(NULL == msg)
 635   {
 636     LOG(GNUNET_ERROR_TYPE_ERROR,
 637         "Problem receiving a watch response, no way to determine which request.\n");
 638     reconnect(h);
 639     return;
 640   }
 641   LOG(GNUNET_ERROR_TYPE_DEBUG, "Received a watch record from service.\n");
 642   record = PEERSTORE_parse_record_message(msg);
 643   PEERSTORE_hash_key(record->sub_system,
 644       record->peer, record->key, &keyhash);
 645   wc = GNUNET_CONTAINER_multihashmap_get(h->watches, &keyhash);
 646   if(NULL != wc->callback)
 647     wc->callback(wc->callback_cls, record, NULL);
 648   PEERSTORE_destroy_record(record);
 649 }
 650
 651 /**
 652  * Callback after MQ envelope is sent
 653  *
 654  * @param cls a 'struct GNUNET_PEERSTORE_WatchContext *'
 655  */
 656 void watch_request_sent (void *cls)
 657 {
 658   struct GNUNET_PEERSTORE_WatchContext *wc = cls;
 659
 660   wc->request_sent = GNUNET_YES;
 661   wc->ev = NULL;
 662 }
 663
 664 /**
 665  * Cancel a watch request
 666  *
 667  * @wc handle to the watch request
 668  */
 669 void
 670 GNUNET_PEERSTORE_watch_cancel(struct GNUNET_PEERSTORE_WatchContext *wc)
 671 {
 672   struct GNUNET_PEERSTORE_Handle *h = wc->h;
 673   struct GNUNET_MQ_Envelope *ev;
 674   struct StoreKeyHashMessage *hm;
 675
 676   LOG(GNUNET_ERROR_TYPE_DEBUG, "Canceling watch.\n");
 677   if(GNUNET_YES == wc->request_sent) /* If request already sent to service, send a cancel request. */
 678   {
 679     ev = GNUNET_MQ_msg(hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_CANCEL);
 680     GNUNET_MQ_send(h->mq, ev);
 681     wc->callback = NULL;
 682     wc->callback_cls = NULL;
 683   }
 684   if(NULL != wc->ev)
 685   {
 686     GNUNET_MQ_send_cancel(wc->ev);
 687     wc->ev = NULL;
 688   }
 689   GNUNET_CONTAINER_multihashmap_remove(h->watches, &wc->keyhash, wc);
 690   GNUNET_free(wc);
 691
 692 }
 693
 694 /**
 695  * Request watching a given key
 696  * User will be notified with any new values added to key
 697  *
 698  * @param h handle to the PEERSTORE service
 699  * @param sub_system name of sub system
 700  * @param peer Peer identity
 701  * @param key entry key string
 702  * @param callback function called with each new value
 703  * @param callback_cls closure for @a callback
 704  * @return Handle to watch request
 705  */
 706 struct GNUNET_PEERSTORE_WatchContext *
 707 GNUNET_PEERSTORE_watch (struct GNUNET_PEERSTORE_Handle *h,
 708     const char *sub_system,
 709     const struct GNUNET_PeerIdentity *peer,
 710     const char *key,
 711     GNUNET_PEERSTORE_Processor callback, void *callback_cls)
 712 {
 713   struct GNUNET_MQ_Envelope *ev;
 714   struct StoreKeyHashMessage *hm;
 715   struct GNUNET_PEERSTORE_WatchContext *wc;
 716
 717   ev = GNUNET_MQ_msg(hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH);
 718   PEERSTORE_hash_key(sub_system, peer, key, &hm->keyhash);
 719   wc = GNUNET_new(struct GNUNET_PEERSTORE_WatchContext);
 720   wc->callback = callback;
 721   wc->callback_cls = callback_cls;
 722   wc->ev = ev;
 723   wc->h = h;
 724   wc->request_sent = GNUNET_NO;
 725   wc->keyhash = hm->keyhash;
 726   if(NULL == h->watches)
 727     h->watches = GNUNET_CONTAINER_multihashmap_create(5, GNUNET_NO);
 728   GNUNET_CONTAINER_multihashmap_put(h->watches, &wc->keyhash,
 729       wc, GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 730   LOG(GNUNET_ERROR_TYPE_DEBUG,
 731       "Sending a watch request for sub system `%s'.\n", sub_system);
 732   GNUNET_MQ_notify_sent(ev, &watch_request_sent, wc);
 733   GNUNET_MQ_send(h->mq, ev);
 734   return wc;
 735 }
 736
 737 /* end of peerstore_api.c */