Changeset 6bb9ef8a16723394cc4e5c511f06fe2173d78a11

Timestamp:

08/13/08 21:36:52 (5 years ago)

Author:

Theo Schlossnagle <jesus@omniti.com>

git-committer:

Theo Schlossnagle <jesus@omniti.com> 1218663412 +0000

git-parent:

[222a9737db8ea7fefdf3ee212cae3e505459add2]

git-author:

Theo Schlossnagle <jesus@omniti.com> 1218663412 +0000

Message:

closes #52. flag-day(noitd,stratcond)

Files:

• src/noit.conf.in (modified) (1 diff)
• src/noit_jlog_listener.c (modified) (2 diffs)
• src/noit_jlog_listener.h (modified) (1 diff)
• src/noit_listener.c (modified) (3 diffs)
• src/noit_listener.h (modified) (2 diffs)
• src/stratcon_jlog_streamer.c (modified) (3 diffs)

src/noit.conf.in

@@ -74 +74 @@
-      <listener address="*" port="32323" ssl="on"/>
-    </consoles>
-log_transit
+control_dispatch
-      <config>
->feed</log
+_transit_feed_name>feed</log_transit_feed_name
-      </config>
-    </listener>

src/noit_jlog_listener.c

@@ -20 +20 @@
-noit_jlog_listener_init() {
-  eventer_name_callback("log_transit", noit_jlog_handler);
+  noit_control_dispatch_delegate(noit_control_dispatch,
+                                 NOIT_JLOG_DATA_FEED,
+                                 noit_jlog_handler);
-}
-
@@ -183 +186 @@
-    char path[PATH_MAX], *sub;
-    jcl = ac->service_ctx = noit_jlog_closure_alloc();
-
+
+
+
-                           (void **)&logname)) {
-
+_transit_feed_name
-      goto socket_error;
-    }

src/noit_jlog_listener.h

@@ -10 +10 @@
-#include "eventer/eventer.h"
-
+#define NOIT_JLOG_DATA_FEED 0xda7afeed
+
-API_EXPORT(void)
-  noit_jlog_listener_init(void);

src/noit_listener.c

@@ -20 +20 @@
-#include "noit_listener.h"
-#include "noit_conf.h"
+
+static noit_hash_table listener_commands = NOIT_HASH_EMPTY;
-
-void
@@ -260 +262 @@
-    calloc(1, sizeof(*listener_closure->dispatch_closure));
-  listener_closure->dispatch_closure->config = config;
+  listener_closure->dispatch_closure->dispatch = handler;
-  listener_closure->dispatch_closure->service_ctx = service_ctx;
-
@@ -337 +340 @@
-  }
-}
+int
+noit_control_dispatch(eventer_t e, int mask, void *closure,
+                      struct timeval *now) {
+  u_int32_t cmd;
+  int len;
+  noit_hash_table *delegation_table;
+  acceptor_closure_t *ac = closure;
+
+  len = e->opset->read(e->fd, &cmd, sizeof(cmd), &mask, e);
+
+  if(len == -1 && errno == EAGAIN)
+    return EVENTER_READ | EVENTER_EXCEPTION;
+
+  if(mask & EVENTER_EXCEPTION || len != sizeof(cmd)) {
+    int newmask;
+socket_error:
+    /* Exceptions cause us to simply snip the connection */
+    eventer_remove_fd(e->fd);
+    e->opset->close(e->fd, &newmask, e);
+    if(ac) acceptor_closure_free(ac);
+    return 0;
+  }
+
+  cmd = ntohl(cmd);
+  /* Lookup cmd and dispatch */
+  if(noit_hash_retrieve(&listener_commands,
+                        (char *)&ac->dispatch, sizeof(ac->dispatch),
+                        (void **)&delegation_table)) {
+    eventer_func_t *eventer_func;
+    if(noit_hash_retrieve(delegation_table,
+                          (char *)&cmd, sizeof(cmd),
+                          (void **)&eventer_func)) {
+      e->callback = *eventer_func;
+      return e->callback(e, mask, closure, now);
+    }
+    else {
+    const char *event_name;
+      noitL(noit_error, "listener (%s %p) has no command: 0x%8x\n",
+            (event_name = eventer_name_for_callback(ac->dispatch))?event_name:"???",
+            delegation_table, cmd);
+    }
+  }
+  else {
+    const char *event_name;
+    noitL(noit_error, "No delegation table for listener (%s %p)\n",
+          (event_name = eventer_name_for_callback(ac->dispatch))?event_name:"???",
+          delegation_table);
+  }
+  goto socket_error;
+}
+void
+noit_control_dispatch_delegate(eventer_func_t listener_dispatch,
+                               u_int32_t cmd,
+                               eventer_func_t delegate_dispatch) {
+  u_int32_t *cmd_copy;
+  eventer_func_t *handler_copy;
+  noit_hash_table *delegation_table;
+  if(!noit_hash_retrieve(&listener_commands,
+                         (char *)&listener_dispatch, sizeof(listener_dispatch),
+                         (void **)delegation_table)) {
+    delegation_table = calloc(1, sizeof(*delegation_table));
+    handler_copy = malloc(sizeof(*handler_copy));
+    *handler_copy = listener_dispatch;
+    noit_hash_store(&listener_commands,
+                    (char *)handler_copy, sizeof(*handler_copy),
+                    delegation_table);
+  }
+  cmd_copy = malloc(sizeof(*cmd_copy));
+  *cmd_copy = cmd;
+  handler_copy = malloc(sizeof(*handler_copy));
+  *handler_copy = delegate_dispatch;
+  noit_hash_replace(delegation_table,
+                    (char *)cmd_copy, sizeof(*cmd_copy),
+                    handler_copy,
+                    free, free);
+}
+
-void
-noit_listener_init(const char *toplevel) {
-  eventer_name_callback("noit_listener_acceptor", noit_listener_acceptor);
-  eventer_name_callback("noit_listener_accept_ssl", noit_listener_accept_ssl);
+  eventer_name_callback("control_dispatch", noit_control_dispatch);
-  noit_listener_reconfig(toplevel);
-}

src/noit_listener.h

@@ -28 +28 @@
-  noit_hash_table *config;
-  void *service_ctx;
+  eventer_func_t dispatch;
-} acceptor_closure_t;
-
@@ -49 +50 @@
-  acceptor_closure_free(acceptor_closure_t *ac);
-
+API_EXPORT(void)
+  noit_control_dispatch_delegate(eventer_func_t listener_dispatch,
+                                 u_int32_t cmd,
+                                 eventer_func_t delegate_dispatch);
+
+int noit_control_dispatch(eventer_t, int, void *, struct timeval *);
+
-#endif

src/stratcon_jlog_streamer.c

@@ -45 +45 @@
-
-  enum {
-
-
-
-
+
+
+
+
+
-  } state;
-  int count;            /* Number of jlog messages we need to read */
@@ -180 +181 @@
-stratcon_jlog_recv_handler(eventer_t e, int mask, void *closure,
-                           struct timeval *now) {
+  static u_int32_t jlog_feed_cmd = 0;
-  jlog_streamer_ctx_t *ctx = closure;
-  int len;
-  jlog_id n_chkpt;
-
+  if(!jlog_feed_cmd) jlog_feed_cmd = htonl(NOIT_JLOG_DATA_FEED);
+
-  if(mask & EVENTER_EXCEPTION || ctx->wants_shutdown) {
- socket_error:
-COUNT
+INITIATE
-    ctx->count = 0;
-    ctx->bytes_read = 0;
@@ -200 +204 @@
-  while(1) {
-    switch(ctx->state) {
+      case WANT_INITIATE:
+        len = e->opset->write(e->fd, &jlog_feed_cmd, sizeof(&jlog_feed_cmd),
+                              &mask, e);
+        if(len < 0) {
+          if(errno == EAGAIN) return mask | EVENTER_EXCEPTION;
+          goto socket_error;
+        }
+        if(len != sizeof(jlog_feed_cmd)) {
+          noitL(noit_error, "short write on initiating stream.\n");
+          goto socket_error;
+        }
+        ctx->state = WANT_COUNT;
+        break;
+
-      case WANT_COUNT:
-        FULLREAD(e, ctx, sizeof(u_int32_t));