hop-2012/experiments/cmsg/relay.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 <sys/types.h>
#include <sys/uio.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <sys/wait.h>
#include <time.h>
#include <sys/time.h>

#include <assert.h>

typedef unsigned char u_char;
#include <event.h>

#include "cmsg_private.h"
#include "harness.h"
#include "relay.h"
#include "net.h"
#include "ref.h"
#include "sexp.h"
#include "sexpio.h"
#include "hashtable.h"
#include "node.h"
#include "messages.h"

#define WANT_MESSAGE_TRACE 0

typedef struct relay_extension_t_ {
  struct sockaddr_in peername;
  char peername_str[256];
  sexp_t *remote_container_name;
  int fd;
  IOHandle *outh;
} relay_extension_t;

static long connection_count = 0;

static void stats_printer(void *arg) {
  while (1) {
    info("%ld connections active\n", connection_count);
    nap(1000);
  }
}

void init_relay(void) {
  spawn(stats_printer, NULL);
}

static sexp_t *relay_extend(node_t *n, sexp_t *args) {
  /* TODO: outbound connections; args==NULL -> server relay, nonNULL -> outbound. */
  n->extension = calloc(1, sizeof(relay_extension_t));
  return NULL;
}

static void relay_destructor(node_t *n) {
  relay_extension_t *r = n->extension;
  delete_iohandle(r->outh);
  r->outh = NULL;
  if (close(r->fd) == -1) {
    /* log errors as warnings here and keep on trucking */
    warn("Closing file descriptor %d produced errno %d: %s\n",
	 r->fd, errno, strerror(errno));
  }
  DECREF(r->remote_container_name, sexp_destructor);
  free(r);
}

static void relay_handle_message(node_t *n, sexp_t *m) {
  relay_extension_t *r = n->extension;

#if WANT_MESSAGE_TRACE
  info("fd %d <-- ", r->fd);
  sexp_writeln(stderr_h, m);
#endif

  BCHECK(!sexp_write(r->outh, m), "relay_handle_message sexp_write");
}

static node_class_t relay_class = {
  .name = "relay",
  .extend = relay_extend,
  .destroy = relay_destructor,
  .handle_message = relay_handle_message
};

static void send_error(IOHandle *h, char const *message, sexp_t *details) {
  sexp_t *m = message_error(sexp_cstring(message), details);
  INCREF(m);
  warn("Sending error: ");
  sexp_writeln(stderr_h, m);
  iohandle_clear_error(h);
  BCHECK(!sexp_write(h, m), "send_error sexp_write");
  DECREF(m, sexp_destructor);
  iohandle_flush(h); /* ignore result here, there's not much we can do with it */
}

static void send_sexp_syntax_error(IOHandle *h, char const *message) {
  char const *url = "http://people.csail.mit.edu/rivest/Sexp.txt";
  send_error(h, message, sexp_cstring(url));
}

static void relay_main(node_t *n) {
  relay_extension_t *r = n->extension;
  IOHandle *inh = new_iohandle(r->fd);
  sexp_t *message = NULL; /* held */
  parsed_message_t p;

  INCREF(n); /* because the caller doesn't hold a ref, and we need to
		drop ours on our death */

  info("Accepted connection from %s on fd %d\n", r->peername_str, r->fd);
  connection_count++;

  iohandle_write(r->outh, cmsg_cstring_bytes("(3:hop1:0)"));
  ICHECK(iohandle_flush(r->outh), "iohandle_flush greeting");

  {
    sexp_t *s = message_subscribe(sexp_bytes(local_container_name()),
				  sexp_empty_bytes, sexp_empty_bytes,
				  sexp_empty_bytes, sexp_empty_bytes);
    INCREF(s);
    sexp_write(r->outh, s);
    DECREF(s, sexp_destructor);
  }

  //iohandle_settimeout(r->inh, 3, 0);

  while (1) {
    DECREF(message, sexp_destructor);
    message = NULL;
    if (!sexp_read(inh, &message)) goto network_error;
    INCREF(message);

#if WANT_MESSAGE_TRACE
    info("fd %d --> ", r->fd);
    sexp_writeln(stderr_h, message);
#endif

    if (parse_post(message, &p) && sexp_stringp(p.post.name)) {
      cmsg_bytes_t nodename = sexp_data(p.post.name);
      if (!send_node(nodename, p.post.body)) {
	warn("Was asked to post to unknown node <<%.*s>>\n", nodename.len, nodename.bytes);
      }
    } else if (parse_subscribe(message, &p)
	       && sexp_stringp(p.subscribe.filter)
	       && sexp_stringp(p.subscribe.reply_sink)
	       && sexp_stringp(p.subscribe.reply_name)) {
      if (bind_node(sexp_data(p.subscribe.filter), n)) {
	post_node(sexp_data(p.subscribe.reply_sink),
		  sexp_data(p.subscribe.reply_name),
		  message_subscribe_ok(p.subscribe.filter),
		  sexp_empty_bytes);

	DECREF(r->remote_container_name, sexp_destructor);
	r->remote_container_name = INCREF(p.subscribe.filter);
      } else {
	cmsg_bytes_t filter = sexp_data(p.subscribe.filter);
	warn("Bind failed <<%.*s>>\n", filter.len, filter.bytes);
      }
    } else if (parse_unsubscribe(message, &p)
	       && sexp_stringp(p.unsubscribe.token)) {
      cmsg_bytes_t id = sexp_data(p.unsubscribe.token);
      if (!unbind_node(id)) {
	warn("Unbind failed <<%.*s>>\n", id.len, id.bytes);
      }
    } else {
      send_error(r->outh, "message not understood", message);
      goto protocol_error;
    }
  }

 network_error:
  if (inh->eof) {
    info("Disconnecting fd %d normally.\n", r->fd);
  } else {
    switch (inh->error_kind) {
      case SEXP_ERROR_OVERFLOW:
	send_sexp_syntax_error(r->outh, "sexp too big");
	break;

      case SEXP_ERROR_SYNTAX:
	send_sexp_syntax_error(r->outh, "sexp syntax error");
	break;

      default:
	warn("Relay handle error 0x%04X on fd %d: %d, %s\n",
	     inh->error_kind, r->fd, inh->error_errno, strerror(inh->error_errno));
	break;
    }
  }

 protocol_error:
  DECREF(message, sexp_destructor);
  delete_iohandle(inh);
  unbind_all_names_for_node(n);
  DECREF(n, node_destructor);

  connection_count--;
}

void start_relay(struct sockaddr_in const *peername, int fd) {
  node_t *n = new_node(&relay_class, NULL, NULL);
  relay_extension_t *r = n->extension;
  r->peername = *peername;
  endpoint_name(&r->peername, CMSG_BYTES(sizeof(r->peername_str), r->peername_str));
  r->remote_container_name = NULL;
  r->fd = fd;
  r->outh = new_iohandle(r->fd);
  spawn((process_main_t) relay_main, n);
}