1

#include "irc_queue.h"

     2

#include "log.h"

     3

     4

// XXX: for ev_base

     5

#include "sock_internal.h"

     6

     7

#include <stdlib.h>

     8

#include <string.h>

     9

#include <time.h>

    10

    11

/**

    12

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

    13

 * time if it's fallen behind.

    14

 */

    15

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

    16

{

    17

    time_t now = time(NULL);

    18

    err_t err;

    19

    int ret;

    20

    21

    log_debug("%s", line_buf);

    22

    23

    // XXX: output buffering?

    24

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

    25

        return (err = -ret);

    26

    27

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

    28

    if (queue->timer < now)

    29

        queue->timer = now;

    30

    31

    // apply penalty

    32

    queue->timer += IRC_QUEUE_PENALTY;

    33

    34

    // ok

    35

    return SUCCESS;

    36

}

    37

    38

/**

    39

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

    40

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

    41

 *

    42

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

    43

 */

    44

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

    45

{

    46

    err_t err;

    47

    48

    // pop the next entry?

    49

    if (next == NULL) {

    50

        // take it from the head of the list

    51

        next = TAILQ_FIRST(&queue->list);

    52

    53

        // nothing to do?

    54

        if (!next)

    55

            return SUCCESS;

    56

    57

        // and then remove it

    58

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

    59

    }

    60

    61

    // send it

    62

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

    63

        goto error;

    64

    65

    // ok, release it

    66

    free(next);

    67

    68

    // done

    69

    return SUCCESS;

    70

    71

error:

    72

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

    73

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

    74

    75

    return err;    

    76

}

    77

    78

/**

    79

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

    80

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

    81

 */

    82

static err_t irc_queue_schedule (struct irc_queue *queue)

    83

{

    84

    time_t now = time(NULL);

    85

    struct timeval tv;

    86

    87

    // calculate the timeout

    88

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

    89

    90

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

    91

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

    92

    tv.tv_usec = 0;

    93

    94

    // setup the timer event

    95

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

    96

        return ERR_EVENT_ADD;

    97

    98

    // ok

    99

    return SUCCESS;

   100

}

   101

   102

/**

   103

 * Our timer callback

   104

 */

   105

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

   106

{

   107

    time_t now = time(NULL);

   108

    struct irc_queue *queue = arg;

   109

    err_t err;

   110

   111

    (void) fd;

   112

    (void) what;

   113

   114

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

   115

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

   116

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

   117

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

   118

            break;

   119

        }

   120

    }

   121

   122

    // reschedule if needed

   123

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

   124

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

   125

            log_err(err, "irc_queue_scheulde");

   126

        }

   127

    }

   128

}

   129

   130

err_t irc_queue_create (struct irc_queue **queue_ptr, struct line_proto *lp, struct error_info *err)

   131

{

   132

    struct irc_queue *queue;

   133

   134

    // alloc

   135

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

   136

        return SET_ERROR(err, ERR_CALLOC);

   137

   138

    // create the timer event

   139

    // XXX: using the sock module ev_base

   140

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

   141

        JUMP_SET_ERROR(err, ERR_EVENT_NEW);

   142

   143

    // initialize

   144

    queue->lp = lp;

   145

    queue->timer = time(NULL);

   146

    TAILQ_INIT(&queue->list); 

   147

   148

    // ok

   149

    *queue_ptr = queue;

   150

   151

    return SUCCESS;

   152

   153

error:

   154

    // cleanup

   155

    free(queue);

   156

   157

    return ERROR_CODE(err);    

   158

}

   159

   160

/**

   161

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

   162

 */

   163

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

   164

{

   165

    char line_buf[IRC_LINE_MAX + 2];

   166

    err_t err;

   167

   168

    // format

   169

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

   170

        return err;

   171

   172

    // add CRLF

   173

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

   174

   175

    // send

   176

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

   177

    return irc_queue_send_buf(queue, line_buf);

   178

}

   179

   180

/**

   181

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

   182

 */

   183

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

   184

{

   185

    struct irc_queue_entry *entry;

   186

    err_t err;

   187

   188

    // alloc

   189

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

   190

        return ERR_CALLOC;

   191

   192

    // format

   193

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

   194

        goto error;

   195

   196

    // add CRLF

   197

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

   198

   199

    // then re-schedule the queue

   200

    if ((err = irc_queue_schedule(queue)))

   201

        goto error;

   202

   203

    // append to end of list

   204

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

   205

   206

    // ok

   207

    return SUCCESS;

   208

   209

error:

   210

    // cleanup

   211

    free(entry);

   212

   213

    return err;    

   214

}

   215

   216

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

   217

{

   218

    // current time

   219

    time_t now = time(NULL);

   220

   221

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

   222

        // timer is OK, send directly

   223

        return irc_queue_send_direct(queue, line);

   224

   225

    } else {

   226

        // enqueue for later transmission

   227

        return irc_queue_put(queue, line);

   228

    }

   229

}

   230

   231

void irc_queue_destroy (struct irc_queue *queue)

   232

{

   233

    struct irc_queue_entry *entry, *next;

   234

   235

    // free all entries

   236

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

   237

        next = TAILQ_NEXT(entry, queue_list);

   238

        free(entry);

   239

    }

   240

   241

    // the event

   242

    event_free(queue->ev);

   243

   244

    // the queue itself

   245

    free(queue);

   246

}

   247