lib/PublicInbox/LeiRemote.pm

   1 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # Make remote externals HTTP(S) inboxes behave like
   5 # PublicInbox::Inbox and PublicInbox::Search/ExtSearch.
   6 # This exists solely for SolverGit.  It is a high-latency a
   7 # synchronous API that is not at all fast.
   8 package PublicInbox::LeiRemote;
   9 use v5.10.1;
  10 use strict;
  11 use IO::Uncompress::Gunzip;
  12 use PublicInbox::OnDestroy;
  13 use PublicInbox::MboxReader;
  14 use PublicInbox::Spawn qw(popen_rd);
  15 use PublicInbox::LeiCurl;
  16 use PublicInbox::ContentHash qw(git_sha);
  17
  18 sub new {
  19         my ($cls, $lei, $uri) = @_;
  20         bless { uri => $uri, lei => $lei }, $cls;
  21 }
  22
  23 sub isrch { $_[0] } # SolverGit expcets this
  24
  25 sub _each_mboxrd_eml { # callback for MboxReader->mboxrd
  26         my ($eml, $self) = @_;
  27         my $lei = $self->{lei};
  28         my $xoids = $lei->{ale}->xoids_for($eml, 1);
  29         my $smsg = bless {}, 'PublicInbox::Smsg';
  30         if ($lei->{sto} && !$xoids) { # memoize locally
  31                 my $res = $lei->{sto}->wq_do('add_eml', $eml);
  32                 $smsg = $res if ref($res) eq ref($smsg);
  33         }
  34         $smsg->{blob} //= $xoids ? (keys(%$xoids))[0]
  35                                 : $lei->git_oid($eml)->hexdigest;
  36         $smsg->populate($eml);
  37         $smsg->{mid} //= '(none)';
  38         push @{$self->{smsg}}, $smsg;
  39 }
  40
  41 sub mset {
  42         my ($self, $qstr, undef) = @_; # $opt ($_[2]) ignored
  43         my $lei = $self->{lei};
  44         my $curl = PublicInbox::LeiCurl->new($lei, $lei->{curl});
  45         push @$curl, '-s', '-d', '';
  46         my $uri = $self->{uri}->clone;
  47         $uri->query_form(q => $qstr, x => 'm', r => 1); # r=1: relevance
  48         my $cmd = $curl->for_uri($self->{lei}, $uri);
  49         $self->{lei}->qerr("# $cmd");
  50         my $rdr = { 2 => $lei->{2}, pgid => 0 };
  51         my ($fh, $pid) = popen_rd($cmd, undef, $rdr);
  52         my $reap = PublicInbox::OnDestroy->new($lei->can('sigint_reap'), $pid);
  53         $self->{smsg} = [];
  54         $fh = IO::Uncompress::Gunzip->new($fh, MultiStream => 1);
  55         PublicInbox::MboxReader->mboxrd($fh, \&_each_mboxrd_eml, $self);
  56         my $err = waitpid($pid, 0) == $pid ? undef
  57                                         : "BUG: waitpid($cmd): $!";
  58         @$reap = (); # cancel OnDestroy
  59         my $wait = $self->{lei}->{sto}->wq_do('done');
  60         die $err if $err;
  61         $self; # we are the mset (and $ibx, and $self)
  62 }
  63
  64 sub size { scalar @{$_[0]->{smsg}} } # size of previous results
  65
  66 sub mset_to_smsg {
  67         my ($self, $ibx, $mset) = @_; # all 3 are $self
  68         wantarray ? ($self->size, @{$self->{smsg}}) : $self->{smsg};
  69 }
  70
  71 sub base_url { "$_[0]->{uri}" }
  72
  73 sub smsg_eml {
  74         my ($self, $smsg) = @_;
  75         if (my $bref = $self->{lei}->ale->git->cat_file($smsg->{blob})) {
  76                 return PublicInbox::Eml->new($bref);
  77         }
  78         $self->{lei}->err("E: $self->{uri} $smsg->{blob} gone <$smsg->{mid}>");
  79         undef;
  80 }
  81
  82 1;