src/multicast/multicast_api.c

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