src/multicast/multicast_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2012, 2013 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file multicast/multicast_api.c
  23  * @brief Multicast service; implements multicast groups using CADET connections.
  24  * @author Christian Grothoff
  25  * @author Gabor X Toth
  26  */
  27
  28 #include "platform.h"
  29 #include "gnunet_util_lib.h"
  30 #include "gnunet_multicast_service.h"
  31 #include "multicast.h"
  32
  33 #define LOG(kind,...) GNUNET_log_from (kind, "multicast-api",__VA_ARGS__)
  34
  35
  36 /**
  37  * Handle for a request to send a message to all multicast group members
  38  * (from the origin).
  39  */
  40 struct GNUNET_MULTICAST_OriginTransmitHandle
  41 {
  42   GNUNET_MULTICAST_OriginTransmitNotify notify;
  43   void *notify_cls;
  44   struct GNUNET_MULTICAST_Origin *origin;
  45
  46   uint64_t message_id;
  47   uint64_t group_generation;
  48   uint64_t fragment_offset;
  49 };
  50
  51
  52 /**
  53  * Handle for a message to be delivered from a member to the origin.
  54  */
  55 struct GNUNET_MULTICAST_MemberTransmitHandle
  56 {
  57   GNUNET_MULTICAST_MemberTransmitNotify notify;
  58   void *notify_cls;
  59   struct GNUNET_MULTICAST_Member *member;
  60
  61   uint64_t request_id;
  62   uint64_t fragment_offset;
  63 };
  64
  65
  66 struct GNUNET_MULTICAST_Group
  67 {
  68   /**
  69    * Configuration to use.
  70    */
  71   const struct GNUNET_CONFIGURATION_Handle *cfg;
  72
  73   /**
  74    * Client connection to the service.
  75    */
  76   struct GNUNET_CLIENT_MANAGER_Connection *client;
  77
  78   /**
  79    * Message to send on reconnect.
  80    */
  81   struct GNUNET_MessageHeader *connect_msg;
  82
  83   GNUNET_MULTICAST_JoinRequestCallback join_req_cb;
  84   GNUNET_MULTICAST_ReplayFragmentCallback replay_frag_cb;
  85   GNUNET_MULTICAST_ReplayMessageCallback replay_msg_cb;
  86   GNUNET_MULTICAST_MessageCallback message_cb;
  87   void *cb_cls;
  88
  89   /**
  90    * Function called after disconnected from the service.
  91    */
  92   GNUNET_ContinuationCallback disconnect_cb;
  93
  94   /**
  95    * Closure for @a disconnect_cb.
  96    */
  97   void *disconnect_cls;
  98
  99   /**
 100    * Are we currently transmitting a message?
 101    */
 102   uint8_t in_transmit;
 103
 104   /**
 105    * Number of MULTICAST_FRAGMENT_ACK messages we are still waiting for.
 106    */
 107   uint8_t acks_pending;
 108
 109   /**
 110    * Is this the origin or a member?
 111    */
 112   uint8_t is_origin;
 113
 114   /**
 115    * Is this channel in the process of disconnecting from the service?
 116    * #GNUNET_YES or #GNUNET_NO
 117    */
 118   uint8_t is_disconnecting;
 119 };
 120
 121
 122 /**
 123  * Handle for the origin of a multicast group.
 124  */
 125 struct GNUNET_MULTICAST_Origin
 126 {
 127   struct GNUNET_MULTICAST_Group grp;
 128   struct GNUNET_MULTICAST_OriginTransmitHandle tmit;
 129
 130   GNUNET_MULTICAST_RequestCallback request_cb;
 131 };
 132
 133
 134 /**
 135  * Handle for a multicast group member.
 136  */
 137 struct GNUNET_MULTICAST_Member
 138 {
 139   struct GNUNET_MULTICAST_Group grp;
 140   struct GNUNET_MULTICAST_MemberTransmitHandle tmit;
 141
 142   GNUNET_MULTICAST_JoinDecisionCallback join_dcsn_cb;
 143
 144   /**
 145    * Replay fragment -> struct GNUNET_MULTICAST_MemberReplayHandle *
 146    */
 147   struct GNUNET_CONTAINER_MultiHashMap *replay_reqs;
 148
 149   uint64_t next_fragment_id;
 150 };
 151
 152
 153 /**
 154  * Handle that identifies a join request.
 155  *
 156  * Used to match calls to #GNUNET_MULTICAST_JoinRequestCallback to the
 157  * corresponding calls to #GNUNET_MULTICAST_join_decision().
 158  */
 159 struct GNUNET_MULTICAST_JoinHandle
 160 {
 161   struct GNUNET_MULTICAST_Group *group;
 162
 163   /**
 164    * Public key of the member requesting join.
 165    */
 166   struct GNUNET_CRYPTO_EcdsaPublicKey member_key;
 167
 168   /**
 169    * Peer identity of the member requesting join.
 170    */
 171   struct GNUNET_PeerIdentity peer;
 172 };
 173
 174
 175 /**
 176  * Opaque handle to a replay request from the multicast service.
 177  */
 178 struct GNUNET_MULTICAST_ReplayHandle
 179 {
 180   struct GNUNET_MULTICAST_Group *grp;
 181   struct MulticastReplayRequestMessage req;
 182 };
 183
 184
 185 /**
 186  * Handle for a replay request.
 187  */
 188 struct GNUNET_MULTICAST_MemberReplayHandle
 189 {
 190 };
 191
 192
 193 static void
 194 origin_to_all (struct GNUNET_MULTICAST_Origin *orig);
 195
 196 static void
 197 member_to_origin (struct GNUNET_MULTICAST_Member *mem);
 198
 199
 200 /**
 201  * Send first message to the service after connecting.
 202  */
 203 static void
 204 group_send_connect_msg (struct GNUNET_MULTICAST_Group *grp)
 205 {
 206   uint16_t cmsg_size = ntohs (grp->connect_msg->size);
 207   struct GNUNET_MessageHeader * cmsg = GNUNET_malloc (cmsg_size);
 208   memcpy (cmsg, grp->connect_msg, cmsg_size);
 209   GNUNET_CLIENT_MANAGER_transmit_now (grp->client, cmsg);
 210 }
 211
 212
 213 /**
 214  * Got disconnected from service.  Reconnect.
 215  */
 216 static void
 217 group_recv_disconnect (void *cls,
 218                         struct GNUNET_CLIENT_MANAGER_Connection *client,
 219                         const struct GNUNET_MessageHeader *msg)
 220 {
 221   struct GNUNET_MULTICAST_Group *
 222     grp = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 223   GNUNET_CLIENT_MANAGER_reconnect (client);
 224   group_send_connect_msg (grp);
 225 }
 226
 227
 228 /**
 229  * Receive join request from service.
 230  */
 231 static void
 232 group_recv_join_request (void *cls,
 233                           struct GNUNET_CLIENT_MANAGER_Connection *client,
 234                           const struct GNUNET_MessageHeader *msg)
 235 {
 236   struct GNUNET_MULTICAST_Group *grp;
 237   const struct MulticastJoinRequestMessage *jreq;
 238   struct GNUNET_MULTICAST_JoinHandle *jh;
 239   const struct GNUNET_MessageHeader *jmsg;
 240
 241   grp = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 242   if (NULL == grp)
 243   {
 244     GNUNET_break (0);
 245     return;
 246   }
 247   if (NULL == grp->join_req_cb)
 248     return;
 249   /* FIXME: this fails to check that 'msg' is well-formed! */
 250   jreq = (const struct MulticastJoinRequestMessage *) msg;
 251   if (sizeof (*jreq) + sizeof (*jmsg) <= ntohs (jreq->header.size))
 252     jmsg = (const struct GNUNET_MessageHeader *) &jreq[1];
 253   else
 254     jmsg = NULL;
 255   jh = GNUNET_malloc (sizeof (*jh));
 256   jh->group = grp;
 257   jh->member_key = jreq->member_key;
 258   jh->peer = jreq->peer;
 259   grp->join_req_cb (grp->cb_cls, &jreq->member_key, jmsg, jh);
 260 }
 261
 262
 263 /**
 264  * Receive multicast message from service.
 265  */
 266 static void
 267 group_recv_message (void *cls,
 268                     struct GNUNET_CLIENT_MANAGER_Connection *client,
 269                     const struct GNUNET_MessageHeader *msg)
 270 {
 271   struct GNUNET_MULTICAST_Group *
 272     grp = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 273   struct GNUNET_MULTICAST_MessageHeader *
 274     mmsg = (struct GNUNET_MULTICAST_MessageHeader *) msg;
 275
 276   if (GNUNET_YES == grp->is_disconnecting)
 277     return;
 278
 279   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 280               "Calling message callback with a message of size %u.\n",
 281               ntohs (mmsg->header.size));
 282
 283   if (NULL != grp->message_cb)
 284     grp->message_cb (grp->cb_cls, mmsg);
 285 }
 286
 287
 288 /**
 289  * Receive message/request fragment acknowledgement from service.
 290  */
 291 static void
 292 group_recv_fragment_ack (void *cls,
 293                          struct GNUNET_CLIENT_MANAGER_Connection *client,
 294                          const struct GNUNET_MessageHeader *msg)
 295 {
 296   struct GNUNET_MULTICAST_Group *
 297     grp = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 298
 299   LOG (GNUNET_ERROR_TYPE_DEBUG,
 300        "%p Got fragment ACK. in_transmit=%u, acks_pending=%u\n",
 301        grp, grp->in_transmit, grp->acks_pending);
 302
 303   if (0 == grp->acks_pending)
 304   {
 305     LOG (GNUNET_ERROR_TYPE_DEBUG,
 306          "%p Ignoring extraneous fragment ACK.\n", grp);
 307     return;
 308   }
 309   grp->acks_pending--;
 310
 311   if (GNUNET_YES != grp->in_transmit)
 312     return;
 313
 314   if (GNUNET_YES == grp->is_origin)
 315     origin_to_all ((struct GNUNET_MULTICAST_Origin *) grp);
 316   else
 317     member_to_origin ((struct GNUNET_MULTICAST_Member *) grp);
 318 }
 319
 320 /**
 321  * Origin receives uniquest request from a member.
 322  */
 323 static void
 324 origin_recv_request (void *cls,
 325                      struct GNUNET_CLIENT_MANAGER_Connection *client,
 326                      const struct GNUNET_MessageHeader *msg)
 327 {
 328   struct GNUNET_MULTICAST_Group *grp;
 329   struct GNUNET_MULTICAST_Origin *
 330     orig = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 331   grp = &orig->grp;
 332   struct GNUNET_MULTICAST_RequestHeader *
 333     req = (struct GNUNET_MULTICAST_RequestHeader *) msg;
 334
 335   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 336               "Calling request callback with a request of size %u.\n",
 337               ntohs (req->header.size));
 338
 339   if (NULL != orig->request_cb)
 340     orig->request_cb (grp->cb_cls, req);
 341 }
 342
 343
 344 /**
 345  * Receive multicast replay request from service.
 346  */
 347 static void
 348 group_recv_replay_request (void *cls,
 349                            struct GNUNET_CLIENT_MANAGER_Connection *client,
 350                            const struct GNUNET_MessageHeader *msg)
 351 {
 352   struct GNUNET_MULTICAST_Group *
 353     grp = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 354   struct MulticastReplayRequestMessage *
 355     rep = (struct MulticastReplayRequestMessage *) msg;
 356
 357   if (GNUNET_YES == grp->is_disconnecting)
 358     return;
 359
 360   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got replay request.\n");
 361
 362   if (0 != rep->fragment_id)
 363   {
 364     if (NULL != grp->replay_frag_cb)
 365     {
 366       struct GNUNET_MULTICAST_ReplayHandle * rh = GNUNET_malloc (sizeof (*rh));
 367       rh->grp = grp;
 368       rh->req = *rep;
 369       grp->replay_frag_cb (grp->cb_cls, &rep->member_key,
 370                            GNUNET_ntohll (rep->fragment_id),
 371                            GNUNET_ntohll (rep->flags), rh);
 372     }
 373   }
 374   else if (0 != rep->message_id)
 375   {
 376     if (NULL != grp->replay_msg_cb)
 377     {
 378       struct GNUNET_MULTICAST_ReplayHandle * rh = GNUNET_malloc (sizeof (*rh));
 379       rh->grp = grp;
 380       rh->req = *rep;
 381       grp->replay_msg_cb (grp->cb_cls, &rep->member_key,
 382                           GNUNET_ntohll (rep->message_id),
 383                           GNUNET_ntohll (rep->fragment_offset),
 384                           GNUNET_ntohll (rep->flags), rh);
 385     }
 386   }
 387 }
 388
 389
 390 /**
 391  * Receive multicast replay request from service.
 392  */
 393 static void
 394 member_recv_replay_response (void *cls,
 395                             struct GNUNET_CLIENT_MANAGER_Connection *client,
 396                             const struct GNUNET_MessageHeader *msg)
 397 {
 398   struct GNUNET_MULTICAST_Group *grp;
 399   struct GNUNET_MULTICAST_Member *
 400     mem = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 401   grp = &mem->grp;
 402   struct MulticastReplayResponseMessage *
 403     res = (struct MulticastReplayResponseMessage *) msg;
 404
 405   if (GNUNET_YES == grp->is_disconnecting)
 406     return;
 407
 408   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got replay response.\n");
 409 }
 410
 411 /**
 412  * Member receives join decision.
 413  */
 414 static void
 415 member_recv_join_decision (void *cls,
 416                            struct GNUNET_CLIENT_MANAGER_Connection *client,
 417                            const struct GNUNET_MessageHeader *msg)
 418 {
 419   struct GNUNET_MULTICAST_Group *grp;
 420   struct GNUNET_MULTICAST_Member *
 421     mem = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*grp));
 422   grp = &mem->grp;
 423
 424   const struct MulticastJoinDecisionMessageHeader *
 425     hdcsn = (const struct MulticastJoinDecisionMessageHeader *) msg;
 426   const struct MulticastJoinDecisionMessage *
 427     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
 428
 429   uint16_t dcsn_size = ntohs (dcsn->header.size);
 430   int is_admitted = ntohl (dcsn->is_admitted);
 431
 432   LOG (GNUNET_ERROR_TYPE_DEBUG,
 433        "%p Member got join decision from multicast: %d\n",
 434        mem, is_admitted);
 435
 436   const struct GNUNET_MessageHeader *join_resp = NULL;
 437   uint16_t join_resp_size = 0;
 438
 439   uint16_t relay_count = ntohl (dcsn->relay_count);
 440   const struct GNUNET_PeerIdentity *relays = NULL;
 441   uint16_t relay_size = relay_count * sizeof (*relays);
 442   if (0 < relay_count)
 443   {
 444     if (dcsn_size < sizeof (*dcsn) + relay_size)
 445     {
 446       GNUNET_break_op (0);
 447       is_admitted = GNUNET_SYSERR;
 448     }
 449     else
 450     {
 451       relays = (struct GNUNET_PeerIdentity *) &dcsn[1];
 452     }
 453   }
 454
 455   if (sizeof (*dcsn) + relay_size + sizeof (*join_resp) <= dcsn_size)
 456   {
 457     join_resp = (const struct GNUNET_MessageHeader *) ((char *) &dcsn[1] + relay_size);
 458     join_resp_size = ntohs (join_resp->size);
 459   }
 460   if (dcsn_size < sizeof (*dcsn) + relay_size + join_resp_size)
 461   {
 462     LOG (GNUNET_ERROR_TYPE_DEBUG,
 463          "Received invalid join decision message from multicast: %u < %u + %u + %u\n",
 464          dcsn_size , sizeof (*dcsn), relay_size, join_resp_size);
 465     GNUNET_break_op (0);
 466     is_admitted = GNUNET_SYSERR;
 467   }
 468
 469   if (NULL != mem->join_dcsn_cb)
 470     mem->join_dcsn_cb (grp->cb_cls, is_admitted, &hdcsn->peer,
 471                        relay_count, relays, join_resp);
 472
 473   // FIXME:
 474   //if (GNUNET_YES != is_admitted)
 475   //  GNUNET_MULTICAST_member_part (mem);
 476 }
 477
 478
 479 /**
 480  * Message handlers for an origin.
 481  */
 482 static struct GNUNET_CLIENT_MANAGER_MessageHandler origin_handlers[] =
 483 {
 484   { group_recv_disconnect, NULL, 0, 0, GNUNET_NO },
 485
 486   { group_recv_message, NULL,
 487     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
 488     sizeof (struct GNUNET_MULTICAST_MessageHeader), GNUNET_YES },
 489
 490   { origin_recv_request, NULL,
 491     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST,
 492     sizeof (struct GNUNET_MULTICAST_RequestHeader), GNUNET_YES },
 493
 494   { group_recv_fragment_ack, NULL,
 495     GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK,
 496     sizeof (struct GNUNET_MessageHeader), GNUNET_YES },
 497
 498   { group_recv_join_request, NULL,
 499     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST,
 500     sizeof (struct MulticastJoinRequestMessage), GNUNET_YES },
 501
 502   { group_recv_replay_request, NULL,
 503     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
 504     sizeof (struct MulticastReplayRequestMessage), GNUNET_NO },
 505
 506   { NULL, NULL, 0, 0, GNUNET_NO }
 507 };
 508
 509
 510 /**
 511  * Message handlers for a member.
 512  */
 513 static struct GNUNET_CLIENT_MANAGER_MessageHandler member_handlers[] =
 514 {
 515   { group_recv_disconnect, NULL, 0, 0, GNUNET_NO },
 516
 517   { group_recv_message, NULL,
 518     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
 519     sizeof (struct GNUNET_MULTICAST_MessageHeader), GNUNET_YES },
 520
 521   { group_recv_fragment_ack, NULL,
 522     GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK,
 523     sizeof (struct GNUNET_MessageHeader), GNUNET_YES },
 524
 525   { group_recv_join_request, NULL,
 526     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST,
 527     sizeof (struct MulticastJoinRequestMessage), GNUNET_YES },
 528
 529   { member_recv_join_decision, NULL,
 530     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION,
 531     sizeof (struct MulticastJoinDecisionMessage), GNUNET_YES },
 532
 533   { group_recv_replay_request, NULL,
 534     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
 535     sizeof (struct MulticastReplayRequestMessage), GNUNET_NO },
 536
 537   { member_recv_replay_response, NULL,
 538     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
 539     sizeof (struct MulticastReplayRequestMessage), GNUNET_NO },
 540
 541   { NULL, NULL, 0, 0, GNUNET_NO }
 542 };
 543
 544
 545 static void
 546 group_cleanup (struct GNUNET_MULTICAST_Group *grp)
 547 {
 548   GNUNET_free (grp->connect_msg);
 549   if (NULL != grp->disconnect_cb)
 550     grp->disconnect_cb (grp->disconnect_cls);
 551 }
 552
 553
 554 static void
 555 origin_cleanup (void *cls)
 556 {
 557   struct GNUNET_MULTICAST_Origin *orig = cls;
 558   group_cleanup (&orig->grp);
 559   GNUNET_free (orig);
 560 }
 561
 562
 563 static void
 564 member_cleanup (void *cls)
 565 {
 566   struct GNUNET_MULTICAST_Member *mem = cls;
 567   group_cleanup (&mem->grp);
 568   GNUNET_free (mem);
 569 }
 570
 571
 572 /**
 573  * Function to call with the decision made for a join request.
 574  *
 575  * Must be called once and only once in response to an invocation of the
 576  * #GNUNET_MULTICAST_JoinRequestCallback.
 577  *
 578  * @param join
 579  *        Join request handle.
 580  * @param is_admitted
 581  *        #GNUNET_YES    if the join is approved,
 582  *        #GNUNET_NO     if it is disapproved,
 583  *        #GNUNET_SYSERR if we cannot answer the request.
 584  * @param relay_count
 585  *        Number of relays given.
 586  * @param relays
 587  *        Array of suggested peers that might be useful relays to use
 588  *        when joining the multicast group (essentially a list of peers that
 589  *        are already part of the multicast group and might thus be willing
 590  *        to help with routing).  If empty, only this local peer (which must
 591  *        be the multicast origin) is a good candidate for building the
 592  *        multicast tree.  Note that it is unnecessary to specify our own
 593  *        peer identity in this array.
 594  * @param join_resp
 595  *        Message to send in response to the joining peer;
 596  *        can also be used to redirect the peer to a different group at the
 597  *        application layer; this response is to be transmitted to the
 598  *        peer that issued the request even if admission is denied.
 599  */
 600 struct GNUNET_MULTICAST_ReplayHandle *
 601 GNUNET_MULTICAST_join_decision (struct GNUNET_MULTICAST_JoinHandle *join,
 602                                 int is_admitted,
 603                                 uint16_t relay_count,
 604                                 const struct GNUNET_PeerIdentity *relays,
 605                                 const struct GNUNET_MessageHeader *join_resp)
 606 {
 607   struct GNUNET_MULTICAST_Group *grp = join->group;
 608   uint16_t join_resp_size = (NULL != join_resp) ? ntohs (join_resp->size) : 0;
 609   uint16_t relay_size = relay_count * sizeof (*relays);
 610
 611   struct MulticastJoinDecisionMessageHeader * hdcsn;
 612   struct MulticastJoinDecisionMessage *dcsn;
 613   hdcsn = GNUNET_malloc (sizeof (*hdcsn) + sizeof (*dcsn)
 614                          + relay_size + join_resp_size);
 615   hdcsn->header.size = htons (sizeof (*hdcsn) + sizeof (*dcsn)
 616                               + relay_size + join_resp_size);
 617   hdcsn->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION);
 618   hdcsn->member_key = join->member_key;
 619   hdcsn->peer = join->peer;
 620
 621   dcsn = (struct MulticastJoinDecisionMessage *) &hdcsn[1];
 622   dcsn->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION);
 623   dcsn->header.size = htons (sizeof (*dcsn) + relay_size + join_resp_size);
 624   dcsn->is_admitted = htonl (is_admitted);
 625   dcsn->relay_count = htonl (relay_count);
 626   if (0 < relay_size)
 627     memcpy (&dcsn[1], relays, relay_size);
 628   if (0 < join_resp_size)
 629     memcpy (((char *) &dcsn[1]) + relay_size, join_resp, join_resp_size);
 630
 631   GNUNET_CLIENT_MANAGER_transmit (grp->client, &hdcsn->header);
 632   GNUNET_free (hdcsn);
 633   GNUNET_free (join);
 634   return NULL;
 635 }
 636
 637
 638 /**
 639  * Replay a message fragment for the multicast group.
 640  *
 641  * @param rh
 642  *        Replay handle identifying which replay operation was requested.
 643  * @param msg
 644  *        Replayed message fragment, NULL if not found / an error occurred.
 645  * @param ec
 646  *        Error code.  See enum GNUNET_MULTICAST_ReplayErrorCode
 647  *        If not #GNUNET_MULTICAST_REC_OK, the replay handle is invalidated.
 648  */
 649 void
 650 GNUNET_MULTICAST_replay_response (struct GNUNET_MULTICAST_ReplayHandle *rh,
 651                                   const struct GNUNET_MessageHeader *msg,
 652                                   enum GNUNET_MULTICAST_ReplayErrorCode ec)
 653 {
 654   uint8_t msg_size = (NULL != msg) ? ntohs (msg->size) : 0;
 655   struct MulticastReplayResponseMessage *
 656     res = GNUNET_malloc (sizeof (*res) + msg_size);
 657   *res = (struct MulticastReplayResponseMessage) {
 658     .header = {
 659       .type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE),
 660       .size = htons (sizeof (*res) + msg_size),
 661     },
 662     .fragment_id = rh->req.fragment_id,
 663     .message_id = rh->req.message_id,
 664     .fragment_offset = rh->req.fragment_offset,
 665     .flags = rh->req.flags,
 666     .error_code = htonl (ec),
 667   };
 668
 669   if (GNUNET_MULTICAST_REC_OK == ec)
 670   {
 671     GNUNET_assert (NULL != msg);
 672     memcpy (&res[1], msg, msg_size);
 673   }
 674
 675   GNUNET_CLIENT_MANAGER_transmit (rh->grp->client, &res->header);
 676   GNUNET_free (res);
 677
 678   if (GNUNET_MULTICAST_REC_OK != ec)
 679     GNUNET_free (rh);
 680 }
 681
 682
 683 /**
 684  * Indicate the end of the replay session.
 685  *
 686  * Invalidates the replay handle.
 687  *
 688  * @param rh
 689  *        Replay session to end.
 690  */
 691 void
 692 GNUNET_MULTICAST_replay_response_end (struct GNUNET_MULTICAST_ReplayHandle *rh)
 693 {
 694   struct MulticastReplayResponseMessage end = {
 695     .header = {
 696       .type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE_END),
 697       .size = htons (sizeof (end)),
 698     },
 699     .fragment_id = rh->req.fragment_id,
 700     .message_id = rh->req.message_id,
 701     .fragment_offset = rh->req.fragment_offset,
 702     .flags = rh->req.flags,
 703   };
 704
 705   GNUNET_CLIENT_MANAGER_transmit (rh->grp->client, &end.header);
 706   GNUNET_free (rh);
 707 }
 708
 709
 710 /**
 711  * Replay a message for the multicast group.
 712  *
 713  * @param rh
 714  *        Replay handle identifying which replay operation was requested.
 715  * @param notify
 716  *        Function to call to get the message.
 717  * @param notify_cls
 718  *        Closure for @a notify.
 719  */
 720 void
 721 GNUNET_MULTICAST_replay_response2 (struct GNUNET_MULTICAST_ReplayHandle *rh,
 722                                    GNUNET_MULTICAST_ReplayTransmitNotify notify,
 723                                    void *notify_cls)
 724 {
 725 }
 726
 727
 728 /**
 729  * Start a multicast group.
 730  *
 731  * Will advertise the origin in the P2P overlay network under the respective
 732  * public key so that other peer can find this peer to join it.  Peers that
 733  * issue GNUNET_MULTICAST_member_join() can then transmit a join request to
 734  * either an existing group member or to the origin.  If the joining is
 735  * approved, the member is cleared for @e replay and will begin to receive
 736  * messages transmitted to the group.  If joining is disapproved, the failed
 737  * candidate will be given a response.  Members in the group can send messages
 738  * to the origin (one at a time).
 739  *
 740  * @param cfg
 741  *        Configuration to use.
 742  * @param priv_key
 743  *        ECC key that will be used to sign messages for this
 744  *        multicast session; public key is used to identify the multicast group;
 745  * @param max_fragment_id
 746  *        Maximum fragment ID already sent to the group.
 747  *        0 for a new group.
 748  * @param join_request_cb
 749  *        Function called to approve / disapprove joining of a peer.
 750  * @param replay_frag_cb
 751  *        Function that can be called to replay a message fragment.
 752  * @param replay_msg_cb
 753  *        Function that can be called to replay a message.
 754  * @param request_cb
 755  *        Function called with message fragments from group members.
 756  * @param message_cb
 757  *        Function called with the message fragments sent to the
 758  *        network by GNUNET_MULTICAST_origin_to_all().  These message fragments
 759  *        should be stored for answering replay requests later.
 760  * @param cls
 761  *        Closure for the various callbacks that follow.
 762  *
 763  * @return Handle for the origin, NULL on error.
 764  */
 765 struct GNUNET_MULTICAST_Origin *
 766 GNUNET_MULTICAST_origin_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
 767                                const struct GNUNET_CRYPTO_EddsaPrivateKey *priv_key,
 768                                uint64_t max_fragment_id,
 769                                GNUNET_MULTICAST_JoinRequestCallback join_request_cb,
 770                                GNUNET_MULTICAST_ReplayFragmentCallback replay_frag_cb,
 771                                GNUNET_MULTICAST_ReplayMessageCallback replay_msg_cb,
 772                                GNUNET_MULTICAST_RequestCallback request_cb,
 773                                GNUNET_MULTICAST_MessageCallback message_cb,
 774                                void *cls)
 775 {
 776   struct GNUNET_MULTICAST_Origin *orig = GNUNET_malloc (sizeof (*orig));
 777   struct GNUNET_MULTICAST_Group *grp = &orig->grp;
 778   struct MulticastOriginStartMessage *start = GNUNET_malloc (sizeof (*start));
 779
 780   start->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_ORIGIN_START);
 781   start->header.size = htons (sizeof (*start));
 782   start->max_fragment_id = max_fragment_id;
 783   memcpy (&start->group_key, priv_key, sizeof (*priv_key));
 784
 785   grp->connect_msg = (struct GNUNET_MessageHeader *) start;
 786   grp->is_origin = GNUNET_YES;
 787   grp->cfg = cfg;
 788
 789   grp->cb_cls = cls;
 790   grp->join_req_cb = join_request_cb;
 791   grp->replay_frag_cb = replay_frag_cb;
 792   grp->replay_msg_cb = replay_msg_cb;
 793   grp->message_cb = message_cb;
 794
 795   orig->request_cb = request_cb;
 796
 797   grp->client = GNUNET_CLIENT_MANAGER_connect (cfg, "multicast", origin_handlers);
 798   GNUNET_CLIENT_MANAGER_set_user_context_ (grp->client, orig, sizeof (*grp));
 799   group_send_connect_msg (grp);
 800
 801   return orig;
 802 }
 803
 804
 805 /**
 806  * Stop a multicast group.
 807  *
 808  * @param origin
 809  *        Multicast group to stop.
 810  */
 811 void
 812 GNUNET_MULTICAST_origin_stop (struct GNUNET_MULTICAST_Origin *orig,
 813                               GNUNET_ContinuationCallback stop_cb,
 814                               void *stop_cls)
 815 {
 816   struct GNUNET_MULTICAST_Group *grp = &orig->grp;
 817
 818   grp->is_disconnecting = GNUNET_YES;
 819   grp->disconnect_cb = stop_cb;
 820   grp->disconnect_cls = stop_cls;
 821
 822   GNUNET_CLIENT_MANAGER_disconnect (orig->grp.client, GNUNET_YES,
 823                                     &origin_cleanup, orig);
 824 }
 825
 826
 827 static void
 828 origin_to_all (struct GNUNET_MULTICAST_Origin *orig)
 829 {
 830   LOG (GNUNET_ERROR_TYPE_DEBUG, "%p origin_to_all()\n", orig);
 831   struct GNUNET_MULTICAST_Group *grp = &orig->grp;
 832   struct GNUNET_MULTICAST_OriginTransmitHandle *tmit = &orig->tmit;
 833   GNUNET_assert (GNUNET_YES == grp->in_transmit);
 834
 835   size_t buf_size = GNUNET_MULTICAST_FRAGMENT_MAX_SIZE;
 836   struct GNUNET_MULTICAST_MessageHeader *msg = GNUNET_malloc (buf_size);
 837   int ret = tmit->notify (tmit->notify_cls, &buf_size, &msg[1]);
 838
 839   if (! (GNUNET_YES == ret || GNUNET_NO == ret)
 840       || GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < buf_size)
 841   {
 842     LOG (GNUNET_ERROR_TYPE_ERROR,
 843          "%p OriginTransmitNotify() returned error or invalid message size.\n",
 844          orig);
 845     /* FIXME: handle error */
 846     GNUNET_free (msg);
 847     return;
 848   }
 849
 850   if (GNUNET_NO == ret && 0 == buf_size)
 851   {
 852     LOG (GNUNET_ERROR_TYPE_DEBUG,
 853          "%p OriginTransmitNotify() - transmission paused.\n", orig);
 854     GNUNET_free (msg);
 855     return; /* Transmission paused. */
 856   }
 857
 858   msg->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
 859   msg->header.size = htons (sizeof (*msg) + buf_size);
 860   msg->message_id = GNUNET_htonll (tmit->message_id);
 861   msg->group_generation = tmit->group_generation;
 862   msg->fragment_offset = GNUNET_htonll (tmit->fragment_offset);
 863   tmit->fragment_offset += sizeof (*msg) + buf_size;
 864
 865   grp->acks_pending++;
 866   GNUNET_CLIENT_MANAGER_transmit (grp->client, &msg->header);
 867   GNUNET_free (msg);
 868
 869   if (GNUNET_YES == ret)
 870     grp->in_transmit = GNUNET_NO;
 871 }
 872
 873
 874 /**
 875  * Send a message to the multicast group.
 876  *
 877  * @param orig
 878  *        Handle to the multicast group.
 879  * @param message_id
 880  *        Application layer ID for the message.  Opaque to multicast.
 881  * @param group_generation
 882  *        Group generation of the message.
 883  *        Documented in struct GNUNET_MULTICAST_MessageHeader.
 884  * @param notify
 885  *        Function to call to get the message.
 886  * @param notify_cls
 887  *        Closure for @a notify.
 888  *
 889  * @return Message handle on success,
 890  *         NULL on error (i.e. another request is already pending).
 891  */
 892 struct GNUNET_MULTICAST_OriginTransmitHandle *
 893 GNUNET_MULTICAST_origin_to_all (struct GNUNET_MULTICAST_Origin *orig,
 894                                 uint64_t message_id,
 895                                 uint64_t group_generation,
 896                                 GNUNET_MULTICAST_OriginTransmitNotify notify,
 897                                 void *notify_cls)
 898 {
 899   struct GNUNET_MULTICAST_Group *grp = &orig->grp;
 900   if (GNUNET_YES == grp->in_transmit)
 901     return NULL;
 902   grp->in_transmit = GNUNET_YES;
 903
 904   struct GNUNET_MULTICAST_OriginTransmitHandle *tmit = &orig->tmit;
 905   tmit->origin = orig;
 906   tmit->message_id = message_id;
 907   tmit->group_generation = group_generation;
 908   tmit->notify = notify;
 909   tmit->notify_cls = notify_cls;
 910
 911   origin_to_all (orig);
 912   return tmit;
 913 }
 914
 915
 916 /**
 917  * Resume message transmission to multicast group.
 918  *
 919  * @param th
 920  *        Transmission to cancel.
 921  */
 922 void
 923 GNUNET_MULTICAST_origin_to_all_resume (struct GNUNET_MULTICAST_OriginTransmitHandle *th)
 924 {
 925   struct GNUNET_MULTICAST_Group *grp = &th->origin->grp;
 926   if (0 != grp->acks_pending || GNUNET_YES != grp->in_transmit)
 927     return;
 928   origin_to_all (th->origin);
 929 }
 930
 931
 932 /**
 933  * Cancel request for message transmission to multicast group.
 934  *
 935  * @param th
 936  *        Transmission to cancel.
 937  */
 938 void
 939 GNUNET_MULTICAST_origin_to_all_cancel (struct GNUNET_MULTICAST_OriginTransmitHandle *th)
 940 {
 941   th->origin->grp.in_transmit = GNUNET_NO;
 942 }
 943
 944
 945 /**
 946  * Join a multicast group.
 947  *
 948  * The entity joining is always the local peer.  Further information about the
 949  * candidate can be provided in the @a join_request message.  If the join fails, the
 950  * @a message_cb is invoked with a (failure) response and then with NULL.  If
 951  * the join succeeds, outstanding (state) messages and ongoing multicast
 952  * messages will be given to the @a message_cb until the member decides to part
 953  * the group.  The @a replay_cb function may be called at any time by the
 954  * multicast service to support relaying messages to other members of the group.
 955  *
 956  * @param cfg
 957  *        Configuration to use.
 958  * @param group_key
 959  *        ECC public key that identifies the group to join.
 960  * @param member_key
 961  *        ECC key that identifies the member
 962  *        and used to sign requests sent to the origin.
 963  * @param origin
 964  *        Peer ID of the origin to send unicast requsets to.  If NULL,
 965  *        unicast requests are sent back via multiple hops on the reverse path
 966  *        of multicast messages.
 967  * @param relay_count
 968  *        Number of peers in the @a relays array.
 969  * @param relays
 970  *        Peer identities of members of the group, which serve as relays
 971  *        and can be used to join the group at. and send the @a join_request to.
 972  *        If empty, the @a join_request is sent directly to the @a origin.
 973  * @param join_msg
 974  *        Application-dependent join message to be passed to the peer @a origin.
 975  * @param join_request_cb
 976  *        Function called to approve / disapprove joining of a peer.
 977  * @param join_decision_cb
 978  *        Function called to inform about the join decision.
 979  * @param replay_frag_cb
 980  *        Function that can be called to replay message fragments
 981  *        this peer already knows from this group. NULL if this
 982  *        client is unable to support replay.
 983  * @param replay_msg_cb
 984  *        Function that can be called to replay message fragments
 985  *        this peer already knows from this group. NULL if this
 986  *        client is unable to support replay.
 987  * @param message_cb
 988  *        Function to be called for all message fragments we
 989  *        receive from the group, excluding those our @a replay_cb
 990  *        already has.
 991  * @param cls
 992  *        Closure for callbacks.
 993  *
 994  * @return Handle for the member, NULL on error.
 995  */
 996 struct GNUNET_MULTICAST_Member *
 997 GNUNET_MULTICAST_member_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
 998                               const struct GNUNET_CRYPTO_EddsaPublicKey *group_key,
 999                               const struct GNUNET_CRYPTO_EcdsaPrivateKey *member_key,
1000                               const struct GNUNET_PeerIdentity *origin,
1001                               uint16_t relay_count,
1002                               const struct GNUNET_PeerIdentity *relays,
1003                               const struct GNUNET_MessageHeader *join_msg,
1004                               GNUNET_MULTICAST_JoinRequestCallback join_request_cb,
1005                               GNUNET_MULTICAST_JoinDecisionCallback join_decision_cb,
1006                               GNUNET_MULTICAST_ReplayFragmentCallback replay_frag_cb,
1007                               GNUNET_MULTICAST_ReplayMessageCallback replay_msg_cb,
1008                               GNUNET_MULTICAST_MessageCallback message_cb,
1009                               void *cls)
1010 {
1011   struct GNUNET_MULTICAST_Member *mem = GNUNET_malloc (sizeof (*mem));
1012   struct GNUNET_MULTICAST_Group *grp = &mem->grp;
1013
1014   uint16_t relay_size = relay_count * sizeof (*relays);
1015   uint16_t join_msg_size = (NULL != join_msg) ? ntohs (join_msg->size) : 0;
1016   struct MulticastMemberJoinMessage *
1017     join = GNUNET_malloc (sizeof (*join) + relay_size + join_msg_size);
1018   join->header.size = htons (sizeof (*join) + relay_size + join_msg_size);
1019   join->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MEMBER_JOIN);
1020   join->group_key = *group_key;
1021   join->member_key = *member_key;
1022   join->origin = *origin;
1023   join->relay_count = ntohl (relay_count);
1024   if (0 < relay_size)
1025     memcpy (&join[1], relays, relay_size);
1026   if (0 < join_msg_size)
1027     memcpy (((char *) &join[1]) + relay_size, join_msg, join_msg_size);
1028
1029   grp->connect_msg = (struct GNUNET_MessageHeader *) join;
1030   grp->is_origin = GNUNET_NO;
1031   grp->cfg = cfg;
1032
1033   mem->join_dcsn_cb = join_decision_cb;
1034   grp->join_req_cb = join_request_cb;
1035   grp->replay_frag_cb = replay_frag_cb;
1036   grp->replay_msg_cb = replay_msg_cb;
1037   grp->message_cb = message_cb;
1038   grp->cb_cls = cls;
1039
1040   grp->client = GNUNET_CLIENT_MANAGER_connect (cfg, "multicast", member_handlers);
1041   GNUNET_CLIENT_MANAGER_set_user_context_ (grp->client, mem, sizeof (*grp));
1042   group_send_connect_msg (grp);
1043
1044   return mem;
1045 }
1046
1047
1048 /**
1049  * Part a multicast group.
1050  *
1051  * Disconnects from all group members and invalidates the @a member handle.
1052  *
1053  * An application-dependent part message can be transmitted beforehand using
1054  * #GNUNET_MULTICAST_member_to_origin())
1055  *
1056  * @param member
1057  *        Membership handle.
1058  */
1059 void
1060 GNUNET_MULTICAST_member_part (struct GNUNET_MULTICAST_Member *mem,
1061                               GNUNET_ContinuationCallback part_cb,
1062                               void *part_cls)
1063 {
1064   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p Member parting.\n", mem);
1065   struct GNUNET_MULTICAST_Group *grp = &mem->grp;
1066
1067   grp->is_disconnecting = GNUNET_YES;
1068   grp->disconnect_cb = part_cb;
1069   grp->disconnect_cls = part_cls;
1070
1071   mem->join_dcsn_cb = NULL;
1072   grp->join_req_cb = NULL;
1073   grp->message_cb = NULL;
1074   grp->replay_msg_cb = NULL;
1075   grp->replay_frag_cb = NULL;
1076
1077   GNUNET_CLIENT_MANAGER_disconnect (mem->grp.client, GNUNET_YES,
1078                                     member_cleanup, mem);
1079 }
1080
1081
1082 void
1083 member_replay_request (struct GNUNET_MULTICAST_Member *mem,
1084                        uint64_t fragment_id,
1085                        uint64_t message_id,
1086                        uint64_t fragment_offset,
1087                        uint64_t flags)
1088 {
1089   struct MulticastReplayRequestMessage rep = {
1090     .header = {
1091       .type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST),
1092       .size = htons (sizeof (rep)),
1093     },
1094     .fragment_id = GNUNET_htonll (fragment_id),
1095     .message_id = GNUNET_htonll (message_id),
1096     .fragment_offset = GNUNET_htonll (fragment_offset),
1097     .flags = GNUNET_htonll (flags),
1098   };
1099   GNUNET_CLIENT_MANAGER_transmit (mem->grp.client, &rep.header);
1100 }
1101
1102
1103 /**
1104  * Request a fragment to be replayed by fragment ID.
1105  *
1106  * Useful if messages below the @e max_known_fragment_id given when joining are
1107  * needed and not known to the client.
1108  *
1109  * @param member
1110  *        Membership handle.
1111  * @param fragment_id
1112  *        ID of a message fragment that this client would like to see replayed.
1113  * @param flags
1114  *        Additional flags for the replay request.
1115  *        It is used and defined by GNUNET_MULTICAST_ReplayFragmentCallback
1116  *
1117  * @return Replay request handle.
1118  */
1119 struct GNUNET_MULTICAST_MemberReplayHandle *
1120 GNUNET_MULTICAST_member_replay_fragment (struct GNUNET_MULTICAST_Member *mem,
1121                                          uint64_t fragment_id,
1122                                          uint64_t flags)
1123 {
1124   member_replay_request (mem, fragment_id, 0, 0, flags);
1125 }
1126
1127
1128 /**
1129  * Request a message fragment to be replayed.
1130  *
1131  * Useful if messages below the @e max_known_fragment_id given when joining are
1132  * needed and not known to the client.
1133  *
1134  * @param member
1135  *        Membership handle.
1136  * @param message_id
1137  *        ID of the message this client would like to see replayed.
1138  * @param fragment_offset
1139  *        Offset of the fragment within the message to replay.
1140  * @param flags
1141  *        Additional flags for the replay request.
1142  *        It is used & defined by GNUNET_MULTICAST_ReplayMessageCallback
1143  *
1144  * @return Replay request handle, NULL on error.
1145  */
1146 struct GNUNET_MULTICAST_MemberReplayHandle *
1147 GNUNET_MULTICAST_member_replay_message (struct GNUNET_MULTICAST_Member *mem,
1148                                         uint64_t message_id,
1149                                         uint64_t fragment_offset,
1150                                         uint64_t flags)
1151 {
1152   member_replay_request (mem, 0, message_id, fragment_offset, flags);
1153 }
1154
1155
1156 static void
1157 member_to_origin (struct GNUNET_MULTICAST_Member *mem)
1158 {
1159   LOG (GNUNET_ERROR_TYPE_DEBUG, "member_to_origin()\n");
1160   struct GNUNET_MULTICAST_Group *grp = &mem->grp;
1161   struct GNUNET_MULTICAST_MemberTransmitHandle *tmit = &mem->tmit;
1162   GNUNET_assert (GNUNET_YES == grp->in_transmit);
1163
1164   size_t buf_size = GNUNET_MULTICAST_FRAGMENT_MAX_SIZE;
1165   struct GNUNET_MULTICAST_RequestHeader *req = GNUNET_malloc (buf_size);
1166   int ret = tmit->notify (tmit->notify_cls, &buf_size, &req[1]);
1167
1168   if (! (GNUNET_YES == ret || GNUNET_NO == ret)
1169       || GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < buf_size)
1170   {
1171     LOG (GNUNET_ERROR_TYPE_ERROR,
1172          "MemberTransmitNotify() returned error or invalid message size. "
1173          "ret=%d, buf_size=%u\n", ret, buf_size);
1174     /* FIXME: handle error */
1175     GNUNET_free (req);
1176     return;
1177   }
1178
1179   if (GNUNET_NO == ret && 0 == buf_size)
1180   {
1181     /* Transmission paused. */
1182     GNUNET_free (req);
1183     return;
1184   }
1185
1186   req->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST);
1187   req->header.size = htons (sizeof (*req) + buf_size);
1188   req->request_id = GNUNET_htonll (tmit->request_id);
1189   req->fragment_offset = GNUNET_ntohll (tmit->fragment_offset);
1190   tmit->fragment_offset += sizeof (*req) + buf_size;
1191
1192   GNUNET_CLIENT_MANAGER_transmit (grp->client, &req->header);
1193   GNUNET_free (req);
1194
1195   if (GNUNET_YES == ret)
1196     grp->in_transmit = GNUNET_NO;
1197 }
1198
1199
1200 /**
1201  * Send a message to the origin of the multicast group.
1202  *
1203  * @param mem
1204  *        Membership handle.
1205  * @param request_id
1206  *        Application layer ID for the request.  Opaque to multicast.
1207  * @param notify
1208  *        Callback to call to get the message.
1209  * @param notify_cls
1210  *        Closure for @a notify.
1211  *
1212  * @return Handle to cancel request, NULL on error (i.e. request already pending).
1213  */
1214 struct GNUNET_MULTICAST_MemberTransmitHandle *
1215 GNUNET_MULTICAST_member_to_origin (struct GNUNET_MULTICAST_Member *mem,
1216                                    uint64_t request_id,
1217                                    GNUNET_MULTICAST_MemberTransmitNotify notify,
1218                                    void *notify_cls)
1219 {
1220   if (GNUNET_YES == mem->grp.in_transmit)
1221     return NULL;
1222   mem->grp.in_transmit = GNUNET_YES;
1223
1224   struct GNUNET_MULTICAST_MemberTransmitHandle *tmit = &mem->tmit;
1225   tmit->member = mem;
1226   tmit->request_id = request_id;
1227   tmit->notify = notify;
1228   tmit->notify_cls = notify_cls;
1229
1230   member_to_origin (mem);
1231   return tmit;
1232 }
1233
1234
1235 /**
1236  * Resume message transmission to origin.
1237  *
1238  * @param th
1239  *        Transmission to cancel.
1240  */
1241 void
1242 GNUNET_MULTICAST_member_to_origin_resume (struct GNUNET_MULTICAST_MemberTransmitHandle *th)
1243 {
1244   struct GNUNET_MULTICAST_Group *grp = &th->member->grp;
1245   if (0 != grp->acks_pending || GNUNET_YES != grp->in_transmit)
1246     return;
1247   member_to_origin (th->member);
1248 }
1249
1250
1251 /**
1252  * Cancel request for message transmission to origin.
1253  *
1254  * @param th
1255  *        Transmission to cancel.
1256  */
1257 void
1258 GNUNET_MULTICAST_member_to_origin_cancel (struct GNUNET_MULTICAST_MemberTransmitHandle *th)
1259 {
1260   th->member->grp.in_transmit = GNUNET_NO;
1261 }
1262
1263
1264 /* end of multicast_api.c */