Hook up the put command with the actual queue.

beanstalkd.c

@@ -10,11 +10,14 @@
 #include "conn.h"
 #include "net.h"
 #include "beanstalkd.h"
+#include "pq.h"
 #include "util.h"
 
 /* job data cannot be greater than this */
 #define JOB_DATA_SIZE_LIMIT ((1 << 16) - 1)
 
+static pq q;
+
 static void
 drop_root()
 {
@@ -100,17 +103,21 @@ reply_word(conn c, const char *word, int len)
 static void
 enqueue_job(conn c)
 {
+ int r;
  job j = c->job;
 
+ c->job = NULL; /* the connection no longer owns this job */
+
  /* check if the trailer is present and correct */
  if (memcmp(j->data + j->data_size - 2, "\r\n", 2)) return conn_close(c);
 
  /* we have a complete job, so let's stick it in the pqueue */
- /* TODO stick it in */
- warn("TODO stick it in");
+ r = pq_give(q, j);
+
+ if (r) return reply_word(c, MSG_INSERTED, MSG_INSERTED_LEN);
 
- c->job = NULL;
- reply_word(c, MSG_INSERTED, MSG_INSERTED_LEN);
+ free(j); /* the job didn't go in the queue, so it goes bye bye */
+ reply_word(c, MSG_NOT_INSERTED, MSG_NOT_INSERTED_LEN);
 }
 
 static void
@@ -321,7 +328,6 @@ h_accept(const int fd, const short which, struct event *ev)
 int
 main(int argc, char **argv)
 {
- /*q q;*/
  int listen_socket;
  struct event listen_evq;
 
@@ -336,7 +342,7 @@ main(int argc, char **argv)
  event_set(&listen_evq, listen_socket, EV_READ | EV_PERSIST, (evh) h_accept, &listen_evq);
  event_add(&listen_evq, NULL);
 
- /*q = q_new();*/
+ q = make_pq(HEAP_SIZE);
 
  event_dispatch();
  return 0;

beanstalkd.h

@@ -6,13 +6,17 @@
 #define HOST INADDR_ANY
 #define PORT 3232
 
+/* space for 16 Mi jobs */
+#define HEAP_SIZE 16 * 1024 * 1024
+
 #define CONSTSTRLEN(m) (sizeof(m) - 1)
 
 #define MSG_INSERTED "INSERTED\r\n"
-#define MSG_ERR "NOT_INSERTED\r\n"
+#define MSG_NOT_INSERTED "NOT_INSERTED\r\n"
 #define MSG_NOTFOUND "NOT_FOUND\r\n"
 
 #define MSG_INSERTED_LEN CONSTSTRLEN(MSG_INSERTED)
+#define MSG_NOT_INSERTED_LEN CONSTSTRLEN(MSG_NOT_INSERTED)
 #define MSG_NOTFOUND_LEN CONSTSTRLEN(MSG_NOTFOUND)
 
 #define CMD_PUT "put "

pq.h

@@ -5,9 +5,6 @@
 
 #include "job.h"
 
-/* space for 16 Mi jobs */
-#define HEAP_SIZE 16 * 1024 * 1024
-
 typedef struct pq {
  unsigned int size;
  unsigned int used;