+struct StateModifyClosure
+{
+ const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key;
+ struct GNUNET_PSYC_ReceiveHandle *recv;
+ enum GNUNET_PSYC_MessageState msg_state;
+ char mod_oper;
+ char *mod_name;
+ char *mod_value;
+ uint64_t mod_value_size;
+ uint64_t mod_value_remaining;
+};
+
+
+static void
+recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
+ uint32_t flags, const struct GNUNET_MessageHeader *msg)
+{
+ struct StateModifyClosure *scls = cls;
+ uint16_t psize;
+ if (NULL == msg)
+ {
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
+ return;
+ }
+
+ switch (ntohs (msg->type))
+ {
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+ {
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_METHOD;
+ break;
+ }
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+ {
+ struct GNUNET_PSYC_MessageModifier *
+ pmod = (struct GNUNET_PSYC_MessageModifier *) msg;
+ psize = ntohs (pmod->header.size);
+ uint16_t name_size = ntohs (pmod->name_size);
+ uint16_t value_size = ntohs (pmod->value_size);
+
+ const char *name = (const char *) &pmod[1];
+ const void *value = name + name_size;
+
+ if (GNUNET_ENV_OP_SET != pmod->oper)
+ { // Apply non-transient operation.
+ if (psize == sizeof (*pmod) + name_size + value_size)
+ {
+ db->state_modify_op (db->cls, scls->channel_key,
+ pmod->oper, name, value, value_size);
+ }
+ else
+ {
+ scls->mod_oper = pmod->oper;
+ scls->mod_name = GNUNET_malloc (name_size);
+ memcpy (scls->mod_name, name, name_size);
+
+ scls->mod_value_size = value_size;
+ scls->mod_value = GNUNET_malloc (scls->mod_value_size);
+ scls->mod_value_remaining
+ = scls->mod_value_size - (psize - sizeof (*pmod) - name_size);
+ memcpy (scls->mod_value, value, value_size - scls->mod_value_remaining);
+ }
+ }
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MODIFIER;
+ break;
+ }
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+ if (GNUNET_ENV_OP_SET != scls->mod_oper)
+ {
+ if (scls->mod_value_remaining == 0)
+ {
+ GNUNET_break_op (0);
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
+ }
+ psize = ntohs (msg->size);
+ memcpy (scls->mod_value + (scls->mod_value_size - scls->mod_value_remaining),
+ &msg[1], psize - sizeof (*msg));
+ scls->mod_value_remaining -= psize - sizeof (*msg);
+ if (0 == scls->mod_value_remaining)
+ {
+ db->state_modify_op (db->cls, scls->channel_key,
+ scls->mod_oper, scls->mod_name,
+ scls->mod_value, scls->mod_value_size);
+ GNUNET_free (scls->mod_name);
+ GNUNET_free (scls->mod_value);
+ scls->mod_oper = 0;
+ scls->mod_name = NULL;
+ scls->mod_value = NULL;
+ scls->mod_value_size = 0;
+ }
+ }
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_MOD_CONT;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_DATA;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_END;
+ break;
+
+ default:
+ scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
+ }
+}
+
+
+static int
+recv_state_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
+ enum GNUNET_PSYCSTORE_MessageFlags flags)
+{
+ struct StateModifyClosure *scls = cls;
+
+ if (NULL == scls->recv)
+ {
+ scls->recv = GNUNET_PSYC_receive_create (NULL, &recv_state_message_part,
+ scls);
+ }
+
+ const struct GNUNET_PSYC_MessageHeader *
+ pmsg = (const struct GNUNET_PSYC_MessageHeader *) &msg[1];
+ GNUNET_PSYC_receive_message (scls->recv, pmsg);
+
+ return GNUNET_YES;
+}
+
+