inter.pl

#
# carton exec hypnotoad -f inter.pl
#
#
use 5.20.0;
use Mojolicious::Lite;
use Mojo::JSON qw(decode_json encode_json);
use Mojo::Redis2;

plugin 'DefaultHelpers';

app->config( hypnotoad => {
	workers => 5,
	# same as morbo
	listen => ["http://*:3000"],
});

helper mystash => sub {

	state $stash = {};
	return $stash;
};
helper log => sub {

	state $LOG = Mojo::Log->new;
	return $LOG;
};

# errors are at odd indices
helper ioErr => sub {
	my ($c, @arr) = @_;

	# rethrow exception
	die($arr[1]) if ref($arr[1]);

	my $i = 0;
	my $err = join "; ",
		grep { $i++ %2 and $_ } @arr;

	die({err => $err}) if $err;
};

helper redis => sub {

	my ($c) = @_;
	state $redis;

	return $redis if $redis;

	$redis = Mojo::Redis2->new;
	my $msgHandler = sub {
		my ($msg) = @_;

		# handle requests
		if (my $request = $msg->{request}) {
			$msg->{response} = delete $msg->{request};

			if ($request eq "getPID") {
				$msg->{result} = $$;
			}
			return $msg if exists $msg->{result};
		}
		# handle responses
		elsif (my $response = $msg->{response}) {
			if ($response eq "getPID") {
				my $arr = $c->mystash->{ $response } ||= [];
				push @$arr, $msg;
			}
			return;
		}

		# nothing to do..
		$c->log->warn("[$$] Don't understand ". $c->dumper($msg));
		return;
	};

	my $redisEvents = {
		# redis connection event
		connection => sub {
			my ($redis, $conn) = @_;
			$c->log->info("[$$] Connected to redis! [$conn->{id}]"); # . $c->dumper($redis)
		},
		# handle redis errors
		error => sub {
			my ($redis, $err) = @_;
			$c->log->error($err);
		},
		# incoming redis messages
		message => sub {
			my ($redis, $msg, $node) = @_;
			$c->log->debug("Incoming redis message for node $node: $msg");

			my $resp = $msgHandler->(decode_json($msg)) or return;
			$redis->publish($node, encode_json($resp), sub {});
		},
	};
	#
	$redis->on($_ => $redisEvents->{$_}) for keys %$redisEvents;

	$redis->subscribe(["WORKERS"], sub {
		my ($self, $err, $res) = @_;

		$c->log->error("error on redis subscribe: $err") if $err;
		$c->log->debug("[$$] subscribed!") if !$err;
	});
	return $redis;
};

get '/__status' => sub {
	my ($c) = @_;

	my $req = {request => "getPID"};
	delete $c->mystash->{ $req->{request} };

	Mojo::IOLoop->delay(
		sub {
			my ($d, $err) = @_;

			$c->redis->publish("WORKERS", encode_json($req), $d->begin);
			Mojo::IOLoop->timer(1 => $d->begin);
		},
		sub {
			my ($d) = @_;
			$c->ioErr(@_);

			$c->render(json => {
				me => $$,
				report => delete $c->mystash->{ $req->{request} }
			});
			# $c->$cb(undef, my $ok=1);
		},
	)
	# ->catch(sub{ $c->$cb($@) })
	;
};

# force connect at startup
Mojo::IOLoop->timer(0 => sub{
	app->redis->backend->time;
});

app->start;