c8750311909e1aa518e10805205a2cf37e0d100d
[oweals/gnunet.git] / src / psyc / psyc_api.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 it
6  * under the terms of the GNU Affero General Public License as published
7  * by the Free Software Foundation, either version 3 of the License,
8  * or (at your 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  * Affero General Public License for more details.
14  *
15  * You should have received a copy of the GNU Affero General Public License
16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
17  */
18
19 /**
20  * @file psyc/psyc_api.c
21  * @brief PSYC service; high-level access to the PSYC protocol
22  *        note that clients of this API are NOT expected to
23  *        understand the PSYC message format, only the semantics!
24  *        Parsing (and serializing) the PSYC stream format is done
25  *        within the implementation of the libgnunetpsyc library,
26  *        and this API deliberately exposes as little as possible
27  *        of the actual data stream format to the application!
28  * @author Gabor X Toth
29  */
30
31 #include <inttypes.h>
32
33 #include "platform.h"
34 #include "gnunet_util_lib.h"
35 #include "gnunet_multicast_service.h"
36 #include "gnunet_psyc_service.h"
37 #include "gnunet_psyc_util_lib.h"
38 #include "psyc.h"
39
40 #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
41
42
43 /**
44  * Handle to access PSYC channel operations for both the master and slaves.
45  */
46 struct GNUNET_PSYC_Channel
47 {
48   /**
49    * Configuration to use.
50    */
51   const struct GNUNET_CONFIGURATION_Handle *cfg;
52
53   /**
54    * Client connection to the service.
55    */
56   struct GNUNET_MQ_Handle *mq;
57
58   /**
59    * Message to send on connect.
60    */
61   struct GNUNET_MQ_Envelope *connect_env;
62
63   /**
64    * Time to wait until we try to reconnect on failure.
65    */
66   struct GNUNET_TIME_Relative reconnect_delay;
67
68   /**
69    * Task for reconnecting when the listener fails.
70    */
71   struct GNUNET_SCHEDULER_Task *reconnect_task;
72
73   /**
74    * Async operations.
75    */
76   struct GNUNET_OP_Handle *op;
77
78   /**
79    * Transmission handle;
80    */
81   struct GNUNET_PSYC_TransmitHandle *tmit;
82
83   /**
84    * Receipt handle;
85    */
86   struct GNUNET_PSYC_ReceiveHandle *recv;
87
88   /**
89    * Function called after disconnected from the service.
90    */
91   GNUNET_ContinuationCallback disconnect_cb;
92
93   /**
94    * Closure for @a disconnect_cb.
95    */
96   void *disconnect_cls;
97
98   /**
99    * Are we polling for incoming messages right now?
100    */
101   uint8_t in_receive;
102
103   /**
104    * Is this a master or slave channel?
105    */
106   uint8_t is_master;
107
108   /**
109    * Is this channel in the process of disconnecting from the service?
110    * #GNUNET_YES or #GNUNET_NO
111    */
112   uint8_t is_disconnecting;
113 };
114
115
116 /**
117  * Handle for the master of a PSYC channel.
118  */
119 struct GNUNET_PSYC_Master
120 {
121   struct GNUNET_PSYC_Channel chn;
122
123   GNUNET_PSYC_MasterStartCallback start_cb;
124
125   /**
126    * Join request callback.
127    */
128   GNUNET_PSYC_JoinRequestCallback join_req_cb;
129
130   /**
131    * Closure for the callbacks.
132    */
133   void *cb_cls;
134 };
135
136
137 /**
138  * Handle for a PSYC channel slave.
139  */
140 struct GNUNET_PSYC_Slave
141 {
142   struct GNUNET_PSYC_Channel chn;
143
144   GNUNET_PSYC_SlaveConnectCallback connect_cb;
145
146   GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb;
147
148   /**
149    * Closure for the callbacks.
150    */
151   void *cb_cls;
152 };
153
154
155 /**
156  * Handle that identifies a join request.
157  *
158  * Used to match calls to #GNUNET_PSYC_JoinRequestCallback to the
159  * corresponding calls to GNUNET_PSYC_join_decision().
160  */
161 struct GNUNET_PSYC_JoinHandle
162 {
163   struct GNUNET_PSYC_Master *mst;
164   struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
165 };
166
167
168 /**
169  * Handle for a pending PSYC transmission operation.
170  */
171 struct GNUNET_PSYC_SlaveTransmitHandle
172 {
173
174 };
175
176
177 struct GNUNET_PSYC_HistoryRequest
178 {
179   /**
180    * Channel.
181    */
182   struct GNUNET_PSYC_Channel *chn;
183
184   /**
185    * Operation ID.
186    */
187   uint64_t op_id;
188
189   /**
190    * Message handler.
191    */
192   struct GNUNET_PSYC_ReceiveHandle *recv;
193
194   /**
195    * Function to call when the operation finished.
196    */
197   GNUNET_ResultCallback result_cb;
198
199   /**
200    * Closure for @a result_cb.
201    */
202   void *cls;
203 };
204
205
206 struct GNUNET_PSYC_StateRequest
207 {
208   /**
209    * Channel.
210    */
211   struct GNUNET_PSYC_Channel *chn;
212
213   /**
214    * Operation ID.
215    */
216   uint64_t op_id;
217
218   /**
219    * State variable result callback.
220    */
221   GNUNET_PSYC_StateVarCallback var_cb;
222
223   /**
224    * Function to call when the operation finished.
225    */
226   GNUNET_ResultCallback result_cb;
227
228   /**
229    * Closure for @a result_cb.
230    */
231   void *cls;
232 };
233
234
235 static int
236 check_channel_result (void *cls,
237                       const struct GNUNET_OperationResultMessage *res)
238 {
239   return GNUNET_OK;
240 }
241
242
243 static void
244 handle_channel_result (void *cls,
245                        const struct GNUNET_OperationResultMessage *res)
246 {
247   struct GNUNET_PSYC_Channel *chn = cls;
248
249   uint16_t size = ntohs (res->header.size);
250   if (size < sizeof (*res))
251   { /* Error, message too small. */
252     GNUNET_break (0);
253     return;
254   }
255
256   uint16_t data_size = size - sizeof (*res);
257   const char *data = (0 < data_size) ? (void *) &res[1] : NULL;
258   GNUNET_OP_result (chn->op, GNUNET_ntohll (res->op_id),
259                     GNUNET_ntohll (res->result_code),
260                     data, data_size, NULL);
261
262   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
263               "handle_channel_result: Received result message with OP ID %" PRIu64 "\n",
264               GNUNET_ntohll (res->op_id));
265 }
266
267
268 static void
269 op_recv_history_result (void *cls, int64_t result,
270                         const void *data, uint16_t data_size)
271 {
272   LOG (GNUNET_ERROR_TYPE_DEBUG,
273        "Received history replay result: %" PRId64 ".\n", result);
274
275   struct GNUNET_PSYC_HistoryRequest *hist = cls;
276
277   if (NULL != hist->result_cb)
278     hist->result_cb (hist->cls, result, data, data_size);
279
280   GNUNET_PSYC_receive_destroy (hist->recv);
281   GNUNET_free (hist);
282 }
283
284
285 static void
286 op_recv_state_result (void *cls, int64_t result,
287                       const void *data, uint16_t data_size)
288 {
289   LOG (GNUNET_ERROR_TYPE_DEBUG,
290        "Received state request result: %" PRId64 ".\n", result);
291
292   struct GNUNET_PSYC_StateRequest *sr = cls;
293
294   if (NULL != sr->result_cb)
295     sr->result_cb (sr->cls, result, data, data_size);
296
297   GNUNET_free (sr);
298 }
299
300
301 static int
302 check_channel_history_result (void *cls,
303                               const struct GNUNET_OperationResultMessage *res)
304 {
305   struct GNUNET_PSYC_MessageHeader *
306     pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
307   uint16_t size = ntohs (res->header.size);
308
309   if ( (NULL == pmsg) ||
310        (size < sizeof (*res) + sizeof (*pmsg)) )
311   { /* Error, message too small. */
312     GNUNET_break_op (0);
313     return GNUNET_SYSERR;
314   }
315   return GNUNET_OK;
316 }
317
318
319 static void
320 handle_channel_history_result (void *cls,
321                                const struct GNUNET_OperationResultMessage *res)
322 {
323   struct GNUNET_PSYC_Channel *chn = cls;
324   struct GNUNET_PSYC_MessageHeader *
325     pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
326   GNUNET_ResultCallback result_cb = NULL;
327   struct GNUNET_PSYC_HistoryRequest *hist = NULL;
328
329   LOG (GNUNET_ERROR_TYPE_DEBUG,
330        "%p Received historic fragment for message #%" PRIu64 ".\n",
331        chn,
332        GNUNET_ntohll (pmsg->message_id));
333
334   if (GNUNET_YES != GNUNET_OP_get (chn->op,
335                                    GNUNET_ntohll (res->op_id),
336                                    &result_cb, (void *) &hist, NULL))
337   { /* Operation not found. */
338     LOG (GNUNET_ERROR_TYPE_WARNING,
339          "%p Replay operation not found for historic fragment of message #%"
340          PRIu64 ".\n",
341          chn, GNUNET_ntohll (pmsg->message_id));
342     return;
343   }
344
345   GNUNET_PSYC_receive_message (hist->recv,
346                                (const struct GNUNET_PSYC_MessageHeader *) pmsg);
347 }
348
349
350 static int
351 check_channel_state_result (void *cls,
352                             const struct GNUNET_OperationResultMessage *res)
353 {
354   const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
355   uint16_t mod_size;
356   uint16_t size;
357
358   if (NULL == mod)
359   {
360     GNUNET_break_op (0);
361     return GNUNET_SYSERR;
362   }
363   mod_size = ntohs (mod->size);
364   size = ntohs (res->header.size);
365   if (size - sizeof (*res) != mod_size)
366   {
367     GNUNET_break_op (0);
368     return GNUNET_SYSERR;
369   }
370   return GNUNET_OK;
371 }
372
373
374 static void
375 handle_channel_state_result (void *cls,
376                              const struct GNUNET_OperationResultMessage *res)
377 {
378   struct GNUNET_PSYC_Channel *chn = cls;
379
380   GNUNET_ResultCallback result_cb = NULL;
381   struct GNUNET_PSYC_StateRequest *sr = NULL;
382
383   if (GNUNET_YES != GNUNET_OP_get (chn->op,
384                                    GNUNET_ntohll (res->op_id),
385                                    &result_cb, (void *) &sr, NULL))
386   { /* Operation not found. */
387     return;
388   }
389
390   const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
391   if (NULL == mod)
392   {
393     GNUNET_break_op (0);
394     return;
395   }
396   uint16_t mod_size = ntohs (mod->size);
397
398   switch (ntohs (mod->type))
399   {
400   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
401   {
402     const struct GNUNET_PSYC_MessageModifier *
403       pmod = (const struct GNUNET_PSYC_MessageModifier *) mod;
404
405     const char *name = (const char *) &pmod[1];
406     uint16_t name_size = ntohs (pmod->name_size);
407     if (0 == name_size
408         || mod_size - sizeof (*pmod) < name_size
409         || '\0' != name[name_size - 1])
410     {
411       GNUNET_break_op (0);
412       return;
413     }
414     sr->var_cb (sr->cls, mod, name, name + name_size,
415                 ntohs (pmod->header.size) - sizeof (*pmod),
416                 ntohs (pmod->value_size));
417     break;
418   }
419
420   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
421     sr->var_cb (sr->cls, mod, NULL, (const char *) &mod[1],
422                 mod_size - sizeof (*mod), 0);
423     break;
424   }
425 }
426
427
428 static int
429 check_channel_message (void *cls,
430                        const struct GNUNET_PSYC_MessageHeader *pmsg)
431 {
432   return GNUNET_OK;
433 }
434
435
436 static void
437 handle_channel_message (void *cls,
438                         const struct GNUNET_PSYC_MessageHeader *pmsg)
439 {
440   struct GNUNET_PSYC_Channel *chn = cls;
441
442   GNUNET_PSYC_receive_message (chn->recv, pmsg);
443 }
444
445
446 static void
447 handle_channel_message_ack (void *cls,
448                             const struct GNUNET_MessageHeader *msg)
449 {
450   struct GNUNET_PSYC_Channel *chn = cls;
451
452   GNUNET_PSYC_transmit_got_ack (chn->tmit);
453 }
454
455
456 static void
457 handle_master_start_ack (void *cls,
458                          const struct GNUNET_PSYC_CountersResultMessage *cres)
459 {
460   struct GNUNET_PSYC_Master *mst = cls;
461
462   int32_t result = ntohl (cres->result_code);
463   if (GNUNET_OK != result && GNUNET_NO != result)
464   {
465     LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master: %ld\n", result);
466     GNUNET_break (0);
467     /* FIXME: disconnect */
468   }
469   if (NULL != mst->start_cb)
470     mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
471 }
472
473
474 static int
475 check_master_join_request (void *cls,
476                            const struct GNUNET_PSYC_JoinRequestMessage *req)
477 {
478   if ( ((sizeof (*req) + sizeof (struct GNUNET_PSYC_Message)) <= ntohs (req->header.size)) &&
479        (NULL == GNUNET_MQ_extract_nested_mh (req)) )
480   {
481     GNUNET_break_op (0);
482     return GNUNET_SYSERR;
483   }
484   return GNUNET_OK;
485 }
486
487
488 static void
489 handle_master_join_request (void *cls,
490                             const struct GNUNET_PSYC_JoinRequestMessage *req)
491 {
492   struct GNUNET_PSYC_Master *mst = cls;
493
494   if (NULL == mst->join_req_cb)
495     return;
496
497   const struct GNUNET_PSYC_Message *join_msg = NULL;
498   if (sizeof (*req) + sizeof (*join_msg) <= ntohs (req->header.size))
499   {
500     join_msg = (struct GNUNET_PSYC_Message *) GNUNET_MQ_extract_nested_mh (req);
501     LOG (GNUNET_ERROR_TYPE_DEBUG,
502          "Received join_msg of type %u and size %u.\n",
503          ntohs (join_msg->header.type),
504          ntohs (join_msg->header.size));
505   }
506
507   struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh));
508   jh->mst = mst;
509   jh->slave_pub_key = req->slave_pub_key;
510
511   if (NULL != mst->join_req_cb)
512     mst->join_req_cb (mst->cb_cls, req, &req->slave_pub_key, join_msg, jh);
513 }
514
515
516 static void
517 handle_slave_join_ack (void *cls,
518                        const struct GNUNET_PSYC_CountersResultMessage *cres)
519 {
520   struct GNUNET_PSYC_Slave *slv = cls;
521
522   int32_t result = ntohl (cres->result_code);
523   if (GNUNET_YES != result && GNUNET_NO != result)
524   {
525     LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n");
526     GNUNET_break (0);
527     /* FIXME: disconnect */
528   }
529   if (NULL != slv->connect_cb)
530     slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
531 }
532
533
534 static int
535 check_slave_join_decision (void *cls,
536                            const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
537 {
538   return GNUNET_OK;
539 }
540
541
542 static void
543 handle_slave_join_decision (void *cls,
544                             const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
545 {
546   struct GNUNET_PSYC_Slave *slv = cls;
547
548   struct GNUNET_PSYC_Message *pmsg = NULL;
549   if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg))
550     pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1];
551
552   if (NULL != slv->join_dcsn_cb)
553     slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg);
554 }
555
556
557 static void
558 channel_cleanup (struct GNUNET_PSYC_Channel *chn)
559 {
560   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
561               "cleaning up channel %p\n",
562               chn);
563   if (NULL != chn->tmit)
564   {
565     GNUNET_PSYC_transmit_destroy (chn->tmit);
566     chn->tmit = NULL;
567   }
568   if (NULL != chn->recv)
569   {
570
571     GNUNET_PSYC_receive_destroy (chn->recv);
572     chn->recv = NULL;
573   }
574   if (NULL != chn->connect_env)
575   {
576     GNUNET_MQ_discard (chn->connect_env);
577     chn->connect_env = NULL;
578   }
579   if (NULL != chn->mq)
580   {
581     GNUNET_MQ_destroy (chn->mq);
582     chn->mq = NULL;
583   }
584   if (NULL != chn->disconnect_cb)
585   {
586     chn->disconnect_cb (chn->disconnect_cls);
587     chn->disconnect_cb = NULL;
588   }
589   GNUNET_free (chn);
590 }
591
592
593 static void
594 handle_channel_part_ack (void *cls,
595                          const struct GNUNET_MessageHeader *msg)
596 {
597   struct GNUNET_PSYC_Channel *chn = cls;
598
599   channel_cleanup (chn); 
600 }
601
602
603 /*** MASTER ***/
604
605
606 static void
607 master_connect (struct GNUNET_PSYC_Master *mst);
608
609
610 static void
611 master_reconnect (void *cls)
612 {
613   master_connect (cls);
614 }
615
616
617 /**
618  * Master client disconnected from service.
619  *
620  * Reconnect after backoff period.
621  */
622 static void
623 master_disconnected (void *cls, enum GNUNET_MQ_Error error)
624 {
625   struct GNUNET_PSYC_Master *mst = cls;
626   struct GNUNET_PSYC_Channel *chn = &mst->chn;
627
628   LOG (GNUNET_ERROR_TYPE_DEBUG,
629        "Master client disconnected (%d), re-connecting\n",
630        (int) error);
631   if (NULL != chn->tmit)
632   {
633     GNUNET_PSYC_transmit_destroy (chn->tmit);
634     chn->tmit = NULL;
635   }
636   if (NULL != chn->mq)
637   {
638     GNUNET_MQ_destroy (chn->mq);
639     chn->mq = NULL;
640   }
641   chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
642                                                       master_reconnect,
643                                                       mst);
644   chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
645 }
646
647
648 static void
649 master_connect (struct GNUNET_PSYC_Master *mst)
650 {
651   struct GNUNET_PSYC_Channel *chn = &mst->chn;
652
653   struct GNUNET_MQ_MessageHandler handlers[] = {
654     GNUNET_MQ_hd_fixed_size (master_start_ack,
655                              GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK,
656                              struct GNUNET_PSYC_CountersResultMessage,
657                              mst),
658     GNUNET_MQ_hd_var_size (master_join_request,
659                            GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
660                            struct GNUNET_PSYC_JoinRequestMessage,
661                            mst),
662     GNUNET_MQ_hd_fixed_size (channel_part_ack,
663                              GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
664                              struct GNUNET_MessageHeader,
665                              chn),
666     GNUNET_MQ_hd_var_size (channel_message,
667                            GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
668                            struct GNUNET_PSYC_MessageHeader,
669                            chn),
670     GNUNET_MQ_hd_fixed_size (channel_message_ack,
671                              GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
672                              struct GNUNET_MessageHeader,
673                              chn),
674     GNUNET_MQ_hd_var_size (channel_history_result,
675                            GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
676                            struct GNUNET_OperationResultMessage,
677                            chn),
678     GNUNET_MQ_hd_var_size (channel_state_result,
679                            GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
680                            struct GNUNET_OperationResultMessage,
681                            chn),
682     GNUNET_MQ_hd_var_size (channel_result,
683                            GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
684                            struct GNUNET_OperationResultMessage,
685                            chn),
686     GNUNET_MQ_handler_end ()
687   };
688
689   chn->mq = GNUNET_CLIENT_connect (chn->cfg,
690                                    "psyc",
691                                    handlers,
692                                    &master_disconnected,
693                                    mst);
694   GNUNET_assert (NULL != chn->mq);
695   chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
696
697   GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
698 }
699
700
701 /**
702  * Start a PSYC master channel.
703  *
704  * Will start a multicast group identified by the given ECC key.  Messages
705  * received from group members will be given to the respective handler methods.
706  * If a new member wants to join a group, the "join" method handler will be
707  * invoked; the join handler must then generate a "join" message to approve the
708  * joining of the new member.  The channel can also change group membership
709  * without explicit requests.  Note that PSYC doesn't itself "understand" join
710  * or part messages, the respective methods must call other PSYC functions to
711  * inform PSYC about the meaning of the respective events.
712  *
713  * @param cfg  Configuration to use (to connect to PSYC service).
714  * @param channel_key  ECC key that will be used to sign messages for this
715  *        PSYC session. The public key is used to identify the PSYC channel.
716  *        Note that end-users will usually not use the private key directly, but
717  *        rather look it up in GNS for places managed by other users, or select
718  *        a file with the private key(s) when setting up their own channels
719  *        FIXME: we'll likely want to use NOT the p521 curve here, but a cheaper
720  *        one in the future.
721  * @param policy  Channel policy specifying join and history restrictions.
722  *        Used to automate join decisions.
723  * @param message_cb  Function to invoke on message parts received from slaves.
724  * @param join_request_cb  Function to invoke when a slave wants to join.
725  * @param master_start_cb  Function to invoke after the channel master started.
726  * @param cls  Closure for @a method and @a join_cb.
727  *
728  * @return Handle for the channel master, NULL on error.
729  */
730 struct GNUNET_PSYC_Master *
731 GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
732                           const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key,
733                           enum GNUNET_PSYC_Policy policy,
734                           GNUNET_PSYC_MasterStartCallback start_cb,
735                           GNUNET_PSYC_JoinRequestCallback join_request_cb,
736                           GNUNET_PSYC_MessageCallback message_cb,
737                           GNUNET_PSYC_MessagePartCallback message_part_cb,
738                           void *cls)
739 {
740   struct GNUNET_PSYC_Master *mst = GNUNET_new (struct GNUNET_PSYC_Master);
741   struct GNUNET_PSYC_Channel *chn = &mst->chn;
742   struct MasterStartRequest *req;
743
744   chn->connect_env = GNUNET_MQ_msg (req,
745                                     GNUNET_MESSAGE_TYPE_PSYC_MASTER_START);
746   req->channel_key = *channel_key;
747   req->policy = policy;
748
749   chn->cfg = cfg;
750   chn->is_master = GNUNET_YES;
751   chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
752
753   chn->op = GNUNET_OP_create ();
754   chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
755
756   mst->start_cb = start_cb;
757   mst->join_req_cb = join_request_cb;
758   mst->cb_cls = cls;
759
760   master_connect (mst);
761   return mst;
762 }
763
764
765 /**
766  * Stop a PSYC master channel.
767  *
768  * @param master PSYC channel master to stop.
769  * @param keep_active  FIXME
770  */
771 void
772 GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst,
773                          int keep_active,
774                          GNUNET_ContinuationCallback stop_cb,
775                          void *stop_cls)
776 {
777   struct GNUNET_PSYC_Channel *chn = &mst->chn;
778   struct GNUNET_MQ_Envelope *env;
779
780   chn->is_disconnecting = GNUNET_YES;
781   chn->disconnect_cb = stop_cb;
782   chn->disconnect_cls = stop_cls;
783   env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
784   GNUNET_MQ_send (chn->mq, env);
785 }
786
787
788 /**
789  * Function to call with the decision made for a join request.
790  *
791  * Must be called once and only once in response to an invocation of the
792  * #GNUNET_PSYC_JoinCallback.
793  *
794  * @param jh Join request handle.
795  * @param is_admitted  #GNUNET_YES    if the join is approved,
796  *                     #GNUNET_NO     if it is disapproved,
797  *                     #GNUNET_SYSERR if we cannot answer the request.
798  * @param relay_count Number of relays given.
799  * @param relays Array of suggested peers that might be useful relays to use
800  *        when joining the multicast group (essentially a list of peers that
801  *        are already part of the multicast group and might thus be willing
802  *        to help with routing).  If empty, only this local peer (which must
803  *        be the multicast origin) is a good candidate for building the
804  *        multicast tree.  Note that it is unnecessary to specify our own
805  *        peer identity in this array.
806  * @param join_resp  Application-dependent join response message.
807  *
808  * @return #GNUNET_OK on success,
809  *         #GNUNET_SYSERR if the message is too large.
810  */
811 int
812 GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
813                            int is_admitted,
814                            uint32_t relay_count,
815                            const struct GNUNET_PeerIdentity *relays,
816                            const struct GNUNET_PSYC_Message *join_resp)
817 {
818   struct GNUNET_PSYC_Channel *chn = &jh->mst->chn;
819   struct GNUNET_PSYC_JoinDecisionMessage *dcsn;
820   uint16_t join_resp_size
821     = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0;
822   uint16_t relay_size = relay_count * sizeof (*relays);
823
824   if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
825       < sizeof (*dcsn) + relay_size + join_resp_size)
826     return GNUNET_SYSERR;
827
828   struct GNUNET_MQ_Envelope *
829     env = GNUNET_MQ_msg_extra (dcsn, relay_size + join_resp_size,
830                                GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
831   dcsn->is_admitted = htonl (is_admitted);
832   dcsn->slave_pub_key = jh->slave_pub_key;
833
834   if (0 < join_resp_size)
835     GNUNET_memcpy (&dcsn[1], join_resp, join_resp_size);
836
837   GNUNET_MQ_send (chn->mq, env);
838   GNUNET_free (jh);
839   return GNUNET_OK;
840 }
841
842
843 /**
844  * Send a message to call a method to all members in the PSYC channel.
845  *
846  * @param master Handle to the PSYC channel.
847  * @param method_name Which method should be invoked.
848  * @param notify_mod Function to call to obtain modifiers.
849  * @param notify_data Function to call to obtain fragments of the data.
850  * @param notify_cls Closure for @a notify_mod and @a notify_data.
851  * @param flags Flags for the message being transmitted.
852  *
853  * @return Transmission handle, NULL on error (i.e. more than one request queued).
854  */
855 struct GNUNET_PSYC_MasterTransmitHandle *
856 GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *mst,
857                              const char *method_name,
858                              GNUNET_PSYC_TransmitNotifyModifier notify_mod,
859                              GNUNET_PSYC_TransmitNotifyData notify_data,
860                              void *notify_cls,
861                              enum GNUNET_PSYC_MasterTransmitFlags flags)
862 {
863   if (GNUNET_OK
864       == GNUNET_PSYC_transmit_message (mst->chn.tmit, method_name, NULL,
865                                        notify_mod, notify_data, notify_cls,
866                                        flags))
867     return (struct GNUNET_PSYC_MasterTransmitHandle *) mst->chn.tmit;
868   else
869     return NULL;
870 }
871
872
873 /**
874  * Resume transmission to the channel.
875  *
876  * @param tmit  Handle of the request that is being resumed.
877  */
878 void
879 GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
880 {
881   GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
882 }
883
884
885 /**
886  * Abort transmission request to the channel.
887  *
888  * @param tmit  Handle of the request that is being aborted.
889  */
890 void
891 GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
892 {
893   GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
894 }
895
896
897 /**
898  * Convert a channel @a master to a @e channel handle to access the @e channel
899  * APIs.
900  *
901  * @param master Channel master handle.
902  *
903  * @return Channel handle, valid for as long as @a master is valid.
904  */
905 struct GNUNET_PSYC_Channel *
906 GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
907 {
908   return &master->chn;
909 }
910
911
912 /*** SLAVE ***/
913
914
915 static void
916 slave_connect (struct GNUNET_PSYC_Slave *slv);
917
918
919 static void
920 slave_reconnect (void *cls)
921 {
922   slave_connect (cls);
923 }
924
925
926 /**
927  * Slave client disconnected from service.
928  *
929  * Reconnect after backoff period.
930  */
931 static void
932 slave_disconnected (void *cls,
933                     enum GNUNET_MQ_Error error)
934 {
935   struct GNUNET_PSYC_Slave *slv = cls;
936   struct GNUNET_PSYC_Channel *chn = &slv->chn;
937
938   LOG (GNUNET_ERROR_TYPE_DEBUG,
939        "Slave client disconnected (%d), re-connecting\n",
940        (int) error);
941   if (NULL != chn->tmit)
942   {
943     GNUNET_PSYC_transmit_destroy (chn->tmit);
944     chn->tmit = NULL;
945   }
946   if (NULL != chn->mq)
947   {
948     GNUNET_MQ_destroy (chn->mq);
949     chn->mq = NULL;
950   }
951   chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
952                                                       &slave_reconnect,
953                                                       slv);
954   chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
955 }
956
957
958 static void
959 slave_connect (struct GNUNET_PSYC_Slave *slv)
960 {
961   struct GNUNET_PSYC_Channel *chn = &slv->chn;
962
963   struct GNUNET_MQ_MessageHandler handlers[] = {
964     GNUNET_MQ_hd_fixed_size (slave_join_ack,
965                              GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK,
966                              struct GNUNET_PSYC_CountersResultMessage,
967                              slv),
968     GNUNET_MQ_hd_var_size (slave_join_decision,
969                            GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
970                            struct GNUNET_PSYC_JoinDecisionMessage,
971                            slv),
972     GNUNET_MQ_hd_fixed_size (channel_part_ack,
973                              GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
974                              struct GNUNET_MessageHeader,
975                              chn),
976     GNUNET_MQ_hd_var_size (channel_message,
977                            GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
978                            struct GNUNET_PSYC_MessageHeader,
979                            chn),
980     GNUNET_MQ_hd_fixed_size (channel_message_ack,
981                              GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
982                              struct GNUNET_MessageHeader,
983                              chn),
984     GNUNET_MQ_hd_var_size (channel_history_result,
985                            GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
986                            struct GNUNET_OperationResultMessage,
987                            chn),
988     GNUNET_MQ_hd_var_size (channel_state_result,
989                            GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
990                            struct GNUNET_OperationResultMessage,
991                            chn),
992     GNUNET_MQ_hd_var_size (channel_result,
993                            GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
994                            struct GNUNET_OperationResultMessage,
995                            chn),
996     GNUNET_MQ_handler_end ()
997   };
998
999   chn->mq = GNUNET_CLIENT_connect (chn->cfg,
1000                                    "psyc",
1001                                    handlers,
1002                                    &slave_disconnected,
1003                                    slv);
1004   if (NULL == chn->mq)
1005   {
1006     chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
1007                                                         &slave_reconnect,
1008                                                         slv);
1009     chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
1010     return;
1011   }
1012   chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
1013
1014   GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
1015 }
1016
1017
1018 /**
1019  * Join a PSYC channel.
1020  *
1021  * The entity joining is always the local peer.  The user must immediately use
1022  * the GNUNET_PSYC_slave_transmit() functions to transmit a @e join_msg to the
1023  * channel; if the join request succeeds, the channel state (and @e recent
1024  * method calls) will be replayed to the joining member.  There is no explicit
1025  * notification on failure (as the channel may simply take days to approve,
1026  * and disapproval is simply being ignored).
1027  *
1028  * @param cfg
1029  *        Configuration to use.
1030  * @param channel_key  ECC public key that identifies the channel we wish to join.
1031  * @param slave_key  ECC private-public key pair that identifies the slave, and
1032  *        used by multicast to sign the join request and subsequent unicast
1033  *        requests sent to the master.
1034  * @param origin  Peer identity of the origin.
1035  * @param relay_count  Number of peers in the @a relays array.
1036  * @param relays  Peer identities of members of the multicast group, which serve
1037  *        as relays and used to join the group at.
1038  * @param message_cb  Function to invoke on message parts received from the
1039  *        channel, typically at least contains method handlers for @e join and
1040  *        @e part.
1041  * @param slave_connect_cb  Function invoked once we have connected to the
1042  *        PSYC service.
1043  * @param join_decision_cb  Function invoked once we have received a join
1044  *        decision.
1045  * @param cls  Closure for @a message_cb and @a slave_joined_cb.
1046  * @param method_name  Method name for the join request.
1047  * @param env  Environment containing transient variables for the request, or NULL.
1048  * @param data  Payload for the join message.
1049  * @param data_size  Number of bytes in @a data.
1050  *
1051  * @return Handle for the slave, NULL on error.
1052  */
1053 struct GNUNET_PSYC_Slave *
1054 GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
1055                         const struct GNUNET_CRYPTO_EddsaPublicKey *channel_pub_key,
1056                         const struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key,
1057                         enum GNUNET_PSYC_SlaveJoinFlags flags,
1058                         const struct GNUNET_PeerIdentity *origin,
1059                         uint32_t relay_count,
1060                         const struct GNUNET_PeerIdentity *relays,
1061                         GNUNET_PSYC_MessageCallback message_cb,
1062                         GNUNET_PSYC_MessagePartCallback message_part_cb,
1063                         GNUNET_PSYC_SlaveConnectCallback connect_cb,
1064                         GNUNET_PSYC_JoinDecisionCallback join_decision_cb,
1065                         void *cls,
1066                         const struct GNUNET_PSYC_Message *join_msg)
1067 {
1068   struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv));
1069   struct GNUNET_PSYC_Channel *chn = &slv->chn;
1070   uint16_t relay_size = relay_count * sizeof (*relays);
1071   uint16_t join_msg_size;
1072   if (NULL == join_msg)
1073     join_msg_size = 0;
1074   else
1075     join_msg_size = ntohs (join_msg->header.size);
1076
1077   struct SlaveJoinRequest *req;
1078   chn->connect_env = GNUNET_MQ_msg_extra (req, relay_size + join_msg_size,
1079                                           GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
1080   req->channel_pub_key = *channel_pub_key;
1081   req->slave_key = *slave_key;
1082   req->origin = *origin;
1083   req->relay_count = htonl (relay_count);
1084   req->flags = htonl (flags);
1085
1086   if (0 < relay_size)
1087     GNUNET_memcpy (&req[1], relays, relay_size);
1088
1089   if (NULL != join_msg)
1090     GNUNET_memcpy ((char *) &req[1] + relay_size, join_msg, join_msg_size);
1091
1092   chn->cfg = cfg;
1093   chn->is_master = GNUNET_NO;
1094   chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
1095
1096   chn->op = GNUNET_OP_create ();
1097   chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
1098
1099   slv->connect_cb = connect_cb;
1100   slv->join_dcsn_cb = join_decision_cb;
1101   slv->cb_cls = cls;
1102
1103   slave_connect (slv);
1104   return slv;
1105 }
1106
1107
1108 /**
1109  * Part a PSYC channel.
1110  *
1111  * Will terminate the connection to the PSYC service.  Polite clients should
1112  * first explicitly send a part request (via GNUNET_PSYC_slave_transmit()).
1113  *
1114  * @param slave Slave handle.
1115  */
1116 void
1117 GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv,
1118                         int keep_active,
1119                         GNUNET_ContinuationCallback part_cb,
1120                         void *part_cls)
1121 {
1122   struct GNUNET_PSYC_Channel *chn = &slv->chn;
1123   struct GNUNET_MQ_Envelope *env;
1124
1125   chn->is_disconnecting = GNUNET_YES;
1126   chn->disconnect_cb = part_cb;
1127   chn->disconnect_cls = part_cls;
1128   env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
1129   GNUNET_MQ_send (chn->mq, env);
1130 }
1131
1132
1133 /**
1134  * Request a message to be sent to the channel master.
1135  *
1136  * @param slave Slave handle.
1137  * @param method_name Which (PSYC) method should be invoked (on host).
1138  * @param notify_mod Function to call to obtain modifiers.
1139  * @param notify_data Function to call to obtain fragments of the data.
1140  * @param notify_cls Closure for @a notify.
1141  * @param flags Flags for the message being transmitted.
1142  *
1143  * @return Transmission handle, NULL on error (i.e. more than one request
1144  *         queued).
1145  */
1146 struct GNUNET_PSYC_SlaveTransmitHandle *
1147 GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slv,
1148                             const char *method_name,
1149                             GNUNET_PSYC_TransmitNotifyModifier notify_mod,
1150                             GNUNET_PSYC_TransmitNotifyData notify_data,
1151                             void *notify_cls,
1152                             enum GNUNET_PSYC_SlaveTransmitFlags flags)
1153
1154 {
1155   if (GNUNET_OK
1156       == GNUNET_PSYC_transmit_message (slv->chn.tmit, method_name, NULL,
1157                                        notify_mod, notify_data, notify_cls,
1158                                        flags))
1159     return (struct GNUNET_PSYC_SlaveTransmitHandle *) slv->chn.tmit;
1160   else
1161     return NULL;
1162 }
1163
1164
1165 /**
1166  * Resume transmission to the master.
1167  *
1168  * @param tmit Handle of the request that is being resumed.
1169  */
1170 void
1171 GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
1172 {
1173   GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
1174 }
1175
1176
1177 /**
1178  * Abort transmission request to master.
1179  *
1180  * @param tmit Handle of the request that is being aborted.
1181  */
1182 void
1183 GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
1184 {
1185   GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
1186 }
1187
1188
1189 /**
1190  * Convert @a slave to a @e channel handle to access the @e channel APIs.
1191  *
1192  * @param slv Slave handle.
1193  *
1194  * @return Channel handle, valid for as long as @a slave is valid.
1195  */
1196 struct GNUNET_PSYC_Channel *
1197 GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slv)
1198 {
1199   return &slv->chn;
1200 }
1201
1202
1203 /**
1204  * Add a slave to the channel's membership list.
1205  *
1206  * Note that this will NOT generate any PSYC traffic, it will merely update the
1207  * local database to modify how we react to <em>membership test</em> queries.
1208  * The channel master still needs to explicitly transmit a @e join message to
1209  * notify other channel members and they then also must still call this function
1210  * in their respective methods handling the @e join message.  This way, how @e
1211  * join and @e part operations are exactly implemented is still up to the
1212  * application; for example, there might be a @e part_all method to kick out
1213  * everyone.
1214  *
1215  * Note that channel slaves are explicitly trusted to execute such methods
1216  * correctly; not doing so correctly will result in either denying other slaves
1217  * access or offering access to channel data to non-members.
1218  *
1219  * @param chn
1220  *        Channel handle.
1221  * @param slave_pub_key
1222  *        Identity of channel slave to add.
1223  * @param announced_at
1224  *        ID of the message that announced the membership change.
1225  * @param effective_since
1226  *        Addition of slave is in effect since this message ID.
1227  * @param result_cb
1228  *        Function to call with the result of the operation.
1229  *        The @e result_code argument is #GNUNET_OK on success, or
1230  *        #GNUNET_SYSERR on error.  In case of an error, the @e data argument
1231  *        can contain an optional error message.
1232  * @param cls
1233  *        Closure for @a result_cb.
1234  */
1235 void
1236 GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
1237                                const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
1238                                uint64_t announced_at,
1239                                uint64_t effective_since,
1240                                GNUNET_ResultCallback result_cb,
1241                                void *cls)
1242 {
1243   struct ChannelMembershipStoreRequest *req;
1244   struct GNUNET_MQ_Envelope *
1245     env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
1246   req->slave_pub_key = *slave_pub_key;
1247   req->announced_at = GNUNET_htonll (announced_at);
1248   req->effective_since = GNUNET_htonll (effective_since);
1249   req->did_join = GNUNET_YES;
1250   req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
1251
1252   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1253               "GNUNET_PSYC_channel_slave_add, OP ID: %" PRIu64 "\n",
1254               GNUNET_ntohll (req->op_id));
1255   GNUNET_MQ_send (chn->mq, env);
1256 }
1257
1258
1259 /**
1260  * Remove a slave from the channel's membership list.
1261  *
1262  * Note that this will NOT generate any PSYC traffic, it will merely update the
1263  * local database to modify how we react to <em>membership test</em> queries.
1264  * The channel master still needs to explicitly transmit a @e part message to
1265  * notify other channel members and they then also must still call this function
1266  * in their respective methods handling the @e part message.  This way, how
1267  * @e join and @e part operations are exactly implemented is still up to the
1268  * application; for example, there might be a @e part_all message to kick out
1269  * everyone.
1270  *
1271  * Note that channel members are explicitly trusted to perform these
1272  * operations correctly; not doing so correctly will result in either
1273  * denying members access or offering access to channel data to
1274  * non-members.
1275  *
1276  * @param chn
1277  *        Channel handle.
1278  * @param slave_pub_key
1279  *        Identity of channel slave to remove.
1280  * @param announced_at
1281  *        ID of the message that announced the membership change.
1282  * @param result_cb
1283  *        Function to call with the result of the operation.
1284  *        The @e result_code argument is #GNUNET_OK on success, or
1285  *        #GNUNET_SYSERR on error.  In case of an error, the @e data argument
1286  *        can contain an optional error message.
1287  * @param cls
1288  *        Closure for @a result_cb.
1289  */
1290 void
1291 GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn,
1292                                   const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
1293                                   uint64_t announced_at,
1294                                   GNUNET_ResultCallback result_cb,
1295                                   void *cls)
1296 {
1297   struct ChannelMembershipStoreRequest *req;
1298   struct GNUNET_MQ_Envelope *
1299     env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
1300   req->slave_pub_key = *slave_pub_key;
1301   req->announced_at = GNUNET_htonll (announced_at);
1302   req->did_join = GNUNET_NO;
1303   req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
1304
1305   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1306               "GNUNET_PSYC_channel_slave_remove, OP ID: %" PRIu64 "\n",
1307               GNUNET_ntohll (req->op_id));
1308   GNUNET_MQ_send (chn->mq, env);
1309 }
1310
1311
1312 static struct GNUNET_PSYC_HistoryRequest *
1313 channel_history_replay (struct GNUNET_PSYC_Channel *chn,
1314                         uint64_t start_message_id,
1315                         uint64_t end_message_id,
1316                         uint64_t message_limit,
1317                         const char *method_prefix,
1318                         uint32_t flags,
1319                         GNUNET_PSYC_MessageCallback message_cb,
1320                         GNUNET_PSYC_MessagePartCallback message_part_cb,
1321                         GNUNET_ResultCallback result_cb,
1322                         void *cls)
1323 {
1324   struct GNUNET_PSYC_HistoryRequestMessage *req;
1325   struct GNUNET_PSYC_HistoryRequest *hist = GNUNET_malloc (sizeof (*hist));
1326   hist->chn = chn;
1327   hist->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
1328   hist->result_cb = result_cb;
1329   hist->cls = cls;
1330   hist->op_id = GNUNET_OP_add (chn->op, op_recv_history_result, hist, NULL);
1331
1332   GNUNET_assert (NULL != method_prefix);
1333   uint16_t method_size = strnlen (method_prefix,
1334                                   GNUNET_MAX_MESSAGE_SIZE
1335                                   - sizeof (*req)) + 1;
1336   GNUNET_assert ('\0' == method_prefix[method_size - 1]);
1337
1338   struct GNUNET_MQ_Envelope *
1339     env = GNUNET_MQ_msg_extra (req, method_size,
1340                                GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY);
1341   req->start_message_id = GNUNET_htonll (start_message_id);
1342   req->end_message_id = GNUNET_htonll (end_message_id);
1343   req->message_limit = GNUNET_htonll (message_limit);
1344   req->flags = htonl (flags);
1345   req->op_id = GNUNET_htonll (hist->op_id);
1346
1347   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1348               "channel_history_replay, OP ID: %" PRIu64 "\n",
1349               GNUNET_ntohll (req->op_id));
1350   GNUNET_memcpy (&req[1], method_prefix, method_size);
1351
1352   GNUNET_MQ_send (chn->mq, env);
1353   return hist;
1354 }
1355
1356
1357 /**
1358  * Request to replay a part of the message history of the channel.
1359  *
1360  * Historic messages (but NOT the state at the time) will be replayed and given
1361  * to the normal method handlers with a #GNUNET_PSYC_MESSAGE_HISTORIC flag set.
1362  *
1363  * Messages are retrieved from the local PSYCstore if available,
1364  * otherwise requested from the network.
1365  *
1366  * @param channel
1367  *        Which channel should be replayed?
1368  * @param start_message_id
1369  *        Earliest interesting point in history.
1370  * @param end_message_id
1371  *        Last (inclusive) interesting point in history.
1372  * @param method_prefix
1373  *        Retrieve only messages with a matching method prefix.
1374  * @param flags
1375  *        OR'ed enum GNUNET_PSYC_HistoryReplayFlags
1376  * @param result_cb
1377  *        Function to call when the requested history has been fully replayed.
1378  * @param cls
1379  *        Closure for the callbacks.
1380  *
1381  * @return Handle to cancel history replay operation.
1382  */
1383 struct GNUNET_PSYC_HistoryRequest *
1384 GNUNET_PSYC_channel_history_replay (struct GNUNET_PSYC_Channel *chn,
1385                                     uint64_t start_message_id,
1386                                     uint64_t end_message_id,
1387                                     const char *method_prefix,
1388                                     uint32_t flags,
1389                                     GNUNET_PSYC_MessageCallback message_cb,
1390                                     GNUNET_PSYC_MessagePartCallback message_part_cb,
1391                                     GNUNET_ResultCallback result_cb,
1392                                     void *cls)
1393 {
1394   return channel_history_replay (chn, start_message_id, end_message_id, 0,
1395                                  method_prefix, flags,
1396                                  message_cb, message_part_cb, result_cb, cls);
1397 }
1398
1399
1400 /**
1401  * Request to replay the latest messages from the message history of the channel.
1402  *
1403  * Historic messages (but NOT the state at the time) will be replayed (given to
1404  * the normal method handlers) if available and if access is permitted.
1405  *
1406  * @param channel
1407  *        Which channel should be replayed?
1408  * @param message_limit
1409  *        Maximum number of messages to replay.
1410  * @param method_prefix
1411  *        Retrieve only messages with a matching method prefix.
1412  *        Use NULL or "" to retrieve all.
1413  * @param flags
1414  *        OR'ed enum GNUNET_PSYC_HistoryReplayFlags
1415  * @param result_cb
1416  *        Function to call when the requested history has been fully replayed.
1417  * @param cls
1418  *        Closure for the callbacks.
1419  *
1420  * @return Handle to cancel history replay operation.
1421  */
1422 struct GNUNET_PSYC_HistoryRequest *
1423 GNUNET_PSYC_channel_history_replay_latest (struct GNUNET_PSYC_Channel *chn,
1424                                            uint64_t message_limit,
1425                                            const char *method_prefix,
1426                                            uint32_t flags,
1427                                            GNUNET_PSYC_MessageCallback message_cb,
1428                                            GNUNET_PSYC_MessagePartCallback message_part_cb,
1429                                            GNUNET_ResultCallback result_cb,
1430                                            void *cls)
1431 {
1432   return channel_history_replay (chn, 0, 0, message_limit, method_prefix, flags,
1433                                  message_cb, message_part_cb, result_cb, cls);
1434 }
1435
1436
1437 void
1438 GNUNET_PSYC_channel_history_replay_cancel (struct GNUNET_PSYC_Channel *channel,
1439                                            struct GNUNET_PSYC_HistoryRequest *hist)
1440 {
1441   GNUNET_PSYC_receive_destroy (hist->recv);
1442   GNUNET_OP_remove (hist->chn->op, hist->op_id);
1443   GNUNET_free (hist);
1444 }
1445
1446
1447 /**
1448  * Retrieve the best matching channel state variable.
1449  *
1450  * If the requested variable name is not present in the state, the nearest
1451  * less-specific name is matched; for example, requesting "_a_b" will match "_a"
1452  * if "_a_b" does not exist.
1453  *
1454  * @param channel
1455  *        Channel handle.
1456  * @param full_name
1457  *        Full name of the requested variable.
1458  *        The actual variable returned might have a shorter name.
1459  * @param var_cb
1460  *        Function called once when a matching state variable is found.
1461  *        Not called if there's no matching state variable.
1462  * @param result_cb
1463  *        Function called after the operation finished.
1464  *        (i.e. all state variables have been returned via @a state_cb)
1465  * @param cls
1466  *        Closure for the callbacks.
1467  */
1468 static struct GNUNET_PSYC_StateRequest *
1469 channel_state_get (struct GNUNET_PSYC_Channel *chn,
1470                    uint16_t type, const char *name,
1471                    GNUNET_PSYC_StateVarCallback var_cb,
1472                    GNUNET_ResultCallback result_cb, void *cls)
1473 {
1474   struct StateRequest *req;
1475   struct GNUNET_PSYC_StateRequest *sr = GNUNET_malloc (sizeof (*sr));
1476   sr->chn = chn;
1477   sr->var_cb = var_cb;
1478   sr->result_cb = result_cb;
1479   sr->cls = cls;
1480   sr->op_id = GNUNET_OP_add (chn->op, op_recv_state_result, sr, NULL);
1481
1482   GNUNET_assert (NULL != name);
1483   size_t name_size = strnlen (name, GNUNET_MAX_MESSAGE_SIZE
1484                               - sizeof (*req)) + 1;
1485   struct GNUNET_MQ_Envelope *
1486     env = GNUNET_MQ_msg_extra (req, name_size, type);
1487   req->op_id = GNUNET_htonll (sr->op_id);
1488
1489   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1490               "channel_state_get, OP ID: %" PRIu64 "\n",
1491               GNUNET_ntohll (req->op_id));
1492
1493   GNUNET_memcpy (&req[1], name, name_size);
1494
1495   GNUNET_MQ_send (chn->mq, env);
1496   return sr;
1497 }
1498
1499
1500 /**
1501  * Retrieve the best matching channel state variable.
1502  *
1503  * If the requested variable name is not present in the state, the nearest
1504  * less-specific name is matched; for example, requesting "_a_b" will match "_a"
1505  * if "_a_b" does not exist.
1506  *
1507  * @param channel
1508  *        Channel handle.
1509  * @param full_name
1510  *        Full name of the requested variable.
1511  *        The actual variable returned might have a shorter name.
1512  * @param var_cb
1513  *        Function called once when a matching state variable is found.
1514  *        Not called if there's no matching state variable.
1515  * @param result_cb
1516  *        Function called after the operation finished.
1517  *        (i.e. all state variables have been returned via @a state_cb)
1518  * @param cls
1519  *        Closure for the callbacks.
1520  */
1521 struct GNUNET_PSYC_StateRequest *
1522 GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *chn,
1523                                const char *full_name,
1524                                GNUNET_PSYC_StateVarCallback var_cb,
1525                                GNUNET_ResultCallback result_cb,
1526                                void *cls)
1527 {
1528   return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET,
1529                             full_name, var_cb, result_cb, cls);
1530
1531 }
1532
1533
1534 /**
1535  * Return all channel state variables whose name matches a given prefix.
1536  *
1537  * A name matches if it starts with the given @a name_prefix, thus requesting
1538  * the empty prefix ("") will match all values; requesting "_a_b" will also
1539  * return values stored under "_a_b_c".
1540  *
1541  * The @a state_cb is invoked on all matching state variables asynchronously, as
1542  * the state is stored in and retrieved from the PSYCstore,
1543  *
1544  * @param channel
1545  *        Channel handle.
1546  * @param name_prefix
1547  *        Prefix of the state variable name to match.
1548  * @param var_cb
1549  *        Function called once when a matching state variable is found.
1550  *        Not called if there's no matching state variable.
1551  * @param result_cb
1552  *        Function called after the operation finished.
1553  *        (i.e. all state variables have been returned via @a state_cb)
1554  * @param cls
1555  *        Closure for the callbacks.
1556  */
1557 struct GNUNET_PSYC_StateRequest *
1558 GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *chn,
1559                                       const char *name_prefix,
1560                                       GNUNET_PSYC_StateVarCallback var_cb,
1561                                       GNUNET_ResultCallback result_cb,
1562                                       void *cls)
1563 {
1564   return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX,
1565                             name_prefix, var_cb, result_cb, cls);
1566 }
1567
1568
1569 /**
1570  * Cancel a state request operation.
1571  *
1572  * @param sr
1573  *        Handle for the operation to cancel.
1574  */
1575 void
1576 GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateRequest *sr)
1577 {
1578   GNUNET_OP_remove (sr->chn->op, sr->op_id);
1579   GNUNET_free (sr);
1580 }
1581
1582 /* end of psyc_api.c */