bgp-sink.cc

Go to the documentation of this file.

#include "bgp-sink.h"
#include <stdlib.h>
#include <string.h>
#include <arpa/inet.h>
#define BGP_SINK_DEFAULT_BUFSZ 65536

namespace libbgp {

BgpSink::BgpSink(bool use_4b_asn) {
    this->buffer_size = BGP_SINK_DEFAULT_BUFSZ;
    this->buffer = (uint8_t *) malloc(buffer_size);
    this->use_4b_asn = use_4b_asn;
    this->logger = NULL;
    offset_start = offset_end = 0;
}

BgpSink::~BgpSink() {
    free(buffer);
}

ssize_t BgpSink::fill(const uint8_t *buffer, size_t len) {
    std::lock_guard<std::recursive_mutex> lock(mutex);
    if (len > buffer_size) {
        if (logger) logger->log(ERROR, "BgpSink::fill: buffer length (%d) > sink size (%d).\n", len, buffer_size);
        return -1;
    }

    // first try settle
    if (offset_end + len > buffer_size) settle();

    // if still too small, expand
    while (offset_end + len > buffer_size) {
        expand();
    }

    memcpy(this->buffer + offset_end, buffer, len);
    offset_end += len;

    return len;
}

ssize_t BgpSink::pour(BgpPacket **pkt) {
    std::lock_guard<std::recursive_mutex> lock(mutex);

    uint8_t *cur = this->buffer + offset_start;

    if (offset_end - offset_start < 19) return 0;
    if (memcmp(cur, "\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff", 16) != 0) {
        if (logger) logger->log(ERROR, "BgpSink::pour: invalid BGP marker.\n");
        return -2;
    }

    uint16_t field_len = ntohs(*(uint16_t *) (cur + 16));

    if (field_len < 19 || field_len > 4096) {
        if (logger) logger->log(ERROR, "BgpSink::pourPtr: invalid BGP packet length (%d).\n", field_len);
        return -2;
    }

    ssize_t bytes = getBytesInSink();
    if (field_len > bytes) return 0; // incomplete packet, wait for more.

    offset_start += field_len;

    BgpPacket *new_pkt = new BgpPacket(logger, use_4b_asn);
    ssize_t par_ret = new_pkt->parse(cur, field_len);

    *pkt = new_pkt;

    if (par_ret < 0) return -1;

    if (par_ret != field_len) throw "bad_packet";

    return par_ret;
}

void BgpSink::settle() {
    if (offset_start > 0) {
        if (offset_start == offset_end) offset_start = offset_end = 0;
        else memmove(buffer, buffer + offset_start, offset_end - offset_start);
    }
}

void BgpSink::expand() {
    size_t new_buf_sz = buffer_size * 2;
    size_t content_sz = getBytesInSink();
    uint8_t *new_buffer = (uint8_t *) malloc(new_buf_sz);
    memcpy(new_buffer, buffer + offset_start, content_sz);
    free(buffer);
    buffer = new_buffer;
    buffer_size = new_buf_sz;
    offset_start = 0;
    offset_end = content_sz;
    if (logger) logger->log(DEBUG, "BgpSink::expand: expanded size to %zu\n", buffer_size);
}

void BgpSink::drain() {
    std::lock_guard<std::recursive_mutex> lock(mutex);
    offset_end = offset_start = 0;
}

size_t BgpSink::getBytesInSink() const {
    return offset_end - offset_start;
}

void BgpSink::setLogger(BgpLogHandler *logger) {
    this->logger = logger;
}

}