X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/2a10b70b1cdbd55e9f2c26ed52dcff5255cd6e96..8b2e93032955c82b9e9958b98090572a28573e84:/lib/eclient.c

diff --git a/lib/eclient.c b/lib/eclient.c
index 2d57d73..d93412a 100644
--- a/lib/eclient.c
+++ b/lib/eclient.c
@@ -37,6 +37,7 @@
 #include <assert.h>
 #include <inttypes.h>
 #include <stddef.h>
+#include <time.h>
 
 #include "log.h"
 #include "mem.h"
@@ -58,6 +59,9 @@
 
 /* TODO: more commands */
 
+/** @brief How often to send data to the server when receiving logs */
+#define LOG_PROD_INTERVAL 10
+
 /* Types *********************************************************************/
 
 /** @brief Client state */
@@ -125,9 +129,23 @@ struct disorder_eclient {
   int rc;                               /**< @brief response code */
   char *line;                           /**< @brief complete line */
   struct vector vec;                    /**< @brief body */
-  const disorder_eclient_log_callbacks *log_callbacks; /**< @brief log callbacks */
+
+  const disorder_eclient_log_callbacks *log_callbacks;
+  /**< @brief log callbacks
+   *
+   * Once disorder_eclient_log() has been issued this is always set.  When we
+   * re-connect it is checked to re-issue the log command.
+   */
   void *log_v;                          /**< @brief user data */
   unsigned long statebits;              /**< @brief latest state */
+
+  time_t last_prod;
+  /**< @brief last time we sent a prod
+   *
+   * When we are receiving log data we send a "prod" byte to the server from
+   * time to time so that we detect broken connections reasonably quickly.  The
+   * server just ignores these bytes.
+   */
 };
 
 /* Forward declarations ******************************************************/
@@ -170,6 +188,7 @@ static void logentry_removed(disorder_eclient *c, int nvec, char **vec);
 static void logentry_scratched(disorder_eclient *c, int nvec, char **vec);
 static void logentry_state(disorder_eclient *c, int nvec, char **vec);
 static void logentry_volume(disorder_eclient *c, int nvec, char **vec);
+static void logentry_rescanned(disorder_eclient *c, int nvec, char **vec);
 
 /* Tables ********************************************************************/
 
@@ -194,6 +213,7 @@ static const struct logentry_handler logentry_handlers[] = {
   LE(recent_added, 2, INT_MAX),
   LE(recent_removed, 1, 1),
   LE(removed, 1, 2),
+  LE(rescanned, 0, 0),
   LE(scratched, 2, 2),
   LE(state, 1, 1),
   LE(volume, 2, 2)
@@ -306,6 +326,7 @@ static int protocol_error(disorder_eclient *c, struct operation *op,
  */
 void disorder_eclient_polled(disorder_eclient *c, unsigned mode) {
   struct operation *op;
+  time_t now;
   
   D(("disorder_eclient_polled fd=%d state=%s mode=[%s %s]",
      c->fd, states[c->state],
@@ -374,6 +395,14 @@ void disorder_eclient_polled(disorder_eclient *c, unsigned mode) {
       c->callbacks->report(c->u, 0);
   }
 
+  /* Queue up a byte to send */
+  if(c->state == state_log
+     && c->output.nvec == 0
+     && time(&now) - c->last_prod > LOG_PROD_INTERVAL) {
+    put(c, "x", 1);
+    c->last_prod = now;
+  }
+  
   if(c->state == state_cmdresponse
      || c->state == state_body
      || c->state == state_log) {
@@ -453,6 +482,7 @@ static int start_connect(void *cc,
     return comms_error(c, "socket: %s", strerror(errno));
   c->eof = 0;
   nonblock(c->fd);
+  cloexec(c->fd);
   if(connect(c->fd, sa, len) < 0) {
     switch(errno) {
     case EINTR:
@@ -649,7 +679,7 @@ static void process_line(disorder_eclient *c, char *line) {
     case 3:
       /* We need to collect the body. */
       c->state = state_body;
-      c->vec.nvec = 0;
+      vector_init(&c->vec);
       break;
     case 4:
       assert(c->log_callbacks != 0);
@@ -1175,6 +1205,35 @@ int disorder_eclient_new_tracks(disorder_eclient *c,
                 "new", limit, (char *)0);
 }
 
+static void rtp_response_opcallback(disorder_eclient *c,
+                                    struct operation *op) {
+  D(("rtp_response_opcallback"));
+  if(c->rc / 100 == 2) {
+    if(op->completed) {
+      int nvec;
+      char **vec = split(c->line + 4, &nvec, SPLIT_QUOTES, 0, 0);
+
+      ((disorder_eclient_list_response *)op->completed)(op->v, nvec, vec);
+    }
+  } else
+    protocol_error(c, op, c->rc, "%s: %s", c->ident, c->line);
+}
+
+/** @brief Determine the RTP target address
+ * @param c Client
+ * @param completed Called with address details
+ * @param v Passed to @p completed
+ *
+ * The address details will be two elements, the first being the hostname and
+ * the second the service (port).
+ */
+int disorder_eclient_rtp_address(disorder_eclient *c,
+                                 disorder_eclient_list_response *completed,
+                                 void *v) {
+  return simple(c, rtp_response_opcallback, (void (*)())completed, v,
+                "rtp-address", (char *)0);
+}
+
 /* Log clients ***************************************************************/
 
 /** @brief Monitor the server log
@@ -1309,6 +1368,13 @@ static void logentry_removed(disorder_eclient *c,
   c->log_callbacks->removed(c->log_v, vec[0], vec[1]);
 }
 
+static void logentry_rescanned(disorder_eclient *c,
+                               int attribute((unused)) nvec,
+                               char attribute((unused)) **vec) {
+  if(!c->log_callbacks->rescanned) return;
+  c->log_callbacks->rescanned(c->log_v);
+}
+
 static void logentry_scratched(disorder_eclient *c,
                                int attribute((unused)) nvec, char **vec) {
   if(!c->log_callbacks->scratched) return;