view release on metacpan or  search on metacpan

lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm  view on Meta::CPAN

sub execute {
	my ( $self, $opt, $args ) = @_;

	# Lazily required for the same reason streamd does it: App::Cmd loads
	# every command module up front, and the rest of the CLI should work
	# on a box without JSON::MaybeXS.
	eval { require JSON::MaybeXS; 1 }
		or die( 'iforest streamc requires JSON::MaybeXS for its wire protocol; install it: ' . $@ );
	$JSON    = JSON::MaybeXS->new( utf8 => 1, canonical => 1 );
	$TIMEOUT = $opt->{'timeout'};

	# Resolve the socket exactly as streamd does (keep in sync with it):
	# without a set the flag is the socket file; with one it is the base
	# run dir holding <set>.sock.
	my $socket;
	if ( defined $opt->{'set'} ) {
		my $run = defined $opt->{'socket'} ? $opt->{'socket'} : '/var/run/iforest_streamd';
		$socket = File::Spec->catfile( $run, $opt->{'set'} . '.sock' );
	} else {
		$socket = defined $opt->{'socket'} ? $opt->{'socket'} : '/var/run/iforest_streamd/streamd.sock';
	}
	die(      '--socket, "'
			. $socket
			. '", is '
			. length($socket)
			. ' bytes; Unix socket paths are limited to ~104 bytes -- use a shorter path'
			. "\n" )
		if length($socket) > 100;

	$SOCK = IO::Socket::UNIX->new( Peer => $socket )
		or die( 'failed to connect to "'
			. $socket . '": '
			. $!
			. ' -- is streamd running'
			. ( defined $opt->{'set'} ? ' with --set ' . $opt->{'set'} : '' ) . '?'
			. "\n" );
	$SOCK->autoflush(1);

	# A daemon dropping us mid-write should surface as the read-side
	# "no reply" error, not a silent SIGPIPE death.
	local $SIG{PIPE} = 'IGNORE';

	return _command( $self, $opt ) if !defined $opt->{'i'};
	return _stream( $self, $opt );
} ## end sub execute

#-------------------------------------------------------------------------------
# wire helpers
#-------------------------------------------------------------------------------

# Send one request and wait for its reply.  The protocol is strictly one
# reply per request on a single connection, so this can block on the
# answer instead of correlating tags.
#
# Args:
#   $msg :: the request as a hashref, ready to encode -- a row/rows
#           message or a cmd message.
#
# Returns: the hashref _read_reply returns, holding the raw reply line and
# its decoded form.  Dies when the daemon says nothing within --timeout
# seconds or closes the connection first.
#
# Example:
#   my $got = _request( { cmd => 'stats' } );
#   $got->{reply}{ok}{seen};
sub _request {
	my ($msg) = @_;
	print {$SOCK} $JSON->encode($msg) . "\n";
	my $reply = _read_reply();
	die( 'no reply from the daemon within ' . $TIMEOUT . 's (or it closed the connection)' . "\n" )
		unless defined $reply;
	return $reply;
}

# Read one newline-terminated reply, waiting no longer than --timeout in
# total.  Whatever arrives past the newline stays in the module's read
# buffer for the next call, so a reply split across reads -- or two
# arriving in one -- both work.
#
# Args: none.  Reads the module's $SOCK, $READ_BUF and $TIMEOUT.
#
# Returns: a hashref of raw (the reply line as it came, for --json
# passthrough) and reply (the decoded hashref), or undef when the deadline
# passes or the daemon closes the connection.  Dies when a line arrives
# but does not parse as JSON.
#
# Example:
#   my $got = _read_reply();
#   print $got->{raw} . "\n" if $opt->{'json'};
sub _read_reply {
	my $deadline = time + $TIMEOUT;
	my $sel      = IO::Select->new($SOCK);
	while ( $READ_BUF !~ /\n/ ) {
		my $left = $deadline - time;
		return undef if $left <= 0 || !$sel->can_read($left);
		my $got = sysread( $SOCK, my $chunk, 65536 );
		return undef unless $got;
		$READ_BUF .= $chunk;
	}
	$READ_BUF =~ s/\A([^\n]*)\n//;
	my $line  = $1;
	my $reply = eval { $JSON->decode($line) };
	die( 'daemon sent an unparseable reply: ' . $@ ) if $@;
	return { raw => $line, reply => $reply };
} ## end sub _read_reply

#-------------------------------------------------------------------------------
# command mode
#-------------------------------------------------------------------------------

# The one-shot control path: send the single command the user asked for,
# print its answer, and exit.  execute dispatches here instead of _stream
# when any of --ping/--stats/--save/--relearn-threshold is present.
#
# Args:
#   $opt :: the parsed command options hashref.  Exactly one of ping,
#           stats, save or relearn_threshold is set; validate has already
#           enforced that.
#
# Returns: 1.  Prints the raw reply line under --json, a two-column table
# for a hash answer, and the bare value otherwise.  Dies naming the