refactored some duplicate code into connection_edge.c

svn:r230

7 files changed, 303 insertions, 508 deletions

diff --git a/src/or/Makefile.am b/src/or/Makefile.am
index 9b7c5a38a..9f25c91c4 100644
--- a/src/or/Makefile.am
+++ b/src/or/Makefile.am
@@ -6,14 +6,14 @@ bin_PROGRAMS = or
 
 or_SOURCES = buffers.c circuit.c command.c connection.c \
              connection_exit.c connection_ap.c connection_op.c connection_or.c config.c \
-             onion.c routers.c directory.c dns.c \
+             onion.c routers.c directory.c dns.c connection_edge.c \
              main.c tor_main.c
 
 or_LDADD = -L../common -lor -lz
 
 test_SOURCES = buffers.c circuit.c command.c connection.c \
              connection_exit.c connection_ap.c connection_op.c connection_or.c config.c \
-             onion.c routers.c directory.c dns.c \
+             onion.c routers.c directory.c dns.c connection_edge.c \
              main.c test.c
 
 test_LDADD = -L../common -lor -lz
diff --git a/src/or/circuit.c b/src/or/circuit.c
index 4de154a93..593c5bddf 100644
--- a/src/or/circuit.c
+++ b/src/or/circuit.c
@@ -329,11 +329,11 @@ int circuit_deliver_data_cell(cell_t *cell, circuit_t *circ, int cell_direction)
 
   if(cell_direction == CELL_DIRECTION_OUT && (!conn || conn->type == CONN_TYPE_EXIT)) {
     log(LOG_DEBUG,"circuit_deliver_data_cell(): Sending to exit.");
-    return connection_exit_process_data_cell(cell, circ);
+    return connection_edge_process_data_cell(cell, circ, EDGE_EXIT);
   }
   if(cell_direction == CELL_DIRECTION_IN && (!conn || conn->type == CONN_TYPE_AP)) {
     log(LOG_DEBUG,"circuit_deliver_data_cell(): Sending to AP.");
-    return connection_ap_process_data_cell(cell, circ);
+    return connection_edge_process_data_cell(cell, circ, EDGE_AP);
   }
   /* else send it as a cell */
   assert(conn);
@@ -536,8 +536,6 @@ void circuit_about_to_close_connection(connection_t *conn) {
    */
   circuit_t *circ;
   connection_t *prevconn, *tmpconn;
-  cell_t cell;
-  int edge_type;
 
   if(!connection_speaks_cells(conn)) {
     /* it's an edge conn. need to remove it from the linked list of
@@ -549,48 +547,29 @@ void circuit_about_to_close_connection(connection_t *conn) {
     if(!circ)
       return;
 
-    memset(&cell, 0, sizeof(cell_t));
-    cell.command = CELL_DATA;
-    cell.length = TOPIC_HEADER_SIZE;
-    *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
-    *cell.payload = TOPIC_COMMAND_END;
-
     if(conn == circ->p_conn) {
       circ->p_conn = conn->next_topic;
-      edge_type = EDGE_AP;
       goto send_end;
     }
     if(conn == circ->n_conn) {
       circ->n_conn = conn->next_topic;
-      edge_type = EDGE_EXIT;
       goto send_end;
     }
     for(prevconn = circ->p_conn; prevconn->next_topic && prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
     if(prevconn->next_topic) {
       prevconn->next_topic = conn->next_topic;
-      edge_type = EDGE_AP;
       goto send_end;
     }
     for(prevconn = circ->n_conn; prevconn->next_topic && prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
     if(prevconn->next_topic) {
       prevconn->next_topic = conn->next_topic;
-      edge_type = EDGE_EXIT;
       goto send_end;
     }
     log(LOG_ERR,"circuit_about_to_close_connection(): edge conn not in circuit's list?");
     assert(0); /* should never get here */
 send_end:
-    if(edge_type == EDGE_AP) { /* send to circ->n_conn */
-      log(LOG_INFO,"circuit_about_to_close_connection(): send data end forward (aci %d).",circ->n_aci);
-      cell.aci = circ->n_aci;
-    } else { /* send to circ->p_conn */
-      assert(edge_type == EDGE_EXIT);
-      log(LOG_INFO,"circuit_about_to_close_connection(): send data end backward (aci %d).",circ->p_aci);
-      cell.aci = circ->p_aci;
-    }
-
-    if(circuit_deliver_data_cell_from_edge(&cell, circ, edge_type) < 0) {
-      log(LOG_DEBUG,"circuit_about_to_close_connection(): circuit_deliver_data_cell_from_edge (%d) failed. Closing.", edge_type);
+    if(connection_edge_send_command(conn, circ, TOPIC_COMMAND_END) < 0) {
+      log(LOG_DEBUG,"circuit_about_to_close_connection(): sending end failed. Closing.");
       circuit_close(circ);
     }
     return;
diff --git a/src/or/connection.c b/src/or/connection.c
index 3a074bce5..576e4962f 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -44,9 +44,12 @@ char *conn_state_to_string[][15] = {
     "connecting",                      /* 1 */
     "open" },                          /* 2 */
   { "ready" }, /* app listener, 0 */
-  { "awaiting dest info",         /* app, 0 */
-    "waiting for OR connection",       /* 1 */
-    "open" },                          /* 2 */
+  { "", /* 0 */
+    "", /* 1 */
+    "", /* 2 */
+    "awaiting dest info",         /* app, 3 */
+    "waiting for OR connection",       /* 4 */
+    "open" },                          /* 5 */
   { "ready" }, /* dir listener, 0 */
   { "connecting",                      /* 0 */
     "sending command",                 /* 1 */
@@ -681,9 +684,8 @@ int connection_process_inbuf(connection_t *conn) {
     case CONN_TYPE_OR:
       return connection_or_process_inbuf(conn);
     case CONN_TYPE_EXIT:
-      return connection_exit_process_inbuf(conn);
     case CONN_TYPE_AP:
-      return connection_ap_process_inbuf(conn);
+      return connection_edge_process_inbuf(conn);
     case CONN_TYPE_DIR:
       return connection_dir_process_inbuf(conn);
     case CONN_TYPE_DNSMASTER:
@@ -838,14 +840,13 @@ int connection_finished_flushing(connection_t *conn) {
 //  log(LOG_DEBUG,"connection_finished_flushing() entered. Socket %u.", conn->s);
 
   switch(conn->type) {
-    case CONN_TYPE_AP:
-      return connection_ap_finished_flushing(conn);
     case CONN_TYPE_OP:
       return connection_op_finished_flushing(conn);
     case CONN_TYPE_OR:
       return connection_or_finished_flushing(conn);
+    case CONN_TYPE_AP:
     case CONN_TYPE_EXIT:
-      return connection_exit_finished_flushing(conn);
+      return connection_edge_finished_flushing(conn);
     case CONN_TYPE_DIR:
       return connection_dir_finished_flushing(conn);
     case CONN_TYPE_DNSMASTER:
diff --git a/src/or/connection_ap.c b/src/or/connection_ap.c
index 4983af6d1..831dd9b16 100644
--- a/src/or/connection_ap.c
+++ b/src/or/connection_ap.c
@@ -6,58 +6,6 @@
 
 extern or_options_t options; /* command-line and config-file options */
 
-int connection_ap_process_inbuf(connection_t *conn) {
-
-  assert(conn && conn->type == CONN_TYPE_AP);
-
-  if(conn->inbuf_reached_eof) {
-#ifdef HALF_OPEN
-    /* eof reached; we're done reading, but we might want to write more. */ 
-    conn->done_receiving = 1;
-    shutdown(conn->s, 0); /* XXX check return, refactor NM */
-    if (conn->done_sending)
-      conn->marked_for_close = 1;
-
-    /* XXX Factor out common logic here and in circuit_about_to_close NM */
-    circ = circuit_get_by_conn(conn);
-    if (!circ)
-      return -1;
-    
-    memset(&cell, 0, sizeof(cell_t));
-    cell.command = CELL_DATA;
-    cell.length = TOPIC_HEADER_SIZE;
-    *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
-    *cell.payload = TOPIC_COMMAND_END;
-    cell.aci = circ->n_aci;
-    if (circuit_deliver_data_cell_from_edge(&cell, circ, EDGE_AP) < 0) {
-      log(LOG_DEBUG,"connection_ap_process_inbuf: circuit_deliver_data_cell_from_edge failed.  Closing");
-      circuit_close(circ);
-    }
-    return 0;
-#else 
-    /* eof reached, kill it. */
-    log(LOG_DEBUG,"connection_ap_process_inbuf(): conn reached eof. Closing.");
-    return -1;
-#endif
-  }
-
-//  log(LOG_DEBUG,"connection_ap_process_inbuf(): state %d.",conn->state);
-
-  switch(conn->state) {
-    case AP_CONN_STATE_SOCKS_WAIT:
-      return ap_handshake_process_socks(conn);
-    case AP_CONN_STATE_OPEN:
-      if(connection_package_raw_inbuf(conn) < 0)
-        return -1;
-      circuit_consider_stop_edge_reading(circuit_get_by_conn(conn), EDGE_AP);
-      return 0;
-    default:
-      log(LOG_DEBUG,"connection_ap_process_inbuf() called in state where I'm waiting. Ignoring buf for now.");
-  }
-
-  return 0;
-}
-
 int ap_handshake_process_socks(connection_t *conn) {
   char c;
   socks4_t socks4_info; 
@@ -397,139 +345,6 @@ int ap_handshake_socks_reply(connection_t *conn, char result) {
   return connection_flush_buf(conn); /* try to flush it, in case we're about to close the conn */
 }
 
-int connection_ap_process_data_cell(cell_t *cell, circuit_t *circ) {
-  connection_t *conn;
-  int topic_command;
-  int topic_id;
-  static int num_seen=0;
-
-  /* an incoming data cell has arrived */
-
-  assert(cell && circ);
-
-  topic_command = *cell->payload;
-  topic_id = ntohs(*(uint16_t *)(cell->payload+2));
-  log(LOG_DEBUG,"connection_ap_process_data_cell(): command %d topic %d", topic_command, topic_id);
-  num_seen++;
-  log(LOG_DEBUG,"connection_ap_process_data_cell(): Now seen %d data cells here.", num_seen);
-
-  circuit_consider_sending_sendme(circ, EDGE_AP);
-
-  for(conn = circ->p_conn; conn && conn->topic_id != topic_id; conn = conn->next_topic) ;
-
-  /* now conn is either NULL, in which case we don't recognize the topic_id, or
-   * it is set, in which case cell is talking about this conn.
-   */
-
-  if(conn && conn->state != AP_CONN_STATE_OPEN) {
-    /* we should not have gotten this cell */
-    log(LOG_DEBUG,"connection_ap_process_data_cell(): Got a data cell when not in 'open' state. Dropping.");
-    return 0;
-  }
-
-  switch(topic_command) {
-    case TOPIC_COMMAND_BEGIN:
-      log(LOG_INFO,"connection_ap_process_data_cell(): topic begin request unsupported. Dropping.");
-      break;
-    case TOPIC_COMMAND_DATA:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_ap_process_data_cell(): data cell dropped, unknown topic %d.",topic_id);
-        return 0;
-      }
-      if(--conn->n_receive_topicwindow < 0) { /* is it below 0 after decrement? */
-        log(LOG_DEBUG,"connection_ap_process_data_cell(): receive_topicwindow at exit below 0. Killing.");
-        return -1; /* exit node breaking protocol. kill the whole circuit. */
-      }
-      log(LOG_DEBUG,"connection_ap_process_data_cell(): willing to receive %d more cells from circ",conn->n_receive_topicwindow);
-
-#ifdef USE_ZLIB
-      if(connection_decompress_to_buf(cell->payload + TOPIC_HEADER_SIZE,
-                                      cell->length - TOPIC_HEADER_SIZE, 
-                                      conn, Z_SYNC_FLUSH) < 0) {
-        log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
-        conn->marked_for_close = 1;
-        return 0;
-      }
-#else
-      if(connection_write_to_buf(cell->payload + TOPIC_HEADER_SIZE,
-                                 cell->length - TOPIC_HEADER_SIZE, conn) < 0) {
-        conn->marked_for_close = 1;
-        return 0;
-      }
-#endif
-      if(connection_consider_sending_sendme(conn, EDGE_AP) < 0)
-        conn->marked_for_close = 1;
-      return 0;
-    case TOPIC_COMMAND_END:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_ap_process_data_cell(): end cell dropped, unknown topic %d.",topic_id);
-        return 0;
-      }
-      log(LOG_DEBUG,"connection_ap_process_data_cell(): end cell for topic %d. Removing topic.",topic_id);
-
-      /* go through and identify who points to conn. remove conn from the list. */
-#if 0
-      if(conn == circ->p_conn) {
-        circ->p_conn = conn->next_topic;
-      }
-      for(prevconn = circ->p_conn; prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
-      prevconn->next_topic = conn->next_topic;
-#endif
-#ifdef HALF_OPEN
-      conn->done_sending = 1;
-      shutdown(conn->s, 1); /* XXX check return; refactor NM */
-      if (conn->done_receiving)
-        conn->marked_for_close = 1;
-#endif
-      conn->marked_for_close = 1;
-      break;
-    case TOPIC_COMMAND_CONNECTED:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_ap_process_data_cell(): connected cell dropped, unknown topic %d.",topic_id);
-        break;
-      }
-      log(LOG_DEBUG,"connection_ap_process_data_cell(): Connected! Notifying application.");
-      if(ap_handshake_socks_reply(conn, SOCKS4_REQUEST_GRANTED) < 0) {
-        conn->marked_for_close = 1;
-      }
-      break;
-    case TOPIC_COMMAND_SENDME:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_ap_process_data_cell(): sendme cell dropped, unknown topic %d.",topic_id);
-        return 0;
-      }
-      conn->p_receive_topicwindow += TOPICWINDOW_INCREMENT;
-      connection_start_reading(conn);
-      connection_package_raw_inbuf(conn); /* handle whatever might still be on the inbuf */
-      circuit_consider_stop_edge_reading(circ, EDGE_AP);
-      break;
-    default:
-      log(LOG_DEBUG,"connection_ap_process_data_cell(): unknown topic command %d.",topic_command);
-  }
-  return 0;
-}
-
-int connection_ap_finished_flushing(connection_t *conn) {
-
-  assert(conn && conn->type == CONN_TYPE_AP);
-
-  switch(conn->state) {
-    case AP_CONN_STATE_OPEN:
-      /* FIXME down the road, we'll clear out circuits that are pending to close */
-      connection_stop_writing(conn);
-#ifdef USE_ZLIB
-      if (connection_decompress_to_buf(NULL, 0, conn, Z_SYNC_FLUSH) < 0)
-        return 0;
-#endif
-      return connection_consider_sending_sendme(conn, EDGE_AP);
-    default:
-      log(LOG_DEBUG,"Bug: connection_ap_finished_flushing() called in unexpected state.");
-      return 0;
-  }
-
-  return 0;
-}
-
 int connection_ap_create_listener(struct sockaddr_in *bindaddr) {
   log(LOG_DEBUG,"connection_create_ap_listener starting");
   return connection_create_listener(bindaddr, CONN_TYPE_AP_LISTENER);
diff --git a/src/or/connection_edge.c b/src/or/connection_edge.c
new file mode 100644
index 000000000..acae44e61
--- /dev/null
+++ b/src/or/connection_edge.c
@@ -0,0 +1,274 @@
+/* Copyright 2001,2002 Roger Dingledine, Matej Pfajfar. */
+/* See LICENSE for licensing information */
+/* $Id$ */
+
+#include "or.h"
+
+extern or_options_t options; /* command-line and config-file options */
+
+int connection_edge_process_inbuf(connection_t *conn) {
+
+  assert(conn);
+  assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);
+
+  if(conn->inbuf_reached_eof) {
+#ifdef HALF_OPEN
+    /* eof reached; we're done reading, but we might want to write more. */ 
+    conn->done_receiving = 1;
+    shutdown(conn->s, 0); /* XXX check return, refactor NM */
+    if (conn->done_sending)
+      conn->marked_for_close = 1;
+
+    /* XXX Factor out common logic here and in circuit_about_to_close NM */
+    circ = circuit_get_by_conn(conn);
+    if (!circ)
+      return -1;
+    
+    memset(&cell, 0, sizeof(cell_t));
+    cell.command = CELL_DATA;
+    cell.length = TOPIC_HEADER_SIZE;
+    *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
+    *cell.payload = TOPIC_COMMAND_END;
+    cell.aci = circ->n_aci;
+    if (circuit_deliver_data_cell_from_edge(&cell, circ, conn->type) < 0) {
+      log(LOG_DEBUG,"connection_edge_process_inbuf: circuit_deliver_data_cell_from_edge failed.  Closing");
+      circuit_close(circ);
+    }
+    return 0;
+#else 
+    /* eof reached, kill it. */
+    log(LOG_DEBUG,"connection_edge_process_inbuf(): conn reached eof. Closing.");
+    return -1;
+#endif
+  }
+
+  switch(conn->state) {
+    case AP_CONN_STATE_SOCKS_WAIT:
+      return ap_handshake_process_socks(conn);
+    case AP_CONN_STATE_OPEN:
+    case EXIT_CONN_STATE_OPEN:
+      if(connection_package_raw_inbuf(conn) < 0)
+        return -1;
+      circuit_consider_stop_edge_reading(circuit_get_by_conn(conn), EDGE_AP);
+      return 0;
+    case EXIT_CONN_STATE_CONNECTING:
+      log(LOG_DEBUG,"connection_edge_process_inbuf(): text from server while in 'connecting' state at exit. Leaving it on buffer.");
+      return 0;
+  }
+
+  return 0;
+}
+
+int connection_edge_send_command(connection_t *conn, circuit_t *circ, int topic_command) {
+  cell_t cell;
+
+  assert(conn);
+
+  if(!circ) {
+    log(LOG_DEBUG,"connection_edge_send_command(): conn has no circ. Closing.");
+    return -1;
+  }
+
+  memset(&cell, 0, sizeof(cell_t));
+  if(conn->type == CONN_TYPE_AP)
+    cell.aci = circ->n_aci;
+  else
+    cell.aci = circ->p_aci;
+  cell.command = CELL_DATA;
+  *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
+  *cell.payload = topic_command;
+  cell.length = TOPIC_HEADER_SIZE;
+  log(LOG_INFO,"connection_edge_send_command(): delivering %d cell %s.", topic_command, conn->type == CONN_TYPE_AP ? "forward" : "backward");
+
+  if(circuit_deliver_data_cell_from_edge(&cell, circ, conn->type) < 0) {
+    log(LOG_DEBUG,"connection_edge_send_command(): circuit_deliver_data_cell failed. Closing.");
+    circuit_close(circ);
+    return 0;
+  }
+  return 0;
+}
+
+int connection_edge_process_data_cell(cell_t *cell, circuit_t *circ, int edge_type) {
+  connection_t *conn;
+  int topic_command;
+  int topic_id;
+  static int num_seen=0;
+
+  /* an incoming data cell has arrived */
+
+  assert(cell && circ);
+
+  topic_command = *cell->payload;
+  topic_id = ntohs(*(uint16_t *)(cell->payload+2));
+  log(LOG_DEBUG,"connection_edge_process_data_cell(): command %d topic %d", topic_command, topic_id);
+  num_seen++;
+  log(LOG_DEBUG,"connection_edge_process_data_cell(): Now seen %d data cells here.", num_seen);
+
+  circuit_consider_sending_sendme(circ, edge_type);
+
+  for(conn = circ->p_conn; conn && conn->topic_id != topic_id; conn = conn->next_topic) ;
+
+  /* now conn is either NULL, in which case we don't recognize the topic_id, or
+   * it is set, in which case cell is talking about this conn.
+   */
+
+  if(conn && conn->state != AP_CONN_STATE_OPEN && conn->state != EXIT_CONN_STATE_OPEN) {
+    if(conn->type == CONN_TYPE_EXIT && topic_command == TOPIC_COMMAND_END) {
+      log(LOG_INFO,"connection_edge_process_data_cell(): Exit got end before we're connected. Marking for close.");
+      conn->marked_for_close = 1;
+    } else {
+      log(LOG_DEBUG,"connection_edge_process_data_cell(): Got an unexpected data cell, not in 'open' state. Dropping.");
+    }
+    return 0;
+  }
+
+  switch(topic_command) {
+    case TOPIC_COMMAND_BEGIN:
+      if(edge_type == EDGE_AP) {
+        log(LOG_INFO,"connection_edge_process_data_cell(): topic begin request unsupported. Dropping.");
+        return 0;
+      } else {
+        if(conn) {
+          log(LOG_INFO,"connection_edge_process_data_cell(): begin cell for known topic. Dropping.");
+          return 0;
+        }
+        return connection_exit_begin_conn(cell, circ);
+      }
+    case TOPIC_COMMAND_DATA:
+      if(!conn) {
+        log(LOG_DEBUG,"connection_edge_process_data_cell(): data cell dropped, unknown topic %d.",topic_id);
+        return 0;
+      }
+      if((edge_type == EDGE_AP && --conn->n_receive_topicwindow < 0) ||
+         (edge_type == EDGE_EXIT && --conn->p_receive_topicwindow < 0)) { /* is it below 0 after decrement? */
+        log(LOG_DEBUG,"connection_edge_process_data_cell(): receive_topicwindow below 0. Killing.");
+        return -1; /* somebody's breaking protocol. kill the whole circuit. */
+      }
+
+#ifdef USE_ZLIB
+      if(connection_decompress_to_buf(cell->payload + TOPIC_HEADER_SIZE,
+                                      cell->length - TOPIC_HEADER_SIZE, 
+                                      conn, Z_SYNC_FLUSH) < 0) {
+        log(LOG_INFO,"connection_edge_process_data_cell(): write to buf failed. Marking for close.");
+        conn->marked_for_close = 1;
+        return 0;
+      }
+#else
+      if(connection_write_to_buf(cell->payload + TOPIC_HEADER_SIZE,
+                                 cell->length - TOPIC_HEADER_SIZE, conn) < 0) {
+        conn->marked_for_close = 1;
+        return 0;
+      }
+#endif
+      if(connection_consider_sending_sendme(conn, edge_type) < 0)
+        conn->marked_for_close = 1;
+      return 0;
+    case TOPIC_COMMAND_END:
+      if(!conn) {
+        log(LOG_DEBUG,"connection_edge_process_data_cell(): end cell dropped, unknown topic %d.",topic_id);
+        return 0;
+      }
+      log(LOG_DEBUG,"connection_edge_process_data_cell(): end cell for topic %d. Removing topic.",topic_id);
+
+      /* go through and identify who points to conn. remove conn from the list. */
+#if 0
+      if(conn == circ->p_conn) {
+        circ->p_conn = conn->next_topic;
+      }
+      for(prevconn = circ->p_conn; prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
+      prevconn->next_topic = conn->next_topic;
+#endif
+#ifdef HALF_OPEN
+      conn->done_sending = 1;
+      shutdown(conn->s, 1); /* XXX check return; refactor NM */
+      if (conn->done_receiving)
+        conn->marked_for_close = 1;
+#endif
+      conn->marked_for_close = 1;
+      break;
+    case TOPIC_COMMAND_CONNECTED:
+      if(edge_type == EDGE_EXIT) {
+        log(LOG_INFO,"connection_edge_process_data_cell(): 'connected' unsupported at exit. Dropping.");
+        return 0;
+      }
+      if(!conn) {
+        log(LOG_DEBUG,"connection_edge_process_data_cell(): connected cell dropped, unknown topic %d.",topic_id);
+        break;
+      }
+      log(LOG_DEBUG,"connection_edge_process_data_cell(): Connected! Notifying application.");
+      if(ap_handshake_socks_reply(conn, SOCKS4_REQUEST_GRANTED) < 0) {
+        conn->marked_for_close = 1;
+      }
+      break;
+    case TOPIC_COMMAND_SENDME:
+      if(!conn) {
+        log(LOG_DEBUG,"connection_edge_process_data_cell(): sendme cell dropped, unknown topic %d.",topic_id);
+        return 0;
+      }
+      if(edge_type == EDGE_AP)
+        conn->p_receive_topicwindow += TOPICWINDOW_INCREMENT;
+      else
+        conn->n_receive_topicwindow += TOPICWINDOW_INCREMENT;
+      connection_start_reading(conn);
+      connection_package_raw_inbuf(conn); /* handle whatever might still be on the inbuf */
+      circuit_consider_stop_edge_reading(circ, edge_type);
+      break;
+    default:
+      log(LOG_DEBUG,"connection_edge_process_data_cell(): unknown topic command %d.",topic_command);
+  }
+  return 0;
+}
+
+int connection_edge_finished_flushing(connection_t *conn) {
+  int e, len=sizeof(e);
+
+  assert(conn);
+  assert(conn->type == CONN_TYPE_AP || conn->type == CONN_TYPE_EXIT);
+
+  switch(conn->state) {
+    case EXIT_CONN_STATE_CONNECTING:
+      if (getsockopt(conn->s, SOL_SOCKET, SO_ERROR, &e, &len) < 0)  { /* not yet */
+        if(errno != EINPROGRESS){
+          /* yuck. kill it. */
+          log(LOG_DEBUG,"connection_edge_finished_flushing(): in-progress exit connect failed. Removing.");
+          return -1;
+        } else {
+          log(LOG_DEBUG,"connection_edge_finished_flushing(): in-progress exit connect still waiting.");
+          return 0; /* no change, see if next time is better */
+        }
+      }
+      /* the connect has finished. */
+
+      log(LOG_DEBUG,"connection_edge_finished_flushing(): Exit connection to %s:%u established.",
+          conn->address,conn->port);
+
+      conn->state = EXIT_CONN_STATE_OPEN;
+      connection_watch_events(conn, POLLIN); /* stop writing, continue reading */
+      if(connection_wants_to_flush(conn)) /* in case there are any queued data cells */
+        connection_start_writing(conn);
+      return
+        connection_exit_send_connected(conn) || /* deliver a 'connected' data cell back through the circuit. */
+        connection_process_inbuf(conn); /* in case the server has written anything */
+    case AP_CONN_STATE_OPEN:
+    case EXIT_CONN_STATE_OPEN:
+      connection_stop_writing(conn);
+#ifdef USE_ZLIB
+      if (connection_decompress_to_buf(NULL, 0, conn, Z_SYNC_FLUSH) < 0)
+        return 0;
+#endif
+      return connection_consider_sending_sendme(conn, conn->type);
+    default:
+      log(LOG_DEBUG,"Bug: connection_edge_finished_flushing() called in unexpected state.");
+      return 0;
+  }
+
+  return 0;
+}
+
+/*
+  Local Variables:
+  mode:c
+  indent-tabs-mode:nil
+  c-basic-offset:2
+  End:
+*/
diff --git a/src/or/connection_exit.c b/src/or/connection_exit.c
index f2fb7b0d2..8403a33a7 100644
--- a/src/or/connection_exit.c
+++ b/src/or/connection_exit.c
@@ -4,135 +4,6 @@
 
 #include "or.h"
 
-int connection_exit_process_inbuf(connection_t *conn) {
-  circuit_t *circ;
-  cell_t cell;
-
-  assert(conn && conn->type == CONN_TYPE_EXIT);
-
-  if(conn->inbuf_reached_eof) {
-#ifdef HALF_OPEN
-    /* eof reached; we're done reading, but we might want to write more. */ 
-    conn->done_receiving = 1;
-    shutdown(conn->s, 0); /* XXX check return, refactor NM */
-    if (conn->done_sending)
-      conn->marked_for_close = 1;
-
-    /* XXX Factor out common logic here and in circuit_about_to_close NM */
-    circ = circuit_get_by_conn(conn);
-    if (!circ)
-      return -1;
-    
-    memset(&cell, 0, sizeof(cell_t));
-    cell.command = CELL_DATA;
-    cell.length = TOPIC_HEADER_SIZE;
-    *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
-    *cell.payload = TOPIC_COMMAND_END;
-    cell.aci = circ->p_aci;
-    if (circuit_deliver_data_cell_from_edge(&cell, circ, EDGE_EXIT) < 0) {
-      log(LOG_DEBUG,"connection_exit_process_inbuf: circuit_deliver_data_cell_from_edge failed.  Closing");
-      circuit_close(circ);
-    }
-    return 0;
-#else 
-
-    /* eof reached, kill it. */
-    log(LOG_DEBUG,"connection_exit_process_inbuf(): conn reached eof. Closing.");
-    return -1;
-#endif
-  }
-
-  log(LOG_DEBUG,"connection_exit_process_inbuf(): state %d.",conn->state);
-
-  switch(conn->state) {
-    case EXIT_CONN_STATE_CONNECTING:
-      log(LOG_DEBUG,"connection_exit_process_inbuf(): text from server while in 'connecting' state. Leaving it on buffer.");
-      return 0;
-    case EXIT_CONN_STATE_OPEN:
-      if(connection_package_raw_inbuf(conn) < 0)
-        return -1;
-      circuit_consider_stop_edge_reading(circuit_get_by_conn(conn), EDGE_EXIT);
-      return 0;
-  }
-  return 0;
-}
-
-int connection_exit_finished_flushing(connection_t *conn) {
-  int e, len=sizeof(e);
-
-  assert(conn && conn->type == CONN_TYPE_EXIT);
-
-  switch(conn->state) {
-    case EXIT_CONN_STATE_CONNECTING:
-      if (getsockopt(conn->s, SOL_SOCKET, SO_ERROR, &e, &len) < 0)  { /* not yet */
-        if(errno != EINPROGRESS){
-          /* yuck. kill it. */
-          log(LOG_DEBUG,"connection_exit_finished_flushing(): in-progress connect failed. Removing.");
-          return -1;
-        } else {
-          log(LOG_DEBUG,"connection_exit_finished_flushing(): in-progress connect still waiting.");
-          return 0; /* no change, see if next time is better */
-        }
-      }
-      /* the connect has finished. */
-
-      log(LOG_DEBUG,"connection_exit_finished_flushing(): Connection to %s:%u established.",
-          conn->address,conn->port);
-
-      conn->state = EXIT_CONN_STATE_OPEN;
-      connection_watch_events(conn, POLLIN); /* stop writing, continue reading */
-      if(connection_wants_to_flush(conn)) /* in case there are any queued data cells */
-        connection_start_writing(conn);
-      return
-        connection_exit_send_connected(conn) || /* deliver a 'connected' data cell back through the circuit. */
-        connection_process_inbuf(conn); /* in case the server has written anything */
-    case EXIT_CONN_STATE_OPEN:
-      /* FIXME down the road, we'll clear out circuits that are pending to close */
-      log(LOG_DEBUG,"connection_exit_finished_flushing(): finished flushing.");
-      connection_stop_writing(conn);
-#ifdef USE_ZLIB
-      if (connection_decompress_to_buf(NULL, 0, conn, Z_SYNC_FLUSH) < 0)
-        return -1;
-#endif
-      connection_consider_sending_sendme(conn, EDGE_EXIT);
-      return 0;
-    default:
-      log(LOG_DEBUG,"Bug: connection_exit_finished_flushing() called in unexpected state.");
-      return 0;
-  }
-
-  return 0;
-}
-
-int connection_exit_send_connected(connection_t *conn) {
-  circuit_t *circ;
-  cell_t cell;
-
-  assert(conn);
-
-  circ = circuit_get_by_conn(conn);
-
-  if(!circ) {
-    log(LOG_DEBUG,"connection_exit_send_connected(): client-side sent destroy just as we completed server connection. Closing.");
-    return -1;
-  }
-
-  memset(&cell, 0, sizeof(cell_t));
-  cell.aci = circ->p_aci;
-  cell.command = CELL_DATA;
-  *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
-  *cell.payload = TOPIC_COMMAND_CONNECTED;
-  cell.length = TOPIC_HEADER_SIZE;
-  log(LOG_INFO,"connection_exit_send_connected(): passing back cell (aci %d).",circ->p_aci);
-
-  if(circuit_deliver_data_cell_from_edge(&cell, circ, EDGE_EXIT) < 0) {
-    log(LOG_DEBUG,"connection_exit_send_connected(): circuit_deliver_data_cell (backward) failed. Closing.");
-    circuit_close(circ);
-    return 0;
-  }
-  return 0;
-}
-
 int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
   connection_t *n_conn;
   char *colon;
@@ -192,149 +63,6 @@ int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
   return 0;
 }
 
-int connection_exit_process_data_cell(cell_t *cell, circuit_t *circ) {
-  connection_t *conn;
-  int topic_command;
-  int topic_id;
-  int len;
-  static int num_seen=0;
-
-  /* an outgoing data cell has arrived */
-
-  assert(cell && circ);
-
-  topic_command = *cell->payload;
-  topic_id = ntohs(*(uint16_t *)(cell->payload+2));
-  log(LOG_DEBUG,"connection_exit_process_data_cell(): command %d topic %d", topic_command, topic_id);
-  num_seen++;
-  log(LOG_DEBUG,"connection_exit_process_data_cell(): Now seen %d data cells here.", num_seen);
-
-  circuit_consider_sending_sendme(circ, EDGE_EXIT);
-
-  for(conn = circ->n_conn; conn && conn->topic_id != topic_id; conn = conn->next_topic) ;
-
-  /* now conn is either NULL, in which case we don't recognize the topic_id, or
-   * it is set, in which case cell is talking about this conn.
-   */
-
-  if(conn && conn->state != EXIT_CONN_STATE_OPEN) {
-    if(topic_command == TOPIC_COMMAND_END) {
-      log(LOG_ERR,"connection_exit_process_data_cell(): Got an end before we're connected. Marking for close.");
-      conn->marked_for_close = 1;
-      return 0;
-    } else {
-      log(LOG_INFO,"connection_exit_process_data_cell(): Got a non-end data cell when not in 'open' state. Dropping.");
-      return 0;
-    }
-  }
-
-  switch(topic_command) {
-    case TOPIC_COMMAND_BEGIN:
-      if(conn) {
-        log(LOG_INFO,"connection_exit_process_data_cell(): begin cell for known topic. Dropping.");
-        return 0;
-      }
-      return connection_exit_begin_conn(cell, circ);
-    case TOPIC_COMMAND_DATA:
-      if(!conn) {
-        log(LOG_INFO,"connection_exit_process_data_cell(): data cell for unknown topic. Dropping.");
-        return 0;
-      }
-      if(--conn->p_receive_topicwindow < 0) { /* is it below 0 after decrement? */
-        log(LOG_DEBUG,"connection_exit_process_data_cell(): receive_topicwindow at exit below 0. Killing.");
-        return -1; /* AP breaking protocol. kill the whole circuit. */
-      }
-      log(LOG_DEBUG,"connection_exit_process_data_cell(): willing to receive %d more cells from circ",conn->p_receive_topicwindow);
-
-      if(conn->state != EXIT_CONN_STATE_OPEN) {
-        log(LOG_DEBUG,"connection_exit_process_data_cell(): data received while resolving/connecting. Queueing.");
-      }
-#ifdef USE_ZLIB
-      log(LOG_DEBUG,"connection_exit_process_data_cell(): uncompressing %d bytes onto outbuf...",cell->length - TOPIC_HEADER_SIZE);
-      len = connection_decompress_to_buf(cell->payload + TOPIC_HEADER_SIZE,
-                                         cell->length - TOPIC_HEADER_SIZE, 
-                                         conn, Z_SYNC_FLUSH);
-      log(LOG_DEBUG,"%d bytes written", len);
-      if (len<0) {
-        log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
-        conn->marked_for_close = 1;
-        return 0;
-      }
-#else
-      log(LOG_DEBUG,"connection_exit_process_data_cell(): put %d bytes on outbuf.",cell->length - TOPIC_HEADER_SIZE);
-      if(connection_write_to_buf(cell->payload + TOPIC_HEADER_SIZE,
-                                 cell->length - TOPIC_HEADER_SIZE, conn) < 0) {
-        log(LOG_INFO,"connection_exit_process_data_cell(): write to buf failed. Marking for close.");
-        conn->marked_for_close = 1;
-        return 0;
-      }
-#endif
-      if(connection_consider_sending_sendme(conn, EDGE_EXIT) < 0)
-        conn->marked_for_close = 1;
-      return 0;
-    case TOPIC_COMMAND_END:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_exit_process_data_cell(): end cell dropped, unknown topic %d.",topic_id);
-        return 0;
-      }
-      log(LOG_DEBUG,"connection_exit_process_data_cell(): end cell for topic %d. Removing topic.",topic_id);
-
-#if 0
-      /* go through and identify who points to conn. remove conn from the list. */
-      if(conn == circ->n_conn) {
-        circ->n_conn = conn->next_topic;
-      }
-      for(prevconn = circ->n_conn; prevconn->next_topic != conn; prevconn = prevconn->next_topic) ;
-      prevconn->next_topic = conn->next_topic;
-#endif
-#ifdef HALF_OPEN
-      conn->done_sending = 1;
-      shutdown(conn->s, 1); /* XXX check return; refactor NM */
-      if (conn->done_receiving)
-        conn->marked_for_close = 1;
-#endif
-        
-      conn->marked_for_close = 1;
-      break;
-    case TOPIC_COMMAND_CONNECTED:
-      log(LOG_INFO,"connection_exit_process_data_cell(): topic connected request unsupported. Dropping.");
-      break;
-    case TOPIC_COMMAND_SENDME:
-      if(!conn) {
-        log(LOG_DEBUG,"connection_exit_process_data_cell(): sendme cell dropped, unknown topic %d.",topic_id);
-        return 0;
-      }
-      conn->n_receive_topicwindow += TOPICWINDOW_INCREMENT;
-      connection_start_reading(conn);
-      connection_package_raw_inbuf(conn); /* handle whatever might still be on the inbuf */
-      circuit_consider_stop_edge_reading(circ, EDGE_EXIT);
-      break;
-    default:
-      log(LOG_DEBUG,"connection_exit_process_data_cell(): unknown topic command %d.",topic_command);
-  }
-  return 0;
-}
-
-#if 0
-static uint32_t address_to_addr(char *address) {
-  struct hostent *rent;
-  uint32_t addr;
-  char *caddr;
-
-  rent = gethostbyname(address);
-  if (!rent) { 
-    log(LOG_ERR,"address_to_addr(): Could not resolve dest addr %s.",address);
-    return 0;
-  }
-  memcpy(&addr, rent->h_addr,rent->h_length); 
-  addr = ntohl(addr); /* get it back to host order */
-  caddr = (char *)&addr;
-  log(LOG_DEBUG,"address_to_addr(): addr is %d %d %d %d",
-    caddr[0], caddr[1], caddr[2], caddr[3]);
-  return addr;
-}
-#endif
-
 int connection_exit_connect(connection_t *conn) {
   int s; /* for the new socket */
   struct sockaddr_in dest_addr;
@@ -391,7 +119,7 @@ int connection_exit_connect(connection_t *conn) {
   connection_watch_events(conn, POLLIN);
 
   /* also, deliver a 'connected' cell back through the circuit. */
-  return connection_exit_send_connected(conn);
+  return connection_edge_send_command(conn, circuit_get_by_conn(conn), TOPIC_COMMAND_CONNECTED);
 }
 
 /*
diff --git a/src/or/or.h b/src/or/or.h
index 80105f38b..df7d4085c 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -102,9 +102,9 @@
 #define EXIT_CONN_STATE_CLOSE_WAIT 4 /* have sent a destroy, awaiting a confirmation */
 #endif
 
-#define AP_CONN_STATE_SOCKS_WAIT 0
-#define AP_CONN_STATE_OR_WAIT 1
-#define AP_CONN_STATE_OPEN 2
+#define AP_CONN_STATE_SOCKS_WAIT 3
+#define AP_CONN_STATE_OR_WAIT 4
+#define AP_CONN_STATE_OPEN 5
 
 #define DIR_CONN_STATE_CONNECTING 0
 #define DIR_CONN_STATE_SENDING_COMMAND 1
@@ -142,8 +142,8 @@
 
 #define CELL_DIRECTION_IN 1
 #define CELL_DIRECTION_OUT 2
-#define EDGE_EXIT 3 /* make direction and edge values not overlap, to help catch bugs */
-#define EDGE_AP 4
+#define EDGE_EXIT CONN_TYPE_EXIT
+#define EDGE_AP CONN_TYPE_AP
 
 #define CIRCWINDOW_START 1000
 #define CIRCWINDOW_INCREMENT 100
@@ -603,8 +603,6 @@ int connection_finished_flushing(connection_t *conn);
 
 /********************************* connection_ap.c ****************************/
 
-int connection_ap_process_inbuf(connection_t *conn);
-
 int ap_handshake_process_socks(connection_t *conn);
 
 int ap_handshake_create_onion(connection_t *conn);
@@ -618,23 +616,23 @@ int ap_handshake_send_onion(connection_t *ap_conn, connection_t *or_conn, circui
 int ap_handshake_send_begin(connection_t *ap_conn, circuit_t *circ);
 
 int ap_handshake_socks_reply(connection_t *conn, char result);
-int connection_ap_send_connected(connection_t *conn);
-int connection_ap_process_data_cell(cell_t *cell, circuit_t *circ);
-
-int connection_ap_finished_flushing(connection_t *conn);
 
 int connection_ap_create_listener(struct sockaddr_in *bindaddr);
 
 int connection_ap_handle_listener_read(connection_t *conn);
 
+/********************************* connection_edge.c ***************************/
+
+int connection_edge_process_inbuf(connection_t *conn);
+int connection_edge_send_command(connection_t *conn, int topic_command);
+int connection_edge_process_data_cell(cell_t *cell, circuit_t *circ, int edge_type);
+int connection_edge_finished_flushing(connection_t *conn);
+
 /********************************* connection_exit.c ***************************/
 
-int connection_exit_process_inbuf(connection_t *conn);
-int connection_exit_package_inbuf(connection_t *conn);
 int connection_exit_send_connected(connection_t *conn);
-int connection_exit_process_data_cell(cell_t *cell, circuit_t *circ);
+int connection_exit_begin_conn(cell_t *cell, circuit_t *circ);
 
-int connection_exit_finished_flushing(connection_t *conn);
 int connection_exit_connect(connection_t *conn);
 
 /********************************* connection_op.c ***************************/