<commit_before>package PEF::Log::Appender::Socket;
use base 'PEF::Log::Appender';
use IO::Socket;
use IO::Socket::INET;
use IO::Socket::UNIX;
use Errno;
use Carp;
use strict;
use warnings;

sub new {
	my ($class, %params) = @_;
	my $self = $class->SUPER::new(%params);
	$self->{owner_pid} = 0;
	$self->reload(\%params);
}

sub reload {
	my ($self, $params) = @_;
	$self->_reload($params);
	my $out = $params->{out} or croak "no output socket";
	$self->_final if $$ != $self->{owner_pid};
	return $self if exists $self->{sock} and $self->{sock} and $out eq $self->{out};
	$self->{out} = $out;
	my ($scheme, $authority, $path, $query, $fragment) =
	  $out =~ m|(?:([^:/?#]+):)?(?://([^/?#]*))?([^?#]*)(?:\?([^#]*))?(?:#(.*))?|;
	my @socket_params;
	my $socket_class;
	if ($scheme eq 'unix') {
		# unix stream
		@socket_params = (
			Type => SOCK_STREAM,
			Peer => $path
		);
		$socket_class = "IO::Socket::UNIX";
	} elsif ($scheme eq 'local') {
		@socket_params = (
			Type => SOCK_DGRAM,
			Peer => $path
		);
		$socket_class = "IO::Socket::UNIX";
	} elsif ($scheme eq 'tcp') {
		@socket_params = (
			Timeout => $params->{timeout} || 3,
			Type => SOCK_STREAM,
			PeerAddr => $authority,
			Proto    => 'tcp',
		);
		$socket_class = "IO::Socket::INET";
	} elsif ($scheme eq 'udp') {
		@socket_params = (
			Timeout => $params->{timeout} || 3,
			Type => SOCK_DGRAM,
			PeerAddr => $authority,
			Proto    => 'udp',
		);
		$socket_class = "IO::Socket::INET";
	} else {
		croak "unknown socket scheme";
	}
	$self->{socket_class}  = $socket_class;
	$self->{socket_params} = \@socket_params;
	$self->_reconnect;
	$self;
}

sub _reconnect {
	my ($self, $params) = @_;
	my $socket_class = $self->{socket_class};
	$self->final;
	$self->{sock} = "$socket_class"->new(@{$self->{socket_params}})
	  or croak "can't connect to $self->{out}: $!";
	$self->{owner_pid} = $$;
}

sub append {
	my ($self, $level, $stream, $msg) = @_;
	my $line = $self->SUPER::append($level, $stream, $msg);
	utf8::encode($line) if utf8::is_utf8($line);
	my $attempts = 0;
	my $octets   = length ($line);
	while (1) {
		my $len = length ($line);
		my $rc  = $self->{sock}->syswrite($line);
		if (!defined $rc) {
			if ($!{EINTR} || $!{EAGAIN}) {
				next;
			}
			if (++$attempts > 5) {
				croak "Socket::append ($len): $!";
			}
			$self->_reconnect;
		} else {
			if ($octets != $rc) {
				substr ($line, 0, $rc) = '';
				$octets -= $rc;
			} else {
				last;
			}
		}
	}
}

sub final {
	if ($_[0]->{sock}) {
		close $_[0]->{sock};
		undef $_[0]->{sock};
	}
}

sub DESTROY {
	$_[0]->final;<commit_msg>fixed typos<commit_after>package PEF::Log::Appender::Socket;
use base 'PEF::Log::Appender';
use IO::Socket;
use IO::Socket::INET;
use IO::Socket::UNIX;
use Errno;
use Carp;
use strict;
use warnings;

sub new {
	my ($class, %params) = @_;
	my $self = $class->SUPER::new(%params);
	$self->{owner_pid} = 0;
	$self->reload(\%params);
}

sub reload {
	my ($self, $params) = @_;
	$self->_reload($params);
	my $out = $params->{out} or croak "no output socket";
	$self->_final if $$ != $self->{owner_pid};
	return $self if exists $self->{sock} and $self->{sock} and $out eq $self->{out};
	$self->{out} = $out;
	my ($scheme, $authority, $path, $query, $fragment) =
	  $out =~ m|(?:([^:/?#]+):)?(?://([^/?#]*))?([^?#]*)(?:\?([^#]*))?(?:#(.*))?|;
	my @socket_params;
	my $socket_class;
	if ($scheme eq 'unix') {
		# unix stream
		@socket_params = (
			Type => SOCK_STREAM,
			Peer => $path
		);
		$socket_class = "IO::Socket::UNIX";
	} elsif ($scheme eq 'local') {
		@socket_params = (
			Type => SOCK_DGRAM,
			Peer => $path
		);
		$socket_class = "IO::Socket::UNIX";
	} elsif ($scheme eq 'tcp') {
		@socket_params = (
			Timeout => $params->{timeout} || 3,
			Type => SOCK_STREAM,
			PeerAddr => $authority,
			Proto    => 'tcp',
		);
		$socket_class = "IO::Socket::INET";
	} elsif ($scheme eq 'udp') {
		@socket_params = (
			Timeout => $params->{timeout} || 3,
			Type => SOCK_DGRAM,
			PeerAddr => $authority,
			Proto    => 'udp',
		);
		$socket_class = "IO::Socket::INET";
	} else {
		croak "unknown socket scheme";
	}
	$self->{socket_class}  = $socket_class;
	$self->{socket_params} = \@socket_params;
	$self->_reconnect;
	$self;
}

sub _reconnect {
	my ($self, $params) = @_;
	my $socket_class = $self->{socket_class};
	$self->_final;
	$self->{sock} = "$socket_class"->new(@{$self->{socket_params}})
	  or croak "can't connect to $self->{out}: $!";
	$self->{owner_pid} = $$;
}

sub append {
	my ($self, $level, $stream, $msg) = @_;
	my $line = $self->SUPER::append($level, $stream, $msg);
	utf8::encode($line) if utf8::is_utf8($line);
	my $attempts = 0;
	my $octets   = length ($line);
	while (1) {
		my $len = length ($line);
		my $rc  = $self->{sock}->syswrite($line);
		if (!defined $rc) {
			if ($!{EINTR} || $!{EAGAIN}) {
				next;
			}
			if (++$attempts > 5) {
				croak "Socket::append ($len): $!";
			}
			$self->_reconnect;
		} else {
			if ($octets != $rc) {
				substr ($line, 0, $rc) = '';
				$octets -= $rc;
			} else {
				last;
			}
		}
	}
}

sub _final {
	if ($_[0]->{sock}) {
		close $_[0]->{sock};
		undef $_[0]->{sock};
	}
}

sub DESTROY {
	$_[0]->_final;
{'fraction_non_alphanumeric': 0.2467456770934525, 'fraction_numerical': 0.00544006217213911, 'mean_word_length': 3.68, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 91, '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': '4487560', 'n_tokens_mistral': 2323, 'n_tokens_neox': 2129, 'n_words': 450}