use Sys::Hostname qw(hostname);
use POSIX qw(strftime);
use PublicInbox::Search;
-use PublicInbox::SearchIdx qw(crlf_adjust prepare_stack is_ancestor
- is_bad_blob);
+use PublicInbox::SearchIdx qw(prepare_stack is_ancestor is_bad_blob);
use PublicInbox::OverIdx;
use PublicInbox::MiscIdx;
use PublicInbox::MID qw(mids);
use PublicInbox::InboxWritable;
use PublicInbox::ContentHash qw(content_hash);
use PublicInbox::Eml;
-use PublicInbox::DS qw(now);
+use PublicInbox::DS qw(now add_timer);
use DBI qw(:sql_types); # SQL_BLOB
sub new {
my ($req) = @_;
my $self = $req->{self};
my $new_smsg = $req->{new_smsg};
-
- # {raw_bytes} may be unset, so just use {bytes}
my $n = $self->{transact_bytes} += $new_smsg->{bytes};
# set flag for PublicInbox::V2Writable::index_todo:
my $oid = $new_smsg->{blob};
my $ibx = delete $req->{ibx} or die 'BUG: {ibx} unset';
$self->{oidx}->add_xref3($docid, $req->{xnum}, $oid, $ibx->eidx_key);
- $idx->index_raw(undef, $eml, $new_smsg, $ibx->eidx_key);
+ $idx->index_eml($eml, $new_smsg, $ibx->eidx_key);
check_batch_limit($req);
}
my $new_smsg = $req->{new_smsg} = bless {
blob => $oid,
}, 'PublicInbox::Smsg';
- $new_smsg->{bytes} = $size + crlf_adjust($$bref);
+ $new_smsg->set_bytes($$bref, $size);
defined($req->{xnum} = cur_ibx_xnum($req, $bref)) or return;
++${$req->{nr}};
do_step($req);
my $top_smsg = pop @$stable;
$top_smsg == $smsg or die 'BUG: top_smsg != smsg';
my $ibx = _ibx_for($self, $sync, $smsg);
- $idx->index_raw(undef, $eml, $smsg, $ibx->eidx_key);
+ $idx->index_eml($eml, $smsg, $ibx->eidx_key);
for my $x (reverse @$stable) {
$ibx = _ibx_for($self, $sync, $x);
my $hdr = delete $x->{hdr} // die 'BUG: no {hdr}';
my $ci = $self->{current_info};
local $self->{current_info} = "$ci #$docid $oid";
my $re_smsg = bless { blob => $oid }, 'PublicInbox::Smsg';
- $re_smsg->{bytes} = $size + crlf_adjust($$bref);
+ $re_smsg->set_bytes($$bref, $size);
my $eml = PublicInbox::Eml->new($bref);
$re_smsg->populate($eml, { autime => $orig_smsg->{ds},
cotime => $orig_smsg->{ts} });
my $self = $req->{self} // die 'BUG: {self} unset';
local $self->{current_info} = "$self->{current_info} $oid";
my $new_smsg = bless { blob => $oid, }, 'PublicInbox::Smsg';
- $new_smsg->{bytes} = $size + crlf_adjust($$bref);
+ $new_smsg->set_bytes($$bref, $size);
my $eml = $req->{eml} = PublicInbox::Eml->new($bref);
$req->{new_smsg} = $new_smsg;
$req->{chash} = content_hash($eml);
$self->with_umask(\&_idx_init, $self, $opt);
$self->{oidx}->begin_lazy;
$self->{oidx}->eidx_prep;
- $self->{midx}->begin_txn;
+ $self->{midx}->create_xdb if @new;
}
sub _watch_commit { # PublicInbox::DS::add_timer callback
delete $self->{-commit_timer};
eidxq_process($self, $self->{-watch_sync});
eidxq_release($self);
- delete local $self->{-watch_sync}->{-regen_fmt};
+ my $fmt = delete $self->{-watch_sync}->{-regen_fmt};
reindex_checkpoint($self, $self->{-watch_sync});
+ $self->{-watch_sync}->{-regen_fmt} = $fmt;
# call event_step => done unless commit_timer is armed
PublicInbox::DS::requeue($self);
$pr->("indexing $ekey\n") if $pr;
$self->idx_init($opt);
sync_inbox($self, $self->{-watch_sync}, $ibx);
- $self->{-commit_timer} //= PublicInbox::DS::add_timer(
- $opt->{'commit-interval'} // 10,
+ $self->{-commit_timer} //= add_timer($opt->{'commit-interval'} // 10,
\&_watch_commit, $self);
}
$pr->("performing initial scan ...\n") if $pr;
my $sync = eidx_sync($self, $opt); # initial sync
return if $sync->{quit};
- my $oldset = PublicInbox::Sigfd::block_signals();
+ my $oldset = PublicInbox::DS::block_signals();
local $self->{current_info} = '';
my $cb = $SIG{__WARN__} || \&CORE::warn;
local $SIG{__WARN__} = sub { $cb->($self->{current_info}, ': ', @_) };
if (!$sigfd) {
# wake up every second to accept signals if we don't
# have signalfd or IO::KQueue:
- PublicInbox::Sigfd::sig_setmask($oldset);
+ PublicInbox::DS::sig_setmask($oldset);
PublicInbox::DS->SetLoopTimeout(1000);
}
PublicInbox::DS->SetPostLoopCallback(sub { !$sync->{quit} });
*atfork_child = \&PublicInbox::V2Writable::atfork_child;
*idx_shard = \&PublicInbox::V2Writable::idx_shard;
*reindex_checkpoint = \&PublicInbox::V2Writable::reindex_checkpoint;
+*checkpoint = \&PublicInbox::V2Writable::checkpoint;
1;