<commit_before>/*
 * Copyright (c) 2008-2013 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

/* Import. */

#include "private.h"

/* Forward. */

#ifdef HAVE_LIBXS
static void free_wrapper(void *, void *);
#endif

/* Internal functions. */

nmsg_res
_output_nmsg_flush(nmsg_output_t output) {
	nmsg_res res = nmsg_res_success;

	if (_nmsg_container_get_num_payloads(output->stream->c) > 0)
		res = _output_nmsg_write_container(output);

	return (res);
}

nmsg_res
_output_nmsg_write(nmsg_output_t output, nmsg_message_t msg) {
	Nmsg__NmsgPayload *np;
	nmsg_res res;

	assert(msg->np != NULL);
	np = msg->np;

	/* set source, output, group if necessary */
	if (output->stream->source != 0) {
		np->source = output->stream->source;
		np->has_source = 1;
	}
	if (output->stream->operator != 0) {
		np->operator_ = output->stream->operator;
		np->has_operator_ = 1;
	}
	if (output->stream->group != 0) {
		np->group = output->stream->group;
		np->has_group = 1;
	}

	pthread_mutex_lock(&output->stream->lock);

	res = _nmsg_container_add(output->stream->c, msg);

	if (res == nmsg_res_container_full) {
		res = _output_nmsg_write_container(output);
		if (res != nmsg_res_success)
			goto out;
		res = _nmsg_container_add(output->stream->c, msg);
		if (res == nmsg_res_container_overfull)
			res = _output_frag_write(output);
	} else if (res == nmsg_res_success && output->stream->buffered == false) {
		res = _output_nmsg_write_container(output);
	} else if (res == nmsg_res_container_overfull) {
		res = _output_frag_write(output);
	}

out:
	pthread_mutex_unlock(&output->stream->lock);
	if (output->stream->rate != NULL)
		nmsg_rate_sleep(output->stream->rate);
	return (res);
}

nmsg_res
_output_nmsg_write_container(nmsg_output_t output) {
	nmsg_res res;
	size_t buf_len;
	uint8_t *buf;

	res = _nmsg_container_serialize(output->stream->c,
					&buf,
					&buf_len,
					true, /* do_header */<commit_msg>output_nmsg: _output_nmsg_write(): do rate-limiting based on actual socket writes, not mere buffering of payloads<commit_after>/*
 * Copyright (c) 2008-2013 by Internet Systems Consortium, Inc. ("ISC")
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
 * OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

/* Import. */

#include "private.h"

/* Forward. */

#ifdef HAVE_LIBXS
static void free_wrapper(void *, void *);
#endif

/* Internal functions. */

nmsg_res
_output_nmsg_flush(nmsg_output_t output) {
	nmsg_res res = nmsg_res_success;

	if (_nmsg_container_get_num_payloads(output->stream->c) > 0)
		res = _output_nmsg_write_container(output);

	return (res);
}

nmsg_res
_output_nmsg_write(nmsg_output_t output, nmsg_message_t msg) {
	Nmsg__NmsgPayload *np;
	nmsg_res res;
	bool did_write = false;

	assert(msg->np != NULL);
	np = msg->np;

	/* set source, output, group if necessary */
	if (output->stream->source != 0) {
		np->source = output->stream->source;
		np->has_source = 1;
	}
	if (output->stream->operator != 0) {
		np->operator_ = output->stream->operator;
		np->has_operator_ = 1;
	}
	if (output->stream->group != 0) {
		np->group = output->stream->group;
		np->has_group = 1;
	}

	pthread_mutex_lock(&output->stream->lock);

	res = _nmsg_container_add(output->stream->c, msg);

	if (res == nmsg_res_container_full) {
		res = _output_nmsg_write_container(output);
		if (res != nmsg_res_success)
			goto out;
		res = _nmsg_container_add(output->stream->c, msg);
		if (res == nmsg_res_container_overfull)
			res = _output_frag_write(output);
		did_write = true;
	} else if (res == nmsg_res_success && output->stream->buffered == false) {
		res = _output_nmsg_write_container(output);
		did_write = true;
	} else if (res == nmsg_res_container_overfull) {
		res = _output_frag_write(output);
		did_write = true;
	}

out:
	pthread_mutex_unlock(&output->stream->lock);
	if (did_write && output->stream->rate != NULL)
		nmsg_rate_sleep(output->stream->rate);

	return (res);
}

nmsg_res
_output_nmsg_write_container(nmsg_output_t output) {
	nmsg_res res;
	size_t buf_len;
	uint8_t *buf;

	res = _nmsg_container_serialize(output->stream->c,
					&buf,
					&buf_len,
					true, /* do_header */
{'fraction_non_alphanumeric': 0.11781978344650394, 'fraction_numerical': 0.005872637181134153, 'mean_word_length': 4.4282868525896415, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 81, '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': '5787592', 'n_tokens_mistral': 1998, 'n_tokens_neox': 1836, 'n_words': 511}