1

#include "irc_queue.h"

     2

#include "log.h"

     3

     4

#include <stdlib.h>

     5

#include <string.h>

     6

#include <time.h>

     7

     8

/**

     9

 * Send a formatted line on the line_proto, and apply the IRC_QUEUE_PENALTY to the timer, resetting it to the current

    10

 * time if it's fallen behind.

    11

 */

    12

static err_t irc_queue_send_buf (struct irc_queue *queue, const char *line_buf)

    13

{

    14

    time_t now = time(NULL);

    15

    err_t err;

    16

    int ret;

    17

    18

    // dump for debug output, without any newline

    19

    log_debug("%.*s", (int) (strstr(line_buf, "\r\n") - line_buf), line_buf);

    20

    21

    // XXX: output buffering?

    22

    if ((ret = line_proto_send(queue->lp, line_buf)) < 0)

    23

        return (err = -ret);

    24

    25

    // reset timer to current time if it's fallen behind

    26

    if (queue->timer < now)

    27

        queue->timer = now;

    28

    29

    // apply penalty

    30

    queue->timer += IRC_QUEUE_PENALTY;

    31

    32

    // ok

    33

    return SUCCESS;

    34

}

    35

    36

/**

    37

 * Pump the queue, sending the next message. If it is given explicitly, then the given message is "pumped", and it is

    38

 * assumed to not be in the queue_list. Otherwise, the next entry is taken from the queue.

    39

 *

    40

 * If the queue is empty and no entry is given, does nothing.

    41

 */

    42

static err_t irc_queue_pump (struct irc_queue *queue, struct irc_queue_entry *next)

    43

{

    44

    err_t err;

    45

    46

    // pop the next entry?

    47

    if (next == NULL) {

    48

        // take it from the head of the list

    49

        next = TAILQ_FIRST(&queue->list);

    50

    51

        // nothing to do?

    52

        if (!next)

    53

            return SUCCESS;

    54

    55

        // and then remove it

    56

        TAILQ_REMOVE(&queue->list, next, queue_list);

    57

    }

    58

    59

    // send it

    60

    if ((err = irc_queue_send_buf(queue, next->line_buf)))

    61

        goto error;

    62

    63

    // ok, release it

    64

    free(next);

    65

    66

    // done

    67

    return SUCCESS;

    68

    69

error:

    70

    // hmm... re-insert and return error, we can try again later

    71

    TAILQ_INSERT_HEAD(&queue->list, next, queue_list);

    72

    73

    return err;    

    74

}

    75

    76

/**

    77

 * Schedule the queue's timer for execution once the timer has expired, or immediately if it's not full. If next is

    78

 * given, it should be the next entry to send, and not part of the queue_list.

    79

 */

    80

static err_t irc_queue_schedule (struct irc_queue *queue)

    81

{

    82

    time_t now = time(NULL);

    83

    struct timeval tv;

    84

    85

    // calculate the timeout

    86

    int timeout = (queue->timer - (now + IRC_QUEUE_WINDOW));

    87

    88

    // setup the timeout, zero if we don't actually need to wait...

    89

    tv.tv_sec = timeout > 0 ? timeout : 0;

    90

    tv.tv_usec = 0;

    91

    92

    // setup the timer event

    93

    if (evtimer_add(queue->ev, &tv))

    94

        return ERR_EVENT_ADD;

    95

    96

    // ok

    97

    return SUCCESS;

    98

}

    99

   100

/**

   101

 * Our timer callback

   102

 */

   103

static void irc_queue_timer (int fd, short what, void *arg)

   104

{

   105

    time_t now = time(NULL);

   106

    struct irc_queue *queue = arg;

   107

    err_t err;

   108

   109

    (void) fd;

   110

    (void) what;

   111

   112

    // pump the queue until our timer is full again, or the queue is empty

   113

    while (queue->timer <= now + IRC_QUEUE_WINDOW && !TAILQ_EMPTY(&queue->list)) {

   114

        if ((err = irc_queue_pump(queue, NULL))) {

   115

            log_warn("irc_queue_pump: %s", error_name(err));

   116

            break;

   117

        }

   118

    }

   119

   120

    // reschedule if needed

   121

    if (!TAILQ_EMPTY(&queue->list)) {

   122

        if ((err = irc_queue_schedule(queue))) {

   123

            log_err(err, "irc_queue_scheulde");

   124

        }

   125

    }

   126

}

   127

   128

err_t irc_queue_create (struct irc_queue **queue_ptr, struct event_base *ev_base, struct line_proto *lp, error_t *err)

   129

{

   130

    struct irc_queue *queue;

   131

   132

    // alloc

   133

    if ((queue = calloc(1, sizeof(*queue))) == NULL)

   134

        return SET_ERROR(err, ERR_CALLOC);

   135

   136

    // create the timer event

   137

    if ((queue->ev = evtimer_new(ev_base, &irc_queue_timer, queue)) == NULL)

   138

        JUMP_SET_ERROR(err, ERR_EVENT_NEW);

   139

   140

    // initialize

   141

    queue->lp = lp;

   142

    queue->timer = time(NULL);

   143

    TAILQ_INIT(&queue->list); 

   144

   145

    // ok

   146

    *queue_ptr = queue;

   147

   148

    return SUCCESS;

   149

   150

error:

   151

    // cleanup

   152

    free(queue);

   153

   154

    return ERROR_CODE(err);    

   155

}

   156

   157

/**

   158

 * Attempt to send a irc_line directly on the queue's line_proto, otherwise enqueueing it for later transmission.

   159

 */

   160

static err_t irc_queue_send_direct (struct irc_queue *queue, const struct irc_line *line)

   161

{

   162

    char line_buf[IRC_LINE_MAX + 2];

   163

    err_t err;

   164

   165

    // format

   166

    if ((err = irc_line_build(line, line_buf)))

   167

        return err;

   168

   169

    // add CRLF

   170

    strcat(line_buf, "\r\n");

   171

   172

    // send

   173

    // XXX: handle send-buffer-full by enqueuing it after all

   174

    return irc_queue_send_buf(queue, line_buf);

   175

}

   176

   177

/**

   178

 * Enqueue a irc_line onto the queue's list, and schedule the timer for execution

   179

 */

   180

static err_t irc_queue_put (struct irc_queue *queue, const struct irc_line *line)

   181

{

   182

    struct irc_queue_entry *entry;

   183

    err_t err;

   184

   185

    // alloc

   186

    if ((entry = calloc(1, sizeof(*entry))) == NULL)

   187

        return ERR_CALLOC;

   188

   189

    // format

   190

    if ((err = irc_line_build(line, entry->line_buf)))

   191

        goto error;

   192

   193

    // add CRLF

   194

    strcat(entry->line_buf, "\r\n");

   195

   196

    // then re-schedule the queue

   197

    if ((err = irc_queue_schedule(queue)))

   198

        goto error;

   199

   200

    // append to end of list

   201

    TAILQ_INSERT_TAIL(&queue->list, entry, queue_list);

   202

   203

    // ok

   204

    return SUCCESS;

   205

   206

error:

   207

    // cleanup

   208

    free(entry);

   209

   210

    return err;    

   211

}

   212

   213

err_t irc_queue_process (struct irc_queue *queue, const struct irc_line *line)

   214

{

   215

    // current time

   216

    time_t now = time(NULL);

   217

   218

    if (queue->timer < now + IRC_QUEUE_WINDOW) {

   219

        // timer is OK, send directly

   220

        return irc_queue_send_direct(queue, line);

   221

   222

    } else {

   223

        // enqueue for later transmission

   224

        return irc_queue_put(queue, line);

   225

    }

   226

}

   227

   228

void irc_queue_destroy (struct irc_queue *queue)

   229

{

   230

    struct irc_queue_entry *entry, *next;

   231

   232

    // free all entries

   233

    for (entry = TAILQ_FIRST(&queue->list); entry; entry = next) {

   234

        next = TAILQ_NEXT(entry, queue_list);

   235

        free(entry);

   236

    }

   237

   238

    // the event

   239

    event_free(queue->ev);

   240

   241

    // the queue itself

   242

    free(queue);

   243

}

   244