hop-2012/experiments/cmsg/subscription.c

/* Copyright 2010, 2011, 2012 Tony Garnock-Jones <tonygarnockjones@gmail.com>.
 *
 * This file is part of Hop.
 *
 * Hop is free software: you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hop is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
 * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
 * License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hop.  If not, see <http://www.gnu.org/licenses/>.
 */
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include <stdint.h>
#include <stddef.h>

#include <assert.h>

#include <ucontext.h>

#include "cmsg_private.h"
#include "ref.h"
#include "sexp.h"
#include "hashtable.h"
#include "node.h"
#include "meta.h"
#include "messages.h"
#include "subscription.h"

void free_subscription(subscription_t *sub) {
  DECREF(sub->uuid, sexp_destructor);
  DECREF(sub->filter, sexp_destructor);
  DECREF(sub->sink, sexp_destructor);
  DECREF(sub->name, sexp_destructor);
  free(sub);
}

void free_subscription_chain(subscription_t *chain) {
  while (chain != NULL) {
    subscription_t *next = chain->link;
    free_subscription(chain);
    chain = next;
  }
}

/* Returns true if the subscription has not been unsubscribed and the
   destination of the subscription exists. */
int send_to_subscription(sexp_t *source,
			 hashtable_t *subscriptions,
			 subscription_t *sub,
			 sexp_t *body)
{
  if (sub->uuid == NULL) {
    free_subscription(sub);
    return 0;
  } else if (!post_node(sexp_data(sub->sink), sexp_data(sub->name), body, sub->uuid)) {
    announce_subscription(source, sub->filter, sub->sink, sub->name, 0);
    hashtable_erase(subscriptions, sexp_data(sub->uuid));
    free_subscription(sub);
    return 0;
  } else {
    return 1;
  }
}

subscription_t *send_to_subscription_chain(sexp_t *source,
					   hashtable_t *subscriptions,
					   subscription_t *chain,
					   sexp_t *body)
{
  subscription_t *top = chain;
  subscription_t *prev = NULL;
  while (chain != NULL) {
    subscription_t *next = chain->link;
    if (!send_to_subscription(source, subscriptions, chain, body)) {
      if (prev == NULL) {
	top = next;
      } else {
	prev->link = next;
      }
    }
    prev = chain;
    chain = next;
  }
  return top;
}

subscription_t *handle_subscribe_message(sexp_t *source,
					 hashtable_t *subscriptions,
					 parsed_message_t *p)
{
  unsigned char uuid[CMSG_UUID_BUF_SIZE];
  if (gen_uuid(uuid) != 0) {
    warn("Could not generate UUID\n");
    return NULL;
  } else {
    subscription_t *sub = malloc(sizeof(*sub));

    sub->uuid = INCREF(sexp_bytes(CMSG_BYTES(sizeof(uuid), uuid)));
    sub->filter = p->subscribe.filter;
    sub->sink = p->subscribe.sink;
    sub->name = p->subscribe.name;
    sub->link = NULL;

    if (!sexp_stringp(sub->filter) || !sexp_stringp(sub->sink) || !sexp_stringp(sub->name)
	|| !sexp_stringp(p->subscribe.reply_sink) || !sexp_stringp(p->subscribe.reply_name)) {
      DECREF(sub->uuid, sexp_destructor);
      free(sub);
      warn("Bad sink/name/reply_sink/reply_name in subscribe");
      return NULL;
    }

    INCREF(sub->filter);
    INCREF(sub->sink);
    INCREF(sub->name);

    hashtable_put(subscriptions, sexp_data(sub->uuid), sub);

    announce_subscription(source, sub->filter, sub->sink, sub->name, 1);

    post_node(sexp_data(p->subscribe.reply_sink),
	      sexp_data(p->subscribe.reply_name),
	      message_subscribe_ok(sub->uuid),
	      sexp_empty_bytes);

    return sub;
  }
}

void handle_unsubscribe_message(sexp_t *source,
				hashtable_t *subscriptions,
				parsed_message_t *p)
{
  cmsg_bytes_t uuid;
  subscription_t *sub;

  if (!sexp_stringp(p->unsubscribe.token)) {
    warn("Invalid unsubscription\n");
    return;
  }

  uuid = sexp_data(p->unsubscribe.token);
  if (hashtable_get(subscriptions, uuid, (void **) &sub)) {
    /* TODO: clean up more eagerly perhaps? */
    announce_subscription(source, sub->filter, sub->sink, sub->name, 0);
    DECREF(sub->uuid, sexp_destructor);
    sub->uuid = NULL;
    hashtable_erase(subscriptions, uuid);
  }
}