lib/System/Introspector/Report/Publish/MediaWiki.pm

   1 package System::Introspector::Report::Publish::MediaWiki;
   2 use Moo;
   3 use Try::Tiny;
   4 use Object::Remote;
   5 use Object::Remote::Future;
   6 use System::Introspector::Report::Logging qw( :log );
   7
   8 use aliased 'System::Introspector::Report::Publish::MediaWiki::Updater';
   9
  10 has page_options => (
  11   is => 'ro',
  12   default => sub { {} },
  13   init_arg => 'page',
  14 );
  15
  16 has process_count => (
  17   is => 'ro',
  18   isa => sub {
  19     die "MediaWiki processes count has to be above 0\n"
  20       if $_[0] < 1;
  21   },
  22   default => sub { 1 },
  23   init_arg => 'processes',
  24 );
  25
  26 has updater_pool => (
  27   is => 'lazy',
  28   init_arg => undef,
  29 );
  30
  31 sub _build_updater_pool {
  32   my ($self) = @_;
  33   return [map {
  34     Updater->new::on('-', $self->connect_info_pairs);
  35   } 1 .. $self->process_count];
  36 }
  37
  38 sub publish {
  39   my ($self, $reports) = @_;
  40   my $pages = $self->page_options;
  41   log_info {
  42     sprintf "Pushing reports to MediaWiki at '%s'", $self->api_uri;
  43   };
  44   for my $page (sort keys %$pages) {
  45     log_verbose { "Publishing $page" };
  46     my $options = $pages->{$page};
  47     if ($page =~ m{\%\(meta:(.+?)\)}) {
  48       $self->_publish_dynamic($reports, $page, $1, $options);
  49     }
  50     else {
  51       log_trace { "Standard page: $page" };
  52       $self->_publish_pages([$reports, $page, $options]);
  53     }
  54   }
  55   log_info { "Finished pushing to MediaWiki" };
  56   return 1;
  57 }
  58
  59 sub _publish_pages {
  60   my ($self, @pages) = @_;
  61   log_trace { sprintf "Publishing %i pages", scalar(@pages) };
  62   my $stream = sub {
  63     my $next = shift @pages
  64       or return undef;
  65     my ($reports, $page_name, $options) = @$next;
  66     my $sorted = $self->_sort_reports($reports, $options->{report} || []);
  67     return [$page_name, $options, $sorted];
  68   };
  69   await_all map {
  70     log_trace {
  71       sprintf "Creating process to handle %i items from the update pool",
  72         scalar(@{ $self->updater_pool }) };
  73     $_->start::process($stream);
  74   } @{ $self->updater_pool };
  75 }
  76
  77 sub _sort_reports {
  78   my ($self, $reports, $included) = @_;
  79   log_trace { "Sorting reports" };
  80   my @matchers = map {
  81     $self->_prepare_matcher_from($_);
  82   } ref($included) ? @{$included} : $included;
  83   my @grouped;
  84   for my $report (@$reports) {
  85     my ($group_idx) = grep {
  86       $self->_match_id($report, $matchers[$_]);
  87     } 0 .. $#matchers;
  88     if (defined $group_idx) {
  89       push @{$grouped[$group_idx]}, $report;
  90     }
  91   }
  92   return [ map { (@$_) } @grouped ];
  93 }
  94
  95 sub _publish_dynamic {
  96   my ($self, $reports, $page_name, $key, $options) = @_;
  97   log_debug { "Creating pages for dynamic page '$page_name'" };
  98   my %reports_by_key;
  99   for my $report (@$reports) {
 100     my $value = $report->{meta}{$key};
 101     next unless defined $value;
 102     push @{ $reports_by_key{$value} ||= [] }, $report;
 103   }
 104   my @pages;
 105   for my $value (keys %reports_by_key) {
 106     (my $effective_page = $page_name)
 107       =~ s{\%\(meta:\Q$key\E\)}{$value};
 108     my $assoc = $reports_by_key{$value};
 109     push @pages, [$assoc, $effective_page, $options];
 110   }
 111   log_trace {
 112       sprintf "%i pages were created for '%s'; publishing them",
 113       scalar(@pages), $page_name
 114   };
 115   $self->_publish_pages(@pages);
 116   return 1;
 117 }
 118
 119 with $_ for qw(
 120   System::Introspector::Report::Publish::API
 121   System::Introspector::Report::Publish::MediaWiki::HasConnectInfo
 122 );
 123
 124 1;