src/psyc/psyc_api.c

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