iothread find msg.

diff --git a/dnstap/dtstream.c b/dnstap/dtstream.c
index 318e04bfc577a5d5bc7741f1e917b9937eb10028..db3cac475810d5802281835712480bfabe163151 100644 (file)
--- a/dnstap/dtstream.c
+++ b/dnstap/dtstream.c
@@ -196,15 +196,59 @@ void dt_io_thread_unregister_queue(struct dt_io_thread* dtio,
        }
 }
 
+/** pick a message from the queue, lock and unlock, true if a message */
+static int pick_msg_from_queue(struct dt_msg_queue* mq, void** buf,
+       size_t* len)
+{
+       lock_basic_lock(&mq->lock);
+       if(mq->first) {
+               struct dt_msg_entry* entry = mq->first;
+               mq->first = entry->next;
+               if(!entry->next) mq->last = NULL;
+               mq->cursize -= entry->len;
+               lock_basic_unlock(&mq->lock);
+
+               *buf = entry->buf;
+               *len = entry->len;
+               free(entry);
+               return 1;
+       }
+       lock_basic_unlock(&mq->lock);
+       return 0;
+}
+
+/** find message in queue, false if no message, true if message to send */
+static int dtio_find_in_queue(struct dt_io_thread* dtio,
+       struct dt_msg_queue* mq)
+{
+       void* buf=NULL;
+       size_t len=0;
+       if(pick_msg_from_queue(mq, &buf, &len)) {
+               dtio->cur_msg = buf;
+               dtio->cur_msg_len = len;
+               dtio->cur_msg_done = 0;
+               return 1;
+       }
+       return 0;
+}
+
 /** find a new message to write, search message queues, false if none */
 static int dtio_find_msg(struct dt_io_thread* dtio)
 {
+       struct dt_io_list_item* item = dtio->io_list;
+       while(item) {
+               if(dtio_find_in_queue(dtio, item->queue))
+                       return 1;
+               item = item->next;
+       }
+       return 0;
 }
 
 /** write more of the current messsage. false if incomplete, true if
  * the message is done */
-static int dtio_write_more(struct dt_io_thread* dtio)
+static int dtio_write_more(int fd, struct dt_io_thread* dtio)
 {
+       return 0;
 }
 
 /** callback for the dnstap events, to write to the output */
@@ -220,7 +264,7 @@ static void dtio_output_cb(int fd, short ATTR_UNUSED(bits), void* arg)
 
        /* write it */
        if(dtio->cur_msg_done < dtio->cur_msg_len) {
-               if(!dtio_write_more(dtio))
+               if(!dtio_write_more(fd, dtio))
                        return;
        }