<commit_before>/* Copyright (c) 2002-2015 Dovecot authors, see the included COPYING file */

#include "lib.h"

#include "ioloop.h"
#include "fd-set-nonblock.h"
#include "net.h"
#include "str.h"
#include "write-full.h"
#include "istream.h"
#include "ostream.h"
#include "process-title.h"
#include "restrict-access.h"
#include "time-util.h"
#include "master-service.h"

#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>

#define OUTBUF_THRESHOLD IO_BLOCK_SIZE

static struct ioloop *ioloop;

enum rawlog_flags {
	RAWLOG_FLAG_LOG_INPUT		= 0x01,
	RAWLOG_FLAG_LOG_OUTPUT		= 0x02,
	RAWLOG_FLAG_LOG_TIMESTAMPS	= 0x04,
	RAWLOG_FLAG_LOG_BOUNDARIES	= 0x10,
	RAWLOG_FLAG_LOG_IP_IN_FILENAME	= 0x20
};

struct rawlog_proxy {
	int client_in_fd, client_out_fd, server_fd;
	struct io *client_io, *server_io;
	struct ostream *client_output, *server_output;

	struct ostream *in_output, *out_output;
	enum rawlog_flags flags;
	bool prev_lf_in, prev_lf_out;
};

static void rawlog_proxy_destroy(struct rawlog_proxy *proxy)
{
	if (proxy->in_output != NULL) {
		o_stream_uncork(proxy->in_output);
		if (o_stream_nfinish(proxy->in_output) < 0) {
			i_error("write(in) failed: %s",
				o_stream_get_error(proxy->in_output));
		}
		o_stream_destroy(&proxy->in_output);
	}
	if (proxy->out_output != NULL) {
		o_stream_uncork(proxy->out_output);
		if (o_stream_nfinish(proxy->out_output) < 0) {
			i_error("write(out) failed: %s",
				o_stream_get_error(proxy->out_output));
		}
		o_stream_destroy(&proxy->out_output);
	}
	if (proxy->client_io != NULL)
		io_remove(&proxy->client_io);
	if (proxy->server_io != NULL)
		io_remove(&proxy->server_io);

	o_stream_destroy(&proxy->client_output);
	o_stream_destroy(&proxy->server_output);

	if (close(proxy->client_in_fd) < 0)
		i_error("close(client_in_fd) failed: %m");
	if (close(proxy->client_out_fd) < 0)
		i_error("close(client_out_fd) failed: %m");
	if (close(proxy->server_fd) < 0)
		i_error("close(server_fd) failed: %m");
	i_free(proxy);

	io_loop_stop(ioloop);
}

static void
write_with_timestamps(struct ostream *output, bool *prev_lf,
		      const unsigned char *data, size_t size)
{
	T_BEGIN {
		const char *timestamp = t_strdup_printf("%ld.%06lu ",
			(long)ioloop_timeval.tv_sec,
			(unsigned long)ioloop_timeval.tv_usec);
		string_t *str = t_str_new(size + 128);
		size_t i;

		if (*prev_lf)
			str_append(str, timestamp);

		for (i = 0; i < size; i++) {
			str_append_c(str, data[i]);
			if (data[i] == '\n' && i+1 != size)
				str_append(str, timestamp);
		}
		*prev_lf = data[i-1] == '\n';
		o_stream_nsend(output, str_data(str), str_len(str));
	} T_END;
}

static void proxy_write_data(struct rawlog_proxy *proxy, struct ostream *output,
			     bool *prev_lf, const void *data, size_t size)
{
	if (output == NULL || output->closed || size == 0)
		return;

	if ((proxy->flags & RAWLOG_FLAG_LOG_BOUNDARIES) != 0)
		o_stream_nsend_str(output, "<<<\n");

	if ((proxy->flags & RAWLOG_FLAG_LOG_TIMESTAMPS) != 0)
		write_with_timestamps(output, prev_lf, data, size);
	else
		o_stream_nsend(output, data, size);

	if ((proxy->flags & RAWLOG_FLAG_LOG_BOUNDARIES) != 0)
		o_stream_nsend_str(output, ">>>\n");
}

static void proxy_write_in(struct rawlog_proxy *proxy,
			   const void *data, size_t size)
{
	proxy_write_data(proxy, proxy->in_output, &proxy->prev_lf_in,
			 data, size);
}

static void proxy_write_out(struct rawlog_proxy *proxy,
			    const void *data, size_t size)
{
	proxy_write_data(proxy, proxy->out_output, &proxy->prev_lf_out,<commit_msg>rawlog: Flush output to disk every second instead of only when buffer is full.<commit_after>/* Copyright (c) 2002-2015 Dovecot authors, see the included COPYING file */

#include "lib.h"

#include "ioloop.h"
#include "fd-set-nonblock.h"
#include "net.h"
#include "str.h"
#include "write-full.h"
#include "istream.h"
#include "ostream.h"
#include "process-title.h"
#include "restrict-access.h"
#include "time-util.h"
#include "master-service.h"

#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>

#define OUTBUF_THRESHOLD IO_BLOCK_SIZE
#define RAWLOG_TIMEOUT_FLUSH_MSECS 1000

static struct ioloop *ioloop;

enum rawlog_flags {
	RAWLOG_FLAG_LOG_INPUT		= 0x01,
	RAWLOG_FLAG_LOG_OUTPUT		= 0x02,
	RAWLOG_FLAG_LOG_TIMESTAMPS	= 0x04,
	RAWLOG_FLAG_LOG_BOUNDARIES	= 0x10,
	RAWLOG_FLAG_LOG_IP_IN_FILENAME	= 0x20
};

struct rawlog_proxy {
	int client_in_fd, client_out_fd, server_fd;
	struct io *client_io, *server_io;
	struct ostream *client_output, *server_output;
	struct timeout *to_flush;

	struct ostream *in_output, *out_output;
	enum rawlog_flags flags;
	bool prev_lf_in, prev_lf_out;
};

static void rawlog_proxy_destroy(struct rawlog_proxy *proxy)
{
	if (proxy->in_output != NULL) {
		o_stream_uncork(proxy->in_output);
		if (o_stream_nfinish(proxy->in_output) < 0) {
			i_error("write(in) failed: %s",
				o_stream_get_error(proxy->in_output));
		}
		o_stream_destroy(&proxy->in_output);
	}
	if (proxy->out_output != NULL) {
		o_stream_uncork(proxy->out_output);
		if (o_stream_nfinish(proxy->out_output) < 0) {
			i_error("write(out) failed: %s",
				o_stream_get_error(proxy->out_output));
		}
		o_stream_destroy(&proxy->out_output);
	}
	if (proxy->client_io != NULL)
		io_remove(&proxy->client_io);
	if (proxy->server_io != NULL)
		io_remove(&proxy->server_io);
	if (proxy->to_flush != NULL)
		timeout_remove(&proxy->to_flush);

	o_stream_destroy(&proxy->client_output);
	o_stream_destroy(&proxy->server_output);

	if (close(proxy->client_in_fd) < 0)
		i_error("close(client_in_fd) failed: %m");
	if (close(proxy->client_out_fd) < 0)
		i_error("close(client_out_fd) failed: %m");
	if (close(proxy->server_fd) < 0)
		i_error("close(server_fd) failed: %m");
	i_free(proxy);

	io_loop_stop(ioloop);
}

static void
write_with_timestamps(struct ostream *output, bool *prev_lf,
		      const unsigned char *data, size_t size)
{
	T_BEGIN {
		const char *timestamp = t_strdup_printf("%ld.%06lu ",
			(long)ioloop_timeval.tv_sec,
			(unsigned long)ioloop_timeval.tv_usec);
		string_t *str = t_str_new(size + 128);
		size_t i;

		if (*prev_lf)
			str_append(str, timestamp);

		for (i = 0; i < size; i++) {
			str_append_c(str, data[i]);
			if (data[i] == '\n' && i+1 != size)
				str_append(str, timestamp);
		}
		*prev_lf = data[i-1] == '\n';
		o_stream_nsend(output, str_data(str), str_len(str));
	} T_END;
}

static void proxy_flush_timeout(struct rawlog_proxy *proxy)
{
	bool flushed = TRUE;

	if (o_stream_flush(proxy->in_output) == 0)
		flushed = FALSE;
	if (o_stream_flush(proxy->out_output) == 0)
		flushed = FALSE;
	if (flushed)
		timeout_remove(&proxy->to_flush);
}

static void proxy_write_data(struct rawlog_proxy *proxy, struct ostream *output,
			     bool *prev_lf, const void *data, size_t size)
{
	if (output == NULL || output->closed || size == 0)
		return;

	if ((proxy->flags & RAWLOG_FLAG_LOG_BOUNDARIES) != 0)
		o_stream_nsend_str(output, "<<<\n");

	if ((proxy->flags & RAWLOG_FLAG_LOG_TIMESTAMPS) != 0)
		write_with_timestamps(output, prev_lf, data, size);
	else
		o_stream_nsend(output, data, size);

	if ((proxy->flags & RAWLOG_FLAG_LOG_BOUNDARIES) != 0)
		o_stream_nsend_str(output, ">>>\n");

	if (proxy->to_flush == NULL) {
		proxy->to_flush = timeout_add(RAWLOG_TIMEOUT_FLUSH_MSECS,
					      proxy_flush_timeout, proxy);
	}
}

static void proxy_write_in(struct rawlog_proxy *proxy,
			   const void *data, size_t size)
{
	proxy_write_data(proxy, proxy->in_output, &proxy->prev_lf_in,
			 data, size);
}

static void proxy_write_out(struct rawlog_proxy *proxy,
			    const void *data, size_t size)
{
	proxy_write_data(proxy, proxy->out_output, &proxy->prev_lf_out,
{'fraction_non_alphanumeric': 0.1442870141687248, 'fraction_numerical': 0.01117899389054985, 'mean_word_length': 5.165064102564102, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 80, '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': '2438279', 'n_tokens_mistral': 3338, 'n_tokens_neox': 3058, 'n_words': 598}