<commit_before>/*****************************************************************************
 * Copyright (C) 2011-2013 <NAME>
 *
 * Author: <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library 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 GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 *****************************************************************************/

#include <stdlib.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <string>

#include "output.h"
#include "log.h"
#define CLASS_MODULE "out"

#define dprintf(fmt, arg...) __dprintf(DBG_OUTPUT, fmt, ##arg)

#define DOUBLE_BUFFER 0
#define PREVENT_RBUF_DEADLOCK 0
#define NON_BLOCKING_TCP_SEND 1

#define HTTP_200_OK  "HTTP/1.1 200 OK"
#define CONTENT_TYPE "Content-type: "
#define TEXT_HTML    "text/html"
#define TEXT_PLAIN   "text/plain"
#define OCTET_STREAM "application/octet-stream"
#define ENC_CHUNKED  "Transfer-Encoding: chunked"
#define CONN_CLOSE   "Connection: close"
#define CRLF         "\r\n"

static char http_conn_close[] =
	CONN_CLOSE
	CRLF
	CRLF;

static const char * __http_response(const char *mimetype)
{
	std::string str;
	str.clear();

	str.append(HTTP_200_OK
		   CRLF);
	if (mimetype) {
		str.append(CONTENT_TYPE);
		str.append(mimetype);
		str.append(CRLF
			   ENC_CHUNKED
			   CRLF);
	}
	str.append(CRLF);

	return str.c_str();
}

const char * http_response(enum output_mimetype mimetype)
{
	const char *str;

	switch (mimetype) {
	default:
	case MIMETYPE_OCTET_STREAM:
		str = OCTET_STREAM;
		break;
	case MIMETYPE_TEXT_PLAIN:
		str = TEXT_PLAIN;
		break;
	case MIMETYPE_TEXT_HTML:
		str = TEXT_HTML;
		break;
	case MIMETYPE_NONE:
		str = NULL;
		break;
	}

	return __http_response(str);
}

ssize_t socket_send(int sockfd, const void *buf, size_t len, int flags,
		    const struct sockaddr *dest_addr, socklen_t addrlen)
{
	if (sockfd < 0)
		return sockfd;
	int ret = 0;
	while (0 >= ret) {
		fd_set fds;
		FD_ZERO(&fds);
		FD_SET(sockfd, &fds);

		struct timeval sel_timeout = {0, 10000};

		ret = select(sockfd + 1, NULL, &fds, NULL, &sel_timeout);
		if (ret == -1) {
			perror("error sending data to socket!");
			/*
			switch (errno) {
			case EBADF:  // invalid file descriptor
			case EINTR:  // signal caught
			case EINVAL: // negative fd or invalid timeout
			case ENOMEM: // unable to allocate memory
			}
			*/
			break;
		}
	}
	return (ret > 0) ?
#if 0
		((dest_addr) && (addrlen)) ?
			sendto(sockfd, buf, len, flags, dest_addr, addrlen) :
			send(sockfd, buf, len, flags) :
#else
		sendto(sockfd, buf, len, flags|MSG_NOSIGNAL, dest_addr, addrlen) :
#endif
		ret;
}

static inline ssize_t stream_crlf(int socket)
{
	return socket_send(socket, CRLF, 2, 0);
}

int stream_http_chunk(int socket, const uint8_t *buf, size_t length, const bool send_zero_length)
{
	if (socket < 0)
		return socket;
#if DBG
	dprintf("(length:%d)", (int)length);
#endif
	if ((length) || (send_zero_length)) {
		int ret = 0;
		char sz[7] = { 0 };
		sprintf(sz, "%x\r\n", (unsigned int)length);

		ret = socket_send(socket, sz, strlen(sz), 0);
		if (ret < 0)
			return ret;
#if 0
		ret = stream_crlf(socket);
		if (ret < 0)
			return ret;
#endif
		if (length) {
			ret = socket_send(socket, buf, length, 0);
			if (ret < 0)
				return ret;

			ret = stream_crlf(socket);
			if (ret < 0)
				return ret;
		}
	}
	return 0;
}


output_stream::output_stream()
  : f_kill_thread(false)
  , f_streaming(false)
  , sock(-1)
  , mimetype(MIMETYPE_OCTET_STREAM)
  , ringbuffer()
  , stream_method(OUTPUT_STREAM_UDP)
  , count_in(0)
  , count_out(0)
  , stream_cb(NULL)
  , stream_cb_priv(NULL)
{
	dprintf("()");
	memset(&ringbuffer, 0, sizeof(ringbuffer));
	memset(&name, 0, sizeof(name));
}

output_stream::~output_stream()
{
	dprintf("(%d)", sock);

	stop();

	dprintf("(stream) %lu packets in, %lu packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
}

#if 1
output_stream::output_stream(const output_stream&)
{
	dprintf("(copy)");
	memset(&ringbuffer, 0, sizeof(ringbuffer));
	f_kill_thread = false;
	f_streaming = false;
	stream_cb = NULL;
	stream_cb_priv = NULL;
	count_in = 0;
	count_out = 0;
	sock = -1;
	mimetype = MIMETYPE_OCTET_STREAM;
	memset(&name, 0, sizeof(name));
}

output_stream& output_stream::operator= (const output_stream& cSource)
{
	dprintf("(operator=)");

	if (this == &cSource)
		return *this;

	memset(&ringbuffer, 0, sizeof(ringbuffer));
	f_kill_thread = false;
	f_streaming = false;
	stream_cb = NULL;
	stream_cb_priv = NULL;
	count_in = 0;
	count_out = 0;
	sock = -1;
	mimetype = MIMETYPE_OCTET_STREAM;
	memset(&name, 0, sizeof(name));

	return *this;
}
#endif

//static
void* output_stream::output_stream_thread(void *p_this)
{
	return static_cast<output_stream*>(p_this)->output_stream_thread();
}

#define OUTPUT_STREAM_PACKET_SIZE ((stream_method == OUTPUT_STREAM_UDP) ? 188*7 : 188*21)
void* output_stream::output_stream_thread()
{
	uint8_t *data = NULL;
	int buf_size, ret = 0;

	dprintf("(%d)", sock);
#if 1
	switch (stream_method) {
	case OUTPUT_STREAM_HTTP:
		const char *str = http_response(mimetype);
		ret = socket_send(sock, str, strlen(str), 0);
		break;
	}
	if (ret < 0) {
		perror("stream header failed");
		goto fail;
	}

#endif
	f_streaming = true;

	/* push data from output_stream buffer to target */
	while (!f_kill_thread) {

		buf_size = ringbuffer.get_size();
		if (buf_size < OUTPUT_STREAM_PACKET_SIZE) {
			usleep(1000);
			continue;
		}

		buf_size = ringbuffer.get_read_ptr((void**)&data, OUTPUT_STREAM_PACKET_SIZE);
		buf_size /= 188;
		buf_size *= 188;

#if PREVENT_RBUF_DEADLOCK
		{
			uint8_t newdata[buf_size];
			memcpy(newdata, data, buf_size);
			ringbuffer.put_read_ptr(buf_size);
			stream(newdata, buf_size);
		}
#else
		stream(data, buf_size);

		ringbuffer.put_read_ptr(buf_size);
#endif
		count_out += buf_size;
#if 0
		dprintf("(thread-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
	}

	f_streaming = false;
#if 0
	switch (stream_method) {
	case OUTPUT_STREAM_HTTP:
		if (stream_http_chunk(sock, (uint8_t *)"", 0, true) < 0)
			perror("stream empty http chunk failed");
		else if (socket_send(sock, http_conn_close, strlen(http_conn_close), 0) < 0)
			perror("stream http connection close failed");
		break;
	}
#endif
fail:
	close_file();
	pthread_exit(NULL);
}

int output_stream::start()
{
	if (f_streaming) {
		dprintf("(%d) already streaming", sock);
		return 0;
	}
	if ((sock < 0) && (stream_method != OUTPUT_STREAM_FUNC))
		return sock;

	dprintf("(%d)", sock);

	ringbuffer.set_capacity(OUTPUT_STREAM_BUF_SIZE);

	int ret = pthread_create(&h_thread, NULL, output_stream_thread, this);
	if (0 != ret)
		perror("pthread_create() failed");

	return ret;
}

void output_stream::stop()
{
	dprintf("(%d)", sock);

	stop_without_wait();

	while (f_streaming)
		usleep(20*1000);

	return;
}

bool output_stream::check()
{
	bool ret = is_streaming();
	if (!ret)
		dprintf("(%d: %s) not streaming!", sock, name);
	else {
		dprintf("(%d: %s) %s %lu in, %lu out",
			sock, name,
			(stream_method == OUTPUT_STREAM_UDP) ? "UDP" :
			(stream_method == OUTPUT_STREAM_TCP) ? "TCP" :
			(stream_method == OUTPUT_STREAM_HTTP) ? "HTTP" :
			(stream_method == OUTPUT_STREAM_FILE) ? "FILE" :
			(stream_method == OUTPUT_STREAM_FUNC) ? "FUNC" : "UNKNOWN",
			count_in / 188, count_out / 188);
	}
	ringbuffer.check();

	return ret;
}

bool output_stream::push(uint8_t* p_data, int size)
{
	/* push data into output_stream buffer */
	if (!ringbuffer.write(p_data, size))
		while (size >= 188)
			if (ringbuffer.write(p_data, 188)) {
				p_data += 188;
				size -= 188;
				count_in += 188;
			} else {
				fprintf(stderr, "%s> FAILED: %d bytes dropped\n", __func__, size);
#if 0
				dprintf("(push-false-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
				return false;
			}
	else count_in += size;
#if 0
	dprintf("(push-true-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
	return true;
}

int output_stream::stream(uint8_t* p_data, int size)
{
	int ret = -1;

	/* stream data to target */
	switch (stream_method) {
	case OUTPUT_STREAM_UDP:
		ret = socket_send(sock, p_data, size, 0, (struct sockaddr*) &ip_addr, sizeof(ip_addr));
		break;
	case OUTPUT_STREAM_TCP:
		ret = socket_send(sock, p_data, size, 0);
		if (ret < 0) {
			stop_without_wait();
			perror("tcp streaming failed");
		}
		break;
	case OUTPUT_STREAM_FILE:
		ret = write(sock, p_data, size);
		if (ret < 0) {
			stop_without_wait();
			perror("file streaming failed");
		}
		break;<commit_msg>output: dont stream when there is no data<commit_after>/*****************************************************************************
 * Copyright (C) 2011-2013 <NAME>
 *
 * Author: <NAME> <<EMAIL>>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library 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 GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 *****************************************************************************/

#include <stdlib.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <string>

#include "output.h"
#include "log.h"
#define CLASS_MODULE "out"

#define dprintf(fmt, arg...) __dprintf(DBG_OUTPUT, fmt, ##arg)

#define DOUBLE_BUFFER 0
#define PREVENT_RBUF_DEADLOCK 0
#define NON_BLOCKING_TCP_SEND 1

#define HTTP_200_OK  "HTTP/1.1 200 OK"
#define CONTENT_TYPE "Content-type: "
#define TEXT_HTML    "text/html"
#define TEXT_PLAIN   "text/plain"
#define OCTET_STREAM "application/octet-stream"
#define ENC_CHUNKED  "Transfer-Encoding: chunked"
#define CONN_CLOSE   "Connection: close"
#define CRLF         "\r\n"

static char http_conn_close[] =
	CONN_CLOSE
	CRLF
	CRLF;

static const char * __http_response(const char *mimetype)
{
	std::string str;
	str.clear();

	str.append(HTTP_200_OK
		   CRLF);
	if (mimetype) {
		str.append(CONTENT_TYPE);
		str.append(mimetype);
		str.append(CRLF
			   ENC_CHUNKED
			   CRLF);
	}
	str.append(CRLF);

	return str.c_str();
}

const char * http_response(enum output_mimetype mimetype)
{
	const char *str;

	switch (mimetype) {
	default:
	case MIMETYPE_OCTET_STREAM:
		str = OCTET_STREAM;
		break;
	case MIMETYPE_TEXT_PLAIN:
		str = TEXT_PLAIN;
		break;
	case MIMETYPE_TEXT_HTML:
		str = TEXT_HTML;
		break;
	case MIMETYPE_NONE:
		str = NULL;
		break;
	}

	return __http_response(str);
}

ssize_t socket_send(int sockfd, const void *buf, size_t len, int flags,
		    const struct sockaddr *dest_addr, socklen_t addrlen)
{
	if (sockfd < 0)
		return sockfd;
	int ret = 0;
	while (0 >= ret) {
		fd_set fds;
		FD_ZERO(&fds);
		FD_SET(sockfd, &fds);

		struct timeval sel_timeout = {0, 10000};

		ret = select(sockfd + 1, NULL, &fds, NULL, &sel_timeout);
		if (ret == -1) {
			perror("error sending data to socket!");
			/*
			switch (errno) {
			case EBADF:  // invalid file descriptor
			case EINTR:  // signal caught
			case EINVAL: // negative fd or invalid timeout
			case ENOMEM: // unable to allocate memory
			}
			*/
			break;
		}
	}
	return (ret > 0) ?
#if 0
		((dest_addr) && (addrlen)) ?
			sendto(sockfd, buf, len, flags, dest_addr, addrlen) :
			send(sockfd, buf, len, flags) :
#else
		sendto(sockfd, buf, len, flags|MSG_NOSIGNAL, dest_addr, addrlen) :
#endif
		ret;
}

static inline ssize_t stream_crlf(int socket)
{
	return socket_send(socket, CRLF, 2, 0);
}

int stream_http_chunk(int socket, const uint8_t *buf, size_t length, const bool send_zero_length)
{
	if (socket < 0)
		return socket;
#if DBG
	dprintf("(length:%d)", (int)length);
#endif
	if ((length) || (send_zero_length)) {
		int ret = 0;
		char sz[7] = { 0 };
		sprintf(sz, "%x\r\n", (unsigned int)length);

		ret = socket_send(socket, sz, strlen(sz), 0);
		if (ret < 0)
			return ret;
#if 0
		ret = stream_crlf(socket);
		if (ret < 0)
			return ret;
#endif
		if (length) {
			ret = socket_send(socket, buf, length, 0);
			if (ret < 0)
				return ret;

			ret = stream_crlf(socket);
			if (ret < 0)
				return ret;
		}
	}
	return 0;
}


output_stream::output_stream()
  : f_kill_thread(false)
  , f_streaming(false)
  , sock(-1)
  , mimetype(MIMETYPE_OCTET_STREAM)
  , ringbuffer()
  , stream_method(OUTPUT_STREAM_UDP)
  , count_in(0)
  , count_out(0)
  , stream_cb(NULL)
  , stream_cb_priv(NULL)
{
	dprintf("()");
	memset(&ringbuffer, 0, sizeof(ringbuffer));
	memset(&name, 0, sizeof(name));
}

output_stream::~output_stream()
{
	dprintf("(%d)", sock);

	stop();

	dprintf("(stream) %lu packets in, %lu packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
}

#if 1
output_stream::output_stream(const output_stream&)
{
	dprintf("(copy)");
	memset(&ringbuffer, 0, sizeof(ringbuffer));
	f_kill_thread = false;
	f_streaming = false;
	stream_cb = NULL;
	stream_cb_priv = NULL;
	count_in = 0;
	count_out = 0;
	sock = -1;
	mimetype = MIMETYPE_OCTET_STREAM;
	memset(&name, 0, sizeof(name));
}

output_stream& output_stream::operator= (const output_stream& cSource)
{
	dprintf("(operator=)");

	if (this == &cSource)
		return *this;

	memset(&ringbuffer, 0, sizeof(ringbuffer));
	f_kill_thread = false;
	f_streaming = false;
	stream_cb = NULL;
	stream_cb_priv = NULL;
	count_in = 0;
	count_out = 0;
	sock = -1;
	mimetype = MIMETYPE_OCTET_STREAM;
	memset(&name, 0, sizeof(name));

	return *this;
}
#endif

//static
void* output_stream::output_stream_thread(void *p_this)
{
	return static_cast<output_stream*>(p_this)->output_stream_thread();
}

#define OUTPUT_STREAM_PACKET_SIZE ((stream_method == OUTPUT_STREAM_UDP) ? 188*7 : 188*21)
void* output_stream::output_stream_thread()
{
	uint8_t *data = NULL;
	int buf_size, ret = 0;

	dprintf("(%d)", sock);
#if 1
	switch (stream_method) {
	case OUTPUT_STREAM_HTTP:
		const char *str = http_response(mimetype);
		ret = socket_send(sock, str, strlen(str), 0);
		break;
	}
	if (ret < 0) {
		perror("stream header failed");
		goto fail;
	}

#endif
	f_streaming = true;

	/* push data from output_stream buffer to target */
	while (!f_kill_thread) {

		buf_size = ringbuffer.get_size();
		if (buf_size < OUTPUT_STREAM_PACKET_SIZE) {
			usleep(1000);
			continue;
		}

		buf_size = ringbuffer.get_read_ptr((void**)&data, OUTPUT_STREAM_PACKET_SIZE);
		buf_size /= 188;
		buf_size *= 188;

#if PREVENT_RBUF_DEADLOCK
		{
			uint8_t newdata[buf_size];
			memcpy(newdata, data, buf_size);
			ringbuffer.put_read_ptr(buf_size);
			stream(newdata, buf_size);
		}
#else
		stream(data, buf_size);

		ringbuffer.put_read_ptr(buf_size);
#endif
		count_out += buf_size;
#if 0
		dprintf("(thread-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
	}

	f_streaming = false;
#if 0
	switch (stream_method) {
	case OUTPUT_STREAM_HTTP:
		if (stream_http_chunk(sock, (uint8_t *)"", 0, true) < 0)
			perror("stream empty http chunk failed");
		else if (socket_send(sock, http_conn_close, strlen(http_conn_close), 0) < 0)
			perror("stream http connection close failed");
		break;
	}
#endif
fail:
	close_file();
	pthread_exit(NULL);
}

int output_stream::start()
{
	if (f_streaming) {
		dprintf("(%d) already streaming", sock);
		return 0;
	}
	if ((sock < 0) && (stream_method != OUTPUT_STREAM_FUNC))
		return sock;

	dprintf("(%d)", sock);

	ringbuffer.set_capacity(OUTPUT_STREAM_BUF_SIZE);

	int ret = pthread_create(&h_thread, NULL, output_stream_thread, this);
	if (0 != ret)
		perror("pthread_create() failed");

	return ret;
}

void output_stream::stop()
{
	dprintf("(%d)", sock);

	stop_without_wait();

	while (f_streaming)
		usleep(20*1000);

	return;
}

bool output_stream::check()
{
	bool ret = is_streaming();
	if (!ret)
		dprintf("(%d: %s) not streaming!", sock, name);
	else {
		dprintf("(%d: %s) %s %lu in, %lu out",
			sock, name,
			(stream_method == OUTPUT_STREAM_UDP) ? "UDP" :
			(stream_method == OUTPUT_STREAM_TCP) ? "TCP" :
			(stream_method == OUTPUT_STREAM_HTTP) ? "HTTP" :
			(stream_method == OUTPUT_STREAM_FILE) ? "FILE" :
			(stream_method == OUTPUT_STREAM_FUNC) ? "FUNC" : "UNKNOWN",
			count_in / 188, count_out / 188);
	}
	ringbuffer.check();

	return ret;
}

bool output_stream::push(uint8_t* p_data, int size)
{
	/* push data into output_stream buffer */
	if (!ringbuffer.write(p_data, size))
		while (size >= 188)
			if (ringbuffer.write(p_data, 188)) {
				p_data += 188;
				size -= 188;
				count_in += 188;
			} else {
				fprintf(stderr, "%s> FAILED: %d bytes dropped\n", __func__, size);
#if 0
				dprintf("(push-false-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
				return false;
			}
	else count_in += size;
#if 0
	dprintf("(push-true-stream) %d packets in, %d packets out, %d packets remain in rbuf", count_in / 188, count_out / 188, ringbuffer.get_size() / 188);
#endif
	return true;
}

int output_stream::stream(uint8_t* p_data, int size)
{
	int ret = -1;

	if ((!p_data) || (!size))
		dprintf("no data to stream!!!");
	/* stream data to target */
	else switch (stream_method) {
	case OUTPUT_STREAM_UDP:
		ret = socket_send(sock, p_data, size, 0, (struct sockaddr*) &ip_addr, sizeof(ip_addr));
		break;
	case OUTPUT_STREAM_TCP:
		ret = socket_send(sock, p_data, size, 0);
		if (ret < 0) {
			stop_without_wait();
			perror("tcp streaming failed");
		}
		break;
	case OUTPUT_STREAM_FILE:
		ret = write(sock, p_data, size);
		if (ret < 0) {
			stop_without_wait();
			perror("file streaming failed");
		}
		break;
{'fraction_non_alphanumeric': 0.15240061404901806, 'fraction_numerical': 0.019691916785771, 'mean_word_length': 4.280044717719396, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 37, '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': '4399364', 'n_tokens_mistral': 7793, 'n_tokens_neox': 7016, 'n_words': 1838}