<commit_before>/*
 * Copyright (c) 2015 <NAME> (<EMAIL>)
 *
 * This is free software: you can redistribute it and/or modify
 * it under the terms of the Artistic License 2.0 as published by
 * The Perl Foundation.
 *
 * This source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * Artistic License 2.0 for more details.
 *
 * You should have received a copy of the Artistic License 2.0
 * along the source as a COPYING file. If not, obtain it from
 * http://www.perlfoundation.org/artistic_license_2_0.
 */

#include <stdio.h>
#include <stdlib.h>

#include <osc_stream_private.h>

static void
_tcp_prefix_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
	uv_tcp_t *socket = (uv_tcp_t *)handle;
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;

	buf->base = cb->buf;
	buf->len = tcp->nchunk < OSC_STREAM_BUF_SIZE ? tcp->nchunk : OSC_STREAM_BUF_SIZE;
}

static void
_tcp_prefix_recv_cb(uv_stream_t *socket, ssize_t nread, const uv_buf_t *buf)
{
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;
	osc_stream_tcp_tx_t *tx = (void *)socket - offsetof(osc_stream_tcp_tx_t, socket);

	if(nread > 0)
	{
		if(nread == sizeof(int32_t))
			tcp->nchunk = ntohl(*(int32_t *)buf->base);
		else if(nread == tcp->nchunk)
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)buf->base, nread, cb->data);
			tcp->nchunk = sizeof(int32_t);
		}
		else // nread != sizeof(int32_t) && nread != nchunk
		{
			//FIXME what should we do here?
			tcp->nchunk = sizeof(int32_t);
			fprintf(stderr, "_tcp_prefix_recv_cb: TCP packet size not matching\n");
		}
	}
	else if (nread < 0)
	{
		if(nread == UV_EOF)
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)disconnect_msg, sizeof(disconnect_msg), cb->data);
		}
		else
			fprintf(stderr, "_tcp_prefix_recv_cb: %s\n", uv_err_name(nread));

		int err;
		if((err = uv_read_stop(socket)))
			fprintf(stderr, "uv_read_stop: %s\n", uv_err_name(err));
		uv_close((uv_handle_t *)socket, NULL);
		uv_cancel((uv_req_t *)&tx->req);

		tcp->tx = inlist_remove(tcp->tx, INLIST_GET(tx));
		free(tx);

		//TODO try to reconnect?
	}
	else // nread == 0
		;
}

static void
_tcp_slip_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
	uv_tcp_t *socket = (uv_tcp_t *)handle;
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;

	buf->base = cb->buf;
	buf->base += tcp->nchunk; // is there remaining chunk from last call?
	buf->len = OSC_STREAM_BUF_SIZE - tcp->nchunk;
}

static void
_tcp_slip_recv_cb(uv_stream_t *socket, ssize_t nread, const uv_buf_t *buf)
{
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;
	osc_stream_tcp_tx_t *tx = (void *)socket - offsetof(osc_stream_tcp_tx_t, socket);
	
	if(nread > 0)
	{
		char *ptr = cb->buf;
		nread += tcp->nchunk; // is there remaining chunk from last call?
		size_t size;
		size_t parsed;
		while( (parsed = slip_decode(ptr, nread, &size)) && (nread > 0) )
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)ptr, size, cb->data);
			ptr += parsed;
			nread -= parsed;
		}
		if(nread > 0) // is there remaining chunk for next call?
		{
			memmove(cb->buf, ptr, nread);
			tcp->nchunk = nread;
		}
		else
			tcp->nchunk = 0;
	}
	else if (nread < 0)
	{
		if(nread == UV_EOF)
		{<commit_msg>handle TCP timeout<commit_after>/*
 * Copyright (c) 2015 <NAME> (<EMAIL>)
 *
 * This is free software: you can redistribute it and/or modify
 * it under the terms of the Artistic License 2.0 as published by
 * The Perl Foundation.
 *
 * This source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * Artistic License 2.0 for more details.
 *
 * You should have received a copy of the Artistic License 2.0
 * along the source as a COPYING file. If not, obtain it from
 * http://www.perlfoundation.org/artistic_license_2_0.
 */

#include <stdio.h>
#include <stdlib.h>

#include <osc_stream_private.h>

static void
_tcp_prefix_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
	uv_tcp_t *socket = (uv_tcp_t *)handle;
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;

	buf->base = cb->buf;
	buf->len = tcp->nchunk < OSC_STREAM_BUF_SIZE ? tcp->nchunk : OSC_STREAM_BUF_SIZE;
}

static void
_tcp_prefix_recv_cb(uv_stream_t *socket, ssize_t nread, const uv_buf_t *buf)
{
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;
	osc_stream_tcp_tx_t *tx = (void *)socket - offsetof(osc_stream_tcp_tx_t, socket);

	if(nread > 0)
	{
		if(nread == sizeof(int32_t))
			tcp->nchunk = ntohl(*(int32_t *)buf->base);
		else if(nread == tcp->nchunk)
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)buf->base, nread, cb->data);
			tcp->nchunk = sizeof(int32_t);
		}
		else // nread != sizeof(int32_t) && nread != nchunk
		{
			//FIXME what should we do here?
			tcp->nchunk = sizeof(int32_t);
			fprintf(stderr, "_tcp_prefix_recv_cb: TCP packet size not matching\n");
		}
	}
	else if (nread < 0)
	{
		if( (nread == UV_EOF) || (nread == UV_ETIMEDOUT) )
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)disconnect_msg, sizeof(disconnect_msg), cb->data);
		}
		else
			fprintf(stderr, "_tcp_prefix_recv_cb: %s\n", uv_err_name(nread));

		int err;
		if((err = uv_read_stop(socket)))
			fprintf(stderr, "uv_read_stop: %s\n", uv_err_name(err));
		uv_close((uv_handle_t *)socket, NULL);
		uv_cancel((uv_req_t *)&tx->req);

		tcp->tx = inlist_remove(tcp->tx, INLIST_GET(tx));
		free(tx);

		//TODO try to reconnect?
	}
	else // nread == 0
		;
}

static void
_tcp_slip_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
{
	uv_tcp_t *socket = (uv_tcp_t *)handle;
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;

	buf->base = cb->buf;
	buf->base += tcp->nchunk; // is there remaining chunk from last call?
	buf->len = OSC_STREAM_BUF_SIZE - tcp->nchunk;
}

static void
_tcp_slip_recv_cb(uv_stream_t *socket, ssize_t nread, const uv_buf_t *buf)
{
	osc_stream_tcp_t *tcp = socket->data;
	osc_stream_t *stream = (void *)tcp - offsetof(osc_stream_t, payload.tcp);
	osc_stream_cb_t *cb = &stream->cb;
	osc_stream_tcp_tx_t *tx = (void *)socket - offsetof(osc_stream_tcp_tx_t, socket);
	
	if(nread > 0)
	{
		char *ptr = cb->buf;
		nread += tcp->nchunk; // is there remaining chunk from last call?
		size_t size;
		size_t parsed;
		while( (parsed = slip_decode(ptr, nread, &size)) && (nread > 0) )
		{
			if(cb->recv)
				cb->recv(stream, (osc_data_t *)ptr, size, cb->data);
			ptr += parsed;
			nread -= parsed;
		}
		if(nread > 0) // is there remaining chunk for next call?
		{
			memmove(cb->buf, ptr, nread);
			tcp->nchunk = nread;
		}
		else
			tcp->nchunk = 0;
	}
	else if (nread < 0)
	{
		if( (nread == UV_EOF) || (nread == UV_ETIMEDOUT) )
		{
{'fraction_non_alphanumeric': 0.14287618286018927, 'fraction_numerical': 0.007996801279488205, 'mean_word_length': 4.0531986531986535, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2921930', 'n_tokens_mistral': 3161, 'n_tokens_neox': 2948, 'n_words': 724}