work: support distinct worker invocations per backend (f136a53b) · Commits · derf / travelynx

lib/Travelynx/Command/work.pm

+171 −160

Original line number	Diff line number	Diff line
		@@ -18,7 +18,7 @@ has description => 'Update real-time data of active journeys';
		has usage => sub { shift->extract_usage };

		sub run {
		my ($self) = @_;
		my ( $self, $backend ) = @_;

		my $now = DateTime->now( time_zone => 'Europe/Berlin' );
		my $checkin_deadline = $now->clone->subtract( hours => 48 );
		@@ -53,7 +53,9 @@ sub run {
		my $arr = $entry->{arr_eva};
		my $train_id = $entry->{train_id};

		if ( $train_id eq 'manual' ) {
		if ( $train_id eq 'manual'
		and ( not $backend or $backend eq 'manual' ) )
		{
		if ( $arr
		and $entry->{real_arr_ts}
		and $now->epoch - $entry->{real_arr_ts} > 600 )
		@@ -66,11 +68,10 @@ sub run {
		uid => $uid
		)->wait;
		}

		next;
		}

		if ( $entry->{is_dbris} ) {
		elsif ( $entry->{is_dbris} and ( not $backend or $backend eq 'dbris' ) )
		{

		eval {

		@@ -206,10 +207,9 @@ sub run {
		$self->app->log->error(
		"work($uid) @ DBRIS $entry->{backend_name}: $@");
		}
		next;
		}

		if ( $entry->{is_efa} ) {
		elsif ( $entry->{is_efa} and ( not $backend or $backend eq 'efa' ) ) {
		eval {
		$self->app->efa->get_journey_p(
		trip_id => $train_id,
		@@ -302,10 +302,10 @@ sub run {
		$self->app->log->error(
		"work($uid) @ EFA $entry->{backend_name}: $@");
		}
		next;
		}

		if ( $entry->{is_motis} ) {
		elsif ( $entry->{is_motis} and ( not $backend or $backend eq 'motis' ) )
		{

		eval {
		$self->app->motis->get_trip_p(
		@@ -326,6 +326,10 @@ sub run {
		stop => $stopover->stop,
		motis => $entry->{backend_name},
		);

		$self->app->log->debug( "mapped "
		. $stopover->stop->id . " to "
		. $stopover->stop->{eva} );
		}
		}

		@@ -399,10 +403,10 @@ sub run {
		$self->app->log->error(
		"work($uid) @ MOTIS $entry->{backend_name}: $@");
		}
		next;
		}

		if ( $entry->{is_hafas} ) {
		elsif ( $entry->{is_hafas} and ( not $backend or $backend eq 'hafas' ) )
		{

		eval {

		@@ -533,7 +537,6 @@ sub run {
		$self->app->log->error(
		"work($uid) @ HAFAS $entry->{backend_name}: $@");
		}
		next;
		}

		# TODO irgendwo ist hier ne race condition wo ein neuer checkin (in HAFAS) mit IRIS-Daten überschrieben wird.
		@@ -545,6 +548,7 @@ sub run {
		# update departure data for up to 15 minutes after departure and
		# delaying automatic checkout by at least 10 minutes.

		elsif ( $entry->{is_iris} and ( not $backend or $backend eq 'iris' ) ) {
		eval {
		if ( $now->epoch - $entry->{real_dep_ts} < 900 ) {
		my $status = $self->app->iris->get_departures(
		@@ -556,7 +560,8 @@ sub run {
		die("get_departures($dep): $status->{errstr}\n");
		}

		my ($train) = List::Util::first { $_->train_id eq $train_id }
		my ($train)
		= List::Util::first { $_->train_id eq $train_id }
		@{ $status->{results} };

		if ( not $train ) {
		@@ -723,20 +728,26 @@ sub run {
		eval { };
		}

		}

		my $started_at = $now;
		my $main_finished_at = DateTime->now( time_zone => 'Europe/Berlin' );
		my $worker_duration = $main_finished_at->epoch - $started_at->epoch;

		if ( $self->app->config->{influxdb}->{url} ) {
		my $tags = q{};
		if ($backend) {
		$tags .= ",backend=${backend}";
		}
		if ( $self->app->mode eq 'development' ) {
		$self->app->log->debug( 'POST '
		. $self->app->config->{influxdb}->{url}
		. " worker runtime_seconds=${worker_duration},errors=${errors},backend_errors=${backend_issues},ratelimit_count=${rate_limit_counts}"
		. " worker${tags} runtime_seconds=${worker_duration},errors=${errors},backend_errors=${backend_issues},ratelimit_count=${rate_limit_counts}"
		);
		}
		else {
		$self->app->ua->post_p( $self->app->config->{influxdb}->{url},
		"worker runtime_seconds=${worker_duration},errors=${errors},backend_errors=${backend_issues},ratelimit_count=${rate_limit_counts}"
		"worker${tags} runtime_seconds=${worker_duration},errors=${errors},backend_errors=${backend_issues},ratelimit_count=${rate_limit_counts}"
		)->wait;
		}
		}

Admin message