peerstore_api_monitor.c File Reference

#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_protocols.h"
#include "peerstore.h"
#include "peerstore_common.h"
#include "gnunet_peerstore_service.h"

Include dependency graph for peerstore_api_monitor.c:

Go to the source code of this file.

Data Structures
struct	GNUNET_PEERSTORE_Monitor
	Context for a monitor. More...

Macros
#define	LOG(kind, ...)

Functions
static void	handle_sync (void *cls, const struct GNUNET_MessageHeader *msg)
static int	check_result (void *cls, const struct PeerstoreRecordMessage *msg)
	When a response for iterate request is received, check the message is well-formed.
static void	handle_result (void *cls, const struct PeerstoreRecordMessage *msg)
	When a response to monitor is received.
static void	reconnect (struct GNUNET_PEERSTORE_Monitor *mc)
static void	mq_error_handler (void *cls, enum GNUNET_MQ_Error err)
struct GNUNET_PEERSTORE_Monitor *	GNUNET_PEERSTORE_monitor_start (const struct GNUNET_CONFIGURATION_Handle *cfg, int iterate_first, const char *sub_system, const struct GNUNET_PeerIdentity *peer, const char *key, GNUNET_SCHEDULER_TaskCallback error_cb, void *error_cb_cls, GNUNET_SCHEDULER_TaskCallback sync_cb, void *sync_cb_cls, GNUNET_PEERSTORE_Processor callback, void *callback_cls)
	Request watching a given key The monitoring can be filtered to contain only records matching peer and/or key.
void	GNUNET_PEERSTORE_monitor_stop (struct GNUNET_PEERSTORE_Monitor *zm)
	Stop monitoring.
void	GNUNET_PEERSTORE_monitor_next (struct GNUNET_PEERSTORE_Monitor *zm, uint64_t limit)
	Calls the monitor processor specified in GNUNET_PEERSTORE_monitor_start for the next record(s).

Macro Definition Documentation

LOG

#define LOG	(		kind,
			...
	)

Value:

                                        GNUNET_log_from (kind, "peerstore-monitor-api", \
                                        __VA_ARGS__)

Definition at line 32 of file peerstore_api_monitor.c.

{
  struct GNUNET_PEERSTORE_Monitor *next;
 
  struct GNUNET_PEERSTORE_Monitor *prev;
 
  GNUNET_SCHEDULER_TaskCallback error_cb;
 
  void *error_cb_cls;
 
  GNUNET_PEERSTORE_Processor callback;
 
  void *callback_cls;
 
  struct GNUNET_HashCode keyhash;
 
  const struct GNUNET_PeerIdentity *peer;
 
  struct GNUNET_PeerIdentity peer_buf;
 
  char *key;
 
  char *sub_system;
 
  uint32_t rid;
 
  const struct GNUNET_CONFIGURATION_Handle *cfg;
 
  GNUNET_SCHEDULER_TaskCallback sync_cb;
 
  void *sync_cb_cls;
 
  struct GNUNET_MQ_Handle *mq;
 
  int iterate_first;
};
 
static void
handle_sync (void *cls, const struct GNUNET_MessageHeader *msg)
{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
 
  if (NULL != mc->sync_cb)
    mc->sync_cb (mc->sync_cb_cls);
}
 
 
static int
check_result (void *cls, const struct PeerstoreRecordMessage *msg)
{
  /* we defer validation to #handle_result */
  return GNUNET_OK;
}
 
 
static void
handle_result (void *cls, const struct PeerstoreRecordMessage *msg)
{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
  struct GNUNET_PEERSTORE_Record *record;
 
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Monitor received RecordMessage\n");
  record = PEERSTORE_parse_record_message (msg);
  if (NULL == record)
  {
    mc->callback (mc->callback_cls,
                  NULL,
                  _ ("Received a malformed response from service."));
  }
  else
  {
    mc->callback (mc->callback_cls, record, NULL);
    PEERSTORE_destroy_record (record);
  }
}
 
 
static void reconnect (struct GNUNET_PEERSTORE_Monitor *mc);
 
static void
mq_error_handler (void *cls, enum GNUNET_MQ_Error err)
{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
 
  reconnect (mc);
}
 
 
static void
reconnect (struct GNUNET_PEERSTORE_Monitor *mc)
{
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_fixed_size (sync,
                             GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_SYNC,
                             struct GNUNET_MessageHeader,
                             mc),
    GNUNET_MQ_hd_var_size (result,
                           GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD,
                           struct PeerstoreRecordMessage, mc),
    GNUNET_MQ_handler_end ()
  };
  struct GNUNET_MQ_Envelope *env;
  struct PeerstoreMonitorStartMessage *sm;
  size_t key_len = 0;
  size_t ss_size = 0;
 
  if (NULL != mc->mq)
  {
    GNUNET_MQ_destroy (mc->mq);
    /* MUST clear @e mq before handing control to the callback: it is
       dangling now, and an error callback that reacts by calling
       #GNUNET_PEERSTORE_monitor_next() would send on freed memory. */
    mc->mq = NULL;
    if (NULL != mc->error_cb)
      mc->error_cb (mc->error_cb_cls);
  }
  mc->mq = GNUNET_CLIENT_connect (mc->cfg,
                                  "peerstore",
                                  handlers,
                                  &mq_error_handler,
                                  mc);
  if (NULL == mc->mq)
    return;
  if (NULL != mc->key)
    key_len = strlen (mc->key) + 1;
  if (NULL != mc->sub_system)
    ss_size = strlen (mc->sub_system) + 1;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending MONITOR_START\n");
  env = GNUNET_MQ_msg_extra (sm,
                             htons (key_len) + htons (ss_size),
                             GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_START);
  sm->iterate_first = htons (mc->iterate_first);
  if (NULL != mc->peer)
  {
    sm->peer = *mc->peer;
    sm->peer_set = htons (GNUNET_YES);
  }
  if (NULL != mc->sub_system)
    GNUNET_memcpy (&sm[1], mc->sub_system, ss_size);
  sm->sub_system_size = htons (ss_size);
  if (NULL != mc->key)
    GNUNET_memcpy (((char*) &sm[1]) + ss_size, mc->key, key_len);
  sm->key_size = htons (key_len);
  GNUNET_MQ_send (mc->mq, env);
}
 
 
struct GNUNET_PEERSTORE_Monitor *
GNUNET_PEERSTORE_monitor_start (
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  int iterate_first,
  const char *sub_system,
  const struct GNUNET_PeerIdentity *peer,
  const char *key,
  GNUNET_SCHEDULER_TaskCallback error_cb,
  void *error_cb_cls,
  GNUNET_SCHEDULER_TaskCallback sync_cb,
  void *sync_cb_cls,
  GNUNET_PEERSTORE_Processor callback,
  void *callback_cls)
{
  struct GNUNET_PEERSTORE_Monitor *mc;
 
  mc = GNUNET_new (struct GNUNET_PEERSTORE_Monitor);
  mc->callback = callback;
  mc->callback_cls = callback_cls;
  mc->sync_cb = sync_cb;
  mc->sync_cb_cls = sync_cb_cls;
  mc->error_cb = error_cb;
  mc->error_cb_cls = error_cb_cls;
  if (NULL != key)
    mc->key = GNUNET_strdup (key);
  if (NULL != peer)
  {
    mc->peer_buf = *peer;
    mc->peer = &mc->peer_buf;
  }
  mc->iterate_first = iterate_first;
  mc->sub_system = GNUNET_strdup (sub_system);
  mc->cfg = cfg;
  reconnect (mc);
  if (NULL == mc->mq)
  {
    GNUNET_free (mc->sub_system);
    GNUNET_free (mc->key);
    GNUNET_free (mc);
    return NULL;
  }
  return mc;
}
 
 
void
GNUNET_PEERSTORE_monitor_stop (struct GNUNET_PEERSTORE_Monitor *zm)
{
  if (NULL != zm->mq)
  {
    GNUNET_MQ_destroy (zm->mq);
    zm->mq = NULL;
  }
  GNUNET_free (zm->sub_system);
  GNUNET_free (zm->key);
  GNUNET_free (zm);
}
 
 
void
GNUNET_PEERSTORE_monitor_next (struct GNUNET_PEERSTORE_Monitor *zm,
                               uint64_t limit)
{
  struct GNUNET_MQ_Envelope *env;
  struct PeerstoreMonitorNextMessage *nm;
 
  if (NULL == zm->mq)
  {
    /* Disconnected; #reconnect() re-sends MONITOR_START and the service
       starts over, so there is nothing to ask for right now. */
    return;
  }
  env = GNUNET_MQ_msg (nm, GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_NEXT);
  nm->limit = GNUNET_htonll (limit);
  GNUNET_MQ_send (zm->mq, env);
}

Function Documentation

handle_sync()

static void handle_sync ( void *  cls, const struct GNUNET_MessageHeader *  msg  )

static

Definition at line 130 of file peerstore_api_monitor.c.

{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
 
  if (NULL != mc->sync_cb)
    mc->sync_cb (mc->sync_cb_cls);
}

References mc, and GNUNET_PEERSTORE_Monitor::sync_cb.

check_result()

static int check_result ( void *  cls, const struct PeerstoreRecordMessage *  msg  )

static

When a response for iterate request is received, check the message is well-formed.

Parameters

cls	a struct GNUNET_PEERSTORE_Handle *
msg	message received

Definition at line 147 of file peerstore_api_monitor.c.

{
  /* we defer validation to #handle_result */
  return GNUNET_OK;
}

References GNUNET_OK.

handle_result()

static void handle_result ( void *  cls, const struct PeerstoreRecordMessage *  msg  )

static

When a response to monitor is received.

Parameters

cls	a struct GNUNET_PEERSTORE_Handle *
msg	message received

Definition at line 161 of file peerstore_api_monitor.c.

{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
  struct GNUNET_PEERSTORE_Record *record;
 
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Monitor received RecordMessage\n");
  record = PEERSTORE_parse_record_message (msg);
  if (NULL == record)
  {
    mc->callback (mc->callback_cls,
                  NULL,
                  _ ("Received a malformed response from service."));
  }
  else
  {
    mc->callback (mc->callback_cls, record, NULL);
    PEERSTORE_destroy_record (record);
  }
}

References _, GNUNET_ERROR_TYPE_DEBUG, LOG, mc, msg, PEERSTORE_destroy_record(), PEERSTORE_parse_record_message(), and record().

Here is the call graph for this function:

reconnect()

static void reconnect ( struct GNUNET_PEERSTORE_Monitor *  mc )

static

Definition at line 194 of file peerstore_api_monitor.c.

{
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_fixed_size (sync,
                             GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_SYNC,
                             struct GNUNET_MessageHeader,
                             mc),
    GNUNET_MQ_hd_var_size (result,
                           GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD,
                           struct PeerstoreRecordMessage, mc),
    GNUNET_MQ_handler_end ()
  };
  struct GNUNET_MQ_Envelope *env;
  struct PeerstoreMonitorStartMessage *sm;
  size_t key_len = 0;
  size_t ss_size = 0;
 
  if (NULL != mc->mq)
  {
    GNUNET_MQ_destroy (mc->mq);
    /* MUST clear @e mq before handing control to the callback: it is
       dangling now, and an error callback that reacts by calling
       #GNUNET_PEERSTORE_monitor_next() would send on freed memory. */
    mc->mq = NULL;
    if (NULL != mc->error_cb)
      mc->error_cb (mc->error_cb_cls);
  }
  mc->mq = GNUNET_CLIENT_connect (mc->cfg,
                                  "peerstore",
                                  handlers,
                                  &mq_error_handler,
                                  mc);
  if (NULL == mc->mq)
    return;
  if (NULL != mc->key)
    key_len = strlen (mc->key) + 1;
  if (NULL != mc->sub_system)
    ss_size = strlen (mc->sub_system) + 1;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending MONITOR_START\n");
  env = GNUNET_MQ_msg_extra (sm,
                             htons (key_len) + htons (ss_size),
                             GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_START);
  sm->iterate_first = htons (mc->iterate_first);
  if (NULL != mc->peer)
  {
    sm->peer = *mc->peer;
    sm->peer_set = htons (GNUNET_YES);
  }
  if (NULL != mc->sub_system)
    GNUNET_memcpy (&sm[1], mc->sub_system, ss_size);
  sm->sub_system_size = htons (ss_size);
  if (NULL != mc->key)
    GNUNET_memcpy (((char*) &sm[1]) + ss_size, mc->key, key_len);
  sm->key_size = htons (key_len);
  GNUNET_MQ_send (mc->mq, env);
}

References env, GNUNET_CLIENT_connect(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_memcpy, GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_START, GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_SYNC, GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD, GNUNET_MQ_destroy(), GNUNET_MQ_handler_end, GNUNET_MQ_hd_fixed_size, GNUNET_MQ_hd_var_size, GNUNET_MQ_msg_extra, GNUNET_MQ_send(), GNUNET_YES, handlers, PeerstoreMonitorStartMessage::iterate_first, PeerstoreMonitorStartMessage::key_size, mc, mq_error_handler(), PeerstoreMonitorStartMessage::peer, PeerstoreMonitorStartMessage::peer_set, result, and PeerstoreMonitorStartMessage::sub_system_size.

Here is the call graph for this function:

mq_error_handler()

static void mq_error_handler ( void *  cls, enum GNUNET_MQ_Error  err  )

static

Definition at line 185 of file peerstore_api_monitor.c.

{
  struct GNUNET_PEERSTORE_Monitor *mc = cls;
 
  reconnect (mc);
}

References mc, and reconnect().

Referenced by reconnect().

Here is the call graph for this function:

Here is the caller graph for this function: