<commit_before>/*
 * =====================================================================================
 *
 *       Filename:  fnet_buff.c
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  11/18/2011 10:43:52
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  YOUR NAME (), 
 *        Company:  
 *
 * =====================================================================================
 */

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include "fev_buff.h"
#include "fev.h"
#include "mbuf.h"

#define FEV_BUFF_DEFAULT_SIZE   (1024 * 4)

struct fev_buff {
    int             fd;
    fev_state*      fstate;
    mbuf*           rbuf;
    mbuf*           wbuf;
    fev_buff_read   read_cb;
    fev_buff_error  error_cb;
    void*           arg;
};

static int fev_read(int fd, void* pbuf, size_t len)
{
    do{
		int read_size = read(fd, pbuf, len);

		if ( read_size == -1 )
		{
			if( errno == EINTR )
				continue;
			else if( errno == EAGAIN )
				return 0;   // we can ignore this and simply to return 0
			return -1;
		}
		else if( read_size == 0 )
			return -1;
		else
			return read_size;
	}while(1);
}

static int fev_write(int fd, const char* pbuf, size_t len)
{
    do{
		int bytes = write(fd, pbuf, len);

		if( bytes > 0 )
			return bytes;
		else
		{
			if( errno == EINTR )
				continue;
			else if( errno == EAGAIN )
				return 0;
			else
				return -1;
		}
	}while(1);
}

static void evbuff_read(fev_state* fev, int fd, int mask, void* arg)
{
    fev_buff* evbuff = (fev_buff*)arg;
    if( evbuff->read_cb ) 
        evbuff->read_cb(fev, evbuff, evbuff->arg);
}

static void evbuff_write(fev_state* fev, int fd, int mask, void* arg)
{
    fev_buff* evbuff = (fev_buff*)arg;

    do{
        int buf_len = mbuf_used(evbuff->wbuf);
        if ( buf_len == 0 ) {
            fev_del_event(evbuff->fstate, evbuff->fd, FEV_WRITE);
            break;
        }

        int bytes = fev_write(evbuff->fd, mbuf_get_head(evbuff->wbuf), buf_len);
        if( bytes > 0 ) {
            mbuf_head_seek(evbuff->wbuf, bytes);
        }
        else if( bytes == 0 ) { // EAGAIN
            break;
        }  
        else {
            if ( evbuff->error_cb ) 
                evbuff->error_cb(fev, evbuff, evbuff->arg);
            return;
        }

    }while(1);
}

fev_buff*	fevbuff_new(
                fev_state* fev,
                int fd,
				fev_buff_read read_cb,		// call when the fd can read
				fev_buff_error error_cb,	// call when socket has error
				void* arg)                  // user argument
{
    if( !fev ) return NULL;

    fev_buff* evbuff = (fev_buff*)malloc(sizeof(fev_buff));
    if( !evbuff ) return NULL;

    evbuff->fstate = fev;
    evbuff->fd = fd;
    evbuff->arg = arg;
    evbuff->read_cb = read_cb;
    evbuff->error_cb = error_cb;

    evbuff->rbuf = mbuf_create(FEV_BUFF_DEFAULT_SIZE);
    if( !evbuff->rbuf ) {
        free(evbuff);
        return NULL;
    }

    evbuff->rbuf = mbuf_create(FEV_BUFF_DEFAULT_SIZE);
    if( !evbuff->wbuf ) {
        mbuf_delete(evbuff->rbuf);
        free(evbuff);
        return NULL;
    }

    if ( fev_reg_event(fev, fd, FEV_READ, evbuff_read, evbuff_write, evbuff) ) {
        mbuf_delete(evbuff->rbuf);
        mbuf_delete(evbuff->wbuf);
        free(evbuff);
        return NULL;
    }

    return evbuff;
}

int  fevbuff_destroy(fev_buff* evbuff)
{
    if( !evbuff ) return -1;

    int fd = evbuff->fd;
    int mask = FEV_READ | FEV_WRITE;
<commit_msg>add test case of fevbuff<commit_after>/*
 * =====================================================================================
 *
 *       Filename:  fnet_buff.c
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  11/18/2011 10:43:52
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  YOUR NAME (), 
 *        Company:  
 *
 * =====================================================================================
 */

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include "fev_buff.h"
#include "fev.h"
#include "mbuf.h"

#define FEV_BUFF_DEFAULT_SIZE   (1024 * 4)

struct fev_buff {
    int             fd;
    fev_state*      fstate;
    mbuf*           rbuf;
    mbuf*           wbuf;
    fev_buff_read   read_cb;
    fev_buff_error  error_cb;
    void*           arg;
};

static int fev_read(int fd, void* pbuf, size_t len)
{
    do{
		int read_size = read(fd, pbuf, len);

		if ( read_size == -1 )
		{
			if( errno == EINTR )
				continue;
			else if( errno == EAGAIN )
				return 0;   // we can ignore this and simply to return 0
			return -1;
		}
		else if( read_size == 0 )
			return -1;
		else
			return read_size;
	}while(1);
}

static int fev_write(int fd, const char* pbuf, size_t len)
{
    do{
		int bytes = write(fd, pbuf, len);

		if( bytes > 0 )
			return bytes;
		else
		{
			if( errno == EINTR )
				continue;
			else if( errno == EAGAIN )
				return 0;
			else
				return -1;
		}
	}while(1);
}

static void evbuff_read(fev_state* fev, int fd, int mask, void* arg)
{
    fev_buff* evbuff = (fev_buff*)arg;
    if( evbuff->read_cb ) 
        evbuff->read_cb(fev, evbuff, evbuff->arg);
}

static void evbuff_write(fev_state* fev, int fd, int mask, void* arg)
{
    fev_buff* evbuff = (fev_buff*)arg;

    do{
        int buf_len = mbuf_used(evbuff->wbuf);
        if ( buf_len == 0 ) {
            fev_del_event(evbuff->fstate, evbuff->fd, FEV_WRITE);
            break;
        }

        int bytes = fev_write(evbuff->fd, mbuf_get_head(evbuff->wbuf), buf_len);
        if( bytes > 0 ) {
            mbuf_head_seek(evbuff->wbuf, bytes);
        }
        else if( bytes == 0 ) { // EAGAIN
            break;
        }  
        else {
            if ( evbuff->error_cb ) 
                evbuff->error_cb(fev, evbuff, evbuff->arg);
            return;
        }

    }while(1);
}

fev_buff*	fevbuff_new(
                fev_state* fev,
                int fd,
				fev_buff_read read_cb,		// call when the fd can read
				fev_buff_error error_cb,	// call when socket has error
				void* arg)                  // user argument
{
    if( !fev ) return NULL;

    fev_buff* evbuff = (fev_buff*)malloc(sizeof(fev_buff));
    if( !evbuff ) return NULL;

    evbuff->fstate = fev;
    evbuff->fd = fd;
    evbuff->arg = arg;
    evbuff->read_cb = read_cb;
    evbuff->error_cb = error_cb;

    evbuff->rbuf = mbuf_create(FEV_BUFF_DEFAULT_SIZE);
    if( !evbuff->rbuf ) {
        free(evbuff);
        return NULL;
    }

    evbuff->wbuf = mbuf_create(FEV_BUFF_DEFAULT_SIZE);
    if( !evbuff->wbuf ) {
        mbuf_delete(evbuff->rbuf);
        free(evbuff);
        return NULL;
    }

    if ( fev_reg_event(fev, fd, FEV_READ, evbuff_read, evbuff_write, evbuff) ) {
        mbuf_delete(evbuff->rbuf);
        mbuf_delete(evbuff->wbuf);
        free(evbuff);
        return NULL;
    }

    return evbuff;
}

int  fevbuff_destroy(fev_buff* evbuff)
{
    if( !evbuff ) return -1;

    int fd = evbuff->fd;
    int mask = FEV_READ | FEV_WRITE;

{'fraction_non_alphanumeric': 0.1676833593417506, 'fraction_numerical': 0.010497942970634132, 'mean_word_length': 2.137516688918558, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6908795', 'n_tokens_mistral': 2803, 'n_tokens_neox': 2562, 'n_words': 582}