xapian-core/backends/remote/remote-database.cc

   1 /** @file remote-database.cc
   2  *  @brief Remote backend database class
   3  */
   4 /* Copyright (C) 2006,2007,2008,2009,2010,2011,2012,2013,2014,2015,2017 Olly Betts
   5  * Copyright (C) 2007,2009,2010 Lemur Consulting Ltd
   6  *
   7  * This program is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU General Public License as
   9  * published by the Free Software Foundation; either version 2 of the
  10  * License, or (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with this program; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  20  */
  21
  22 #include <config.h>
  23
  24 #include "remote-database.h"
  25
  26 #include <signal.h>
  27
  28 #include "api/msetinternal.h"
  29 #include "api/smallvector.h"
  30 #include "backends/inmemory/inmemory_positionlist.h"
  31 #include "net_postlist.h"
  32 #include "net_termlist.h"
  33 #include "remote-document.h"
  34 #include "omassert.h"
  35 #include "realtime.h"
  36 #include "net/length.h"
  37 #include "net/serialise.h"
  38 #include "net/serialise-error.h"
  39 #include "serialise-double.h"
  40 #include "str.h"
  41 #include "stringutils.h" // For STRINGIZE().
  42 #include "weight/weightinternal.h"
  43
  44 #include <cerrno>
  45 #include <memory>
  46 #include <string>
  47 #include <vector>
  48
  49 #include "xapian/constants.h"
  50 #include "xapian/error.h"
  51 #include "xapian/matchspy.h"
  52
  53 using namespace std;
  54 using Xapian::Internal::intrusive_ptr;
  55
  56 [[noreturn]]
  57 static void
  58 throw_handshake_failed(const string & context)
  59 {
  60     throw Xapian::NetworkError("Handshake failed - is this a Xapian server?",
  61                                context);
  62 }
  63
  64 [[noreturn]]
  65 static void
  66 throw_connection_closed_unexpectedly()
  67 {
  68     throw Xapian::NetworkError("Connection closed unexpectedly");
  69 }
  70
  71 RemoteDatabase::RemoteDatabase(int fd, double timeout_,
  72                                const string & context_, bool writable,
  73                                int flags)
  74     : Xapian::Database::Internal(writable ?
  75                                  TRANSACTION_NONE :
  76                                  TRANSACTION_READONLY),
  77       link(fd, fd, context_),
  78       context(context_),
  79       cached_stats_valid(),
  80       mru_valstats(),
  81       mru_slot(Xapian::BAD_VALUENO),
  82       timeout(timeout_)
  83 {
  84 #ifndef __WIN32__
  85     // It's simplest to just ignore SIGPIPE.  We'll still know if the
  86     // connection dies because we'll get EPIPE back from write().
  87     if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) {
  88         throw Xapian::NetworkError("Couldn't set SIGPIPE to SIG_IGN", errno);
  89     }
  90 #endif
  91
  92     update_stats(MSG_MAX);
  93
  94     if (writable) {
  95         if (flags & Xapian::DB_RETRY_LOCK) {
  96             const string & body = encode_length(flags & Xapian::DB_RETRY_LOCK);
  97             update_stats(MSG_WRITEACCESS, body);
  98         } else {
  99             update_stats(MSG_WRITEACCESS);
 100         }
 101     }
 102 }
 103
 104 Xapian::termcount
 105 RemoteDatabase::positionlist_count(Xapian::docid did,
 106                                    const std::string& term) const
 107 {
 108     if (cached_stats_valid && !has_positional_info)
 109         return 0;
 110
 111     send_message(MSG_POSITIONLISTCOUNT, encode_length(did) + term);
 112
 113     string message;
 114     get_message(message, REPLY_POSITIONLISTCOUNT);
 115     const char * p = message.data();
 116     const char * p_end = p + message.size();
 117     Xapian::termcount count;
 118     decode_length(&p, p_end, count);
 119     if (p != p_end) {
 120         throw Xapian::NetworkError("Bad REPLY_POSITIONLISTCOUNT message received", context);
 121     }
 122     return count;
 123 }
 124
 125 void
 126 RemoteDatabase::keep_alive()
 127 {
 128     send_message(MSG_KEEPALIVE, string());
 129     string message;
 130     get_message(message, REPLY_DONE);
 131 }
 132
 133 TermList *
 134 RemoteDatabase::open_metadata_keylist(const std::string &prefix) const
 135 {
 136     // Ensure that total_length and doccount are up-to-date.
 137     if (!cached_stats_valid) update_stats();
 138
 139     send_message(MSG_METADATAKEYLIST, prefix);
 140
 141     string message;
 142     unique_ptr<NetworkTermList> tlist(
 143         new NetworkTermList(0, doccount,
 144                             intrusive_ptr<const RemoteDatabase>(this),
 145                             0));
 146     vector<NetworkTermListItem> & items = tlist->items;
 147
 148     string term = prefix;
 149     while (get_message_or_done(message, REPLY_METADATAKEYLIST)) {
 150         NetworkTermListItem item;
 151         term.resize(size_t(static_cast<unsigned char>(message[0])));
 152         term.append(message, 1, string::npos);
 153         item.tname = term;
 154         items.push_back(item);
 155     }
 156
 157     tlist->current_position = tlist->items.begin();
 158     return tlist.release();
 159 }
 160
 161 TermList *
 162 RemoteDatabase::open_term_list(Xapian::docid did) const
 163 {
 164     Assert(did);
 165
 166     // Ensure that total_length and doccount are up-to-date.
 167     if (!cached_stats_valid) update_stats();
 168
 169     send_message(MSG_TERMLIST, encode_length(did));
 170
 171     string message;
 172     get_message(message, REPLY_DOCLENGTH);
 173     const char * p = message.c_str();
 174     const char * p_end = p + message.size();
 175     Xapian::termcount doclen;
 176     decode_length(&p, p_end, doclen);
 177     if (p != p_end) {
 178         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context);
 179     }
 180
 181     unique_ptr<NetworkTermList> tlist(
 182         new NetworkTermList(doclen, doccount,
 183                             intrusive_ptr<const RemoteDatabase>(this),
 184                             did));
 185     vector<NetworkTermListItem> & items = tlist->items;
 186
 187     string term;
 188     while (get_message_or_done(message, REPLY_TERMLIST)) {
 189         NetworkTermListItem item;
 190         p = message.data();
 191         p_end = p + message.size();
 192         decode_length(&p, p_end, item.wdf);
 193         decode_length(&p, p_end, item.termfreq);
 194         term.resize(size_t(static_cast<unsigned char>(*p++)));
 195         term.append(p, p_end);
 196         item.tname = term;
 197         items.push_back(item);
 198     }
 199
 200     tlist->current_position = tlist->items.begin();
 201     return tlist.release();
 202 }
 203
 204 TermList *
 205 RemoteDatabase::open_term_list_direct(Xapian::docid did) const
 206 {
 207     return RemoteDatabase::open_term_list(did);
 208 }
 209
 210 TermList *
 211 RemoteDatabase::open_allterms(const string & prefix) const {
 212     // Ensure that total_length and doccount are up-to-date.
 213     if (!cached_stats_valid) update_stats();
 214
 215     send_message(MSG_ALLTERMS, prefix);
 216
 217     unique_ptr<NetworkTermList> tlist(
 218         new NetworkTermList(0, doccount,
 219                             intrusive_ptr<const RemoteDatabase>(this),
 220                             0));
 221     vector<NetworkTermListItem> & items = tlist->items;
 222
 223     string term = prefix;
 224     string message;
 225     while (get_message_or_done(message, REPLY_ALLTERMS)) {
 226         NetworkTermListItem item;
 227         const char * p = message.data();
 228         const char * p_end = p + message.size();
 229         decode_length(&p, p_end, item.termfreq);
 230         term.resize(size_t(static_cast<unsigned char>(*p++)));
 231         term.append(p, p_end);
 232         item.tname = term;
 233         items.push_back(item);
 234     }
 235
 236     tlist->current_position = tlist->items.begin();
 237     return tlist.release();
 238 }
 239
 240 PostList *
 241 RemoteDatabase::open_post_list(const string& term) const
 242 {
 243     return RemoteDatabase::open_leaf_post_list(term, false);
 244 }
 245
 246 LeafPostList *
 247 RemoteDatabase::open_leaf_post_list(const string& term, bool) const
 248 {
 249     return new NetworkPostList(intrusive_ptr<const RemoteDatabase>(this), term);
 250 }
 251
 252 Xapian::doccount
 253 RemoteDatabase::read_post_list(const string &term, NetworkPostList & pl) const
 254 {
 255     send_message(MSG_POSTLIST, term);
 256
 257     string message;
 258     get_message(message, REPLY_POSTLISTSTART);
 259
 260     const char * p = message.data();
 261     const char * p_end = p + message.size();
 262     Xapian::doccount termfreq;
 263     decode_length(&p, p_end, termfreq);
 264
 265     while (get_message_or_done(message, REPLY_POSTLISTITEM)) {
 266         pl.append_posting(message);
 267     }
 268
 269     return termfreq;
 270 }
 271
 272 PositionList *
 273 RemoteDatabase::open_position_list(Xapian::docid did, const string &term) const
 274 {
 275     send_message(MSG_POSITIONLIST, encode_length(did) + term);
 276
 277     Xapian::VecCOW<Xapian::termpos> positions;
 278
 279     string message;
 280     Xapian::termpos lastpos = static_cast<Xapian::termpos>(-1);
 281     while (get_message_or_done(message, REPLY_POSITIONLIST)) {
 282         const char * p = message.data();
 283         const char * p_end = p + message.size();
 284         Xapian::termpos inc;
 285         decode_length(&p, p_end, inc);
 286         lastpos += inc + 1;
 287         positions.push_back(lastpos);
 288     }
 289
 290     return new InMemoryPositionList(std::move(positions));
 291 }
 292
 293 bool
 294 RemoteDatabase::has_positions() const
 295 {
 296     if (!cached_stats_valid) update_stats();
 297     return has_positional_info;
 298 }
 299
 300 bool
 301 RemoteDatabase::reopen()
 302 {
 303     mru_slot = Xapian::BAD_VALUENO;
 304     return update_stats(MSG_REOPEN);
 305 }
 306
 307 void
 308 RemoteDatabase::close()
 309 {
 310     do_close();
 311 }
 312
 313 // Currently lazy is used:
 314 //
 315 // * To implement API flag Xapian::DOC_ASSUME_VALID which can be specified when
 316 //   calling method Database::get_document()
 317 //
 318 // * To read values for backends without streamed values in SlowValueList
 319 //
 320 // * If you call get_data(), values_begin() or values_count() on a Document
 321 //   object passed to a KeyMaker, MatchDecider, MatchSpy during the match
 322 //
 323 // The first is relevant to the remote backend, but doesn't happen during
 324 // the match.
 325 //
 326 // SlowValueList is used with the remote backend, but not to read values
 327 // during the match.
 328 //
 329 // KeyMaker and MatchSpy happens on the server with the remote backend, so
 330 // they aren't relevant here.
 331 //
 332 // So the cases which are relevant to the remote backend don't matter during
 333 // the match, and so we can ignore the lazy flag here without affecting matcher
 334 // performance.
 335 Xapian::Document::Internal *
 336 RemoteDatabase::open_document(Xapian::docid did, bool /*lazy*/) const
 337 {
 338     Assert(did);
 339
 340     send_message(MSG_DOCUMENT, encode_length(did));
 341     string doc_data;
 342     map<Xapian::valueno, string> values;
 343     get_message(doc_data, REPLY_DOCDATA);
 344
 345     string message;
 346     while (get_message_or_done(message, REPLY_VALUE)) {
 347         const char * p = message.data();
 348         const char * p_end = p + message.size();
 349         Xapian::valueno slot;
 350         decode_length(&p, p_end, slot);
 351         values.insert(make_pair(slot, string(p, p_end)));
 352     }
 353
 354     return new RemoteDocument(this, did, doc_data, std::move(values));
 355 }
 356
 357 bool
 358 RemoteDatabase::update_stats(message_type msg_code, const string & body) const
 359 {
 360     // MSG_MAX signals that we're handling the opening greeting, which isn't in
 361     // response to an explicit message.
 362     if (msg_code != MSG_MAX)
 363         send_message(msg_code, body);
 364
 365     string message;
 366     if (!get_message_or_done(message, REPLY_UPDATE)) {
 367         // The database was already open at the latest revision.
 368         return false;
 369     }
 370
 371     if (message.size() < 3) {
 372         throw_handshake_failed(context);
 373     }
 374     const char *p = message.c_str();
 375     const char *p_end = p + message.size();
 376
 377     // The protocol major versions must match.  The protocol minor version of
 378     // the server must be >= that of the client.
 379     int protocol_major = static_cast<unsigned char>(*p++);
 380     int protocol_minor = static_cast<unsigned char>(*p++);
 381     if (protocol_major != XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION ||
 382         protocol_minor < XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION) {
 383         string errmsg("Server supports protocol version");
 384         if (protocol_minor) {
 385             errmsg += "s ";
 386             errmsg += str(protocol_major);
 387             errmsg += ".0 to ";
 388         }
 389         errmsg += str(protocol_major);
 390         errmsg += '.';
 391         errmsg += str(protocol_minor);
 392         errmsg +=
 393             " - client is using "
 394             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MAJOR_VERSION)
 395             "."
 396             STRINGIZE(XAPIAN_REMOTE_PROTOCOL_MINOR_VERSION);
 397         throw Xapian::NetworkError(errmsg, context);
 398     }
 399
 400     decode_length(&p, p_end, doccount);
 401     decode_length(&p, p_end, lastdocid);
 402     lastdocid += doccount;
 403     decode_length(&p, p_end, doclen_lbound);
 404     decode_length(&p, p_end, doclen_ubound);
 405     doclen_ubound += doclen_lbound;
 406     if (p == p_end) {
 407         throw Xapian::NetworkError("Bad stats update message received", context);
 408     }
 409     has_positional_info = (*p++ == '1');
 410     decode_length(&p, p_end, total_length);
 411     uuid.assign(p, p_end);
 412     cached_stats_valid = true;
 413     return true;
 414 }
 415
 416 Xapian::doccount
 417 RemoteDatabase::get_doccount() const
 418 {
 419     if (!cached_stats_valid) update_stats();
 420     return doccount;
 421 }
 422
 423 Xapian::docid
 424 RemoteDatabase::get_lastdocid() const
 425 {
 426     if (!cached_stats_valid) update_stats();
 427     return lastdocid;
 428 }
 429
 430 Xapian::totallength
 431 RemoteDatabase::get_total_length() const
 432 {
 433     if (!cached_stats_valid) update_stats();
 434     return total_length;
 435 }
 436
 437 bool
 438 RemoteDatabase::term_exists(const string & tname) const
 439 {
 440     if (tname.empty()) {
 441         return get_doccount() != 0;
 442     }
 443     send_message(MSG_TERMEXISTS, tname);
 444     string message;
 445     reply_type type = get_message(message,
 446                                   REPLY_TERMEXISTS,
 447                                   REPLY_TERMDOESNTEXIST);
 448     return (type == REPLY_TERMEXISTS);
 449 }
 450
 451 void
 452 RemoteDatabase::get_freqs(const string & term,
 453                           Xapian::doccount * termfreq_ptr,
 454                           Xapian::termcount * collfreq_ptr) const
 455 {
 456     Assert(!term.empty());
 457     string message;
 458     const char * p;
 459     const char * p_end;
 460     if (termfreq_ptr) {
 461         if (collfreq_ptr) {
 462             send_message(MSG_FREQS, term);
 463             get_message(message, REPLY_FREQS);
 464         } else {
 465             send_message(MSG_TERMFREQ, term);
 466             get_message(message, REPLY_TERMFREQ);
 467         }
 468         p = message.data();
 469         p_end = p + message.size();
 470         decode_length(&p, p_end, *termfreq_ptr);
 471     } else if (collfreq_ptr) {
 472         send_message(MSG_COLLFREQ, term);
 473         get_message(message, REPLY_COLLFREQ);
 474         p = message.data();
 475         p_end = p + message.size();
 476     }
 477     if (collfreq_ptr) {
 478         decode_length(&p, p_end, *collfreq_ptr);
 479     }
 480 }
 481
 482 void
 483 RemoteDatabase::read_value_stats(Xapian::valueno slot) const
 484 {
 485     if (mru_slot != slot) {
 486         send_message(MSG_VALUESTATS, encode_length(slot));
 487         string message;
 488         get_message(message, REPLY_VALUESTATS);
 489         const char * p = message.data();
 490         const char * p_end = p + message.size();
 491         mru_slot = slot;
 492         decode_length(&p, p_end, mru_valstats.freq);
 493         size_t len;
 494         decode_length_and_check(&p, p_end, len);
 495         mru_valstats.lower_bound.assign(p, len);
 496         p += len;
 497         decode_length_and_check(&p, p_end, len);
 498         mru_valstats.upper_bound.assign(p, len);
 499         p += len;
 500         if (p != p_end) {
 501             throw Xapian::NetworkError("Bad REPLY_VALUESTATS message received", context);
 502         }
 503     }
 504 }
 505
 506 Xapian::doccount
 507 RemoteDatabase::get_value_freq(Xapian::valueno slot) const
 508 {
 509     read_value_stats(slot);
 510     return mru_valstats.freq;
 511 }
 512
 513 std::string
 514 RemoteDatabase::get_value_lower_bound(Xapian::valueno slot) const
 515 {
 516     read_value_stats(slot);
 517     return mru_valstats.lower_bound;
 518 }
 519
 520 std::string
 521 RemoteDatabase::get_value_upper_bound(Xapian::valueno slot) const
 522 {
 523     read_value_stats(slot);
 524     return mru_valstats.upper_bound;
 525 }
 526
 527 Xapian::termcount
 528 RemoteDatabase::get_doclength_lower_bound() const
 529 {
 530     return doclen_lbound;
 531 }
 532
 533 Xapian::termcount
 534 RemoteDatabase::get_doclength_upper_bound() const
 535 {
 536     return doclen_ubound;
 537 }
 538
 539 Xapian::termcount
 540 RemoteDatabase::get_wdf_upper_bound(const string &) const
 541 {
 542     // The default implementation returns get_collection_freq(), but we
 543     // don't want the overhead of a remote message and reply per query
 544     // term, and we can get called in the middle of a remote exchange
 545     // too.  FIXME: handle this bound in the stats local/remote code...
 546     return doclen_ubound;
 547 }
 548
 549 Xapian::termcount
 550 RemoteDatabase::get_doclength(Xapian::docid did) const
 551 {
 552     Assert(did != 0);
 553     send_message(MSG_DOCLENGTH, encode_length(did));
 554     string message;
 555     get_message(message, REPLY_DOCLENGTH);
 556     const char * p = message.c_str();
 557     const char * p_end = p + message.size();
 558     Xapian::termcount doclen;
 559     decode_length(&p, p_end, doclen);
 560     if (p != p_end) {
 561         throw Xapian::NetworkError("Bad REPLY_DOCLENGTH message received", context);
 562     }
 563     return doclen;
 564 }
 565
 566 Xapian::termcount
 567 RemoteDatabase::get_unique_terms(Xapian::docid did) const
 568 {
 569     Assert(did != 0);
 570     send_message(MSG_UNIQUETERMS, encode_length(did));
 571     string message;
 572     get_message(message, REPLY_UNIQUETERMS);
 573     const char * p = message.c_str();
 574     const char * p_end = p + message.size();
 575     Xapian::termcount doclen;
 576     decode_length(&p, p_end, doclen);
 577     if (p != p_end) {
 578         throw Xapian::NetworkError("Bad REPLY_UNIQUETERMS message received", context);
 579     }
 580     return doclen;
 581 }
 582
 583 reply_type
 584 RemoteDatabase::get_message(string &result,
 585                             reply_type required_type,
 586                             reply_type required_type2) const
 587 {
 588     double end_time = RealTime::end_time(timeout);
 589     int type = link.get_message(result, end_time);
 590     if (type < 0)
 591         throw_connection_closed_unexpectedly();
 592     if (rare(type) >= REPLY_MAX) {
 593         if (required_type == REPLY_UPDATE)
 594             throw_handshake_failed(context);
 595         string errmsg("Invalid reply type ");
 596         errmsg += str(type);
 597         throw Xapian::NetworkError(errmsg);
 598     }
 599     if (type == REPLY_EXCEPTION) {
 600         unserialise_error(result, "REMOTE:", context);
 601     }
 602     if (type != required_type && type != required_type2) {
 603         string errmsg("Expecting reply type ");
 604         errmsg += str(int(required_type));
 605         if (required_type2 != required_type) {
 606             errmsg += " or ";
 607             errmsg += str(int(required_type2));
 608         }
 609         errmsg += ", got ";
 610         errmsg += str(type);
 611         throw Xapian::NetworkError(errmsg);
 612     }
 613
 614     return static_cast<reply_type>(type);
 615 }
 616
 617 void
 618 RemoteDatabase::send_message(message_type type, const string &message) const
 619 {
 620     double end_time = RealTime::end_time(timeout);
 621     link.send_message(static_cast<unsigned char>(type), message, end_time);
 622 }
 623
 624 void
 625 RemoteDatabase::do_close()
 626 {
 627     bool writable = !is_read_only();
 628
 629     // The dtor hasn't really been called!  FIXME: This works, but means any
 630     // exceptions from end_transaction()/commit() are swallowed, which is
 631     // not entirely desirable.
 632     dtor_called();
 633
 634     // If we're writable, wait for a confirmation of the close, so we know that
 635     // changes have been written and flushed, and the database write lock
 636     // released.  For the non-writable case, there's no need to wait, so don't
 637     // slow down searching by waiting here.
 638     link.do_close(writable);
 639 }
 640
 641 void
 642 RemoteDatabase::set_query(const Xapian::Query& query,
 643                           Xapian::termcount qlen,
 644                           Xapian::valueno collapse_key,
 645                           Xapian::doccount collapse_max,
 646                           Xapian::Enquire::docid_order order,
 647                           Xapian::valueno sort_key,
 648                           Xapian::Enquire::Internal::sort_setting sort_by,
 649                           bool sort_value_forward,
 650                           double time_limit,
 651                           int percent_threshold, double weight_threshold,
 652                           const Xapian::Weight& wtscheme,
 653                           const Xapian::RSet &omrset,
 654                           const vector<opt_ptr_spy>& matchspies) const
 655 {
 656     string tmp = query.serialise();
 657     string message = encode_length(tmp.size());
 658     message += tmp;
 659
 660     // Serialise assorted Enquire settings.
 661     message += encode_length(qlen);
 662     message += encode_length(collapse_max);
 663     if (collapse_max) message += encode_length(collapse_key);
 664     message += char('0' + order);
 665     message += encode_length(sort_key);
 666     message += char('0' + sort_by);
 667     message += char('0' + sort_value_forward);
 668     message += serialise_double(time_limit);
 669     message += char(percent_threshold);
 670     message += serialise_double(weight_threshold);
 671
 672     tmp = wtscheme.name();
 673     message += encode_length(tmp.size());
 674     message += tmp;
 675
 676     tmp = wtscheme.serialise();
 677     message += encode_length(tmp.size());
 678     message += tmp;
 679
 680     tmp = serialise_rset(omrset);
 681     message += encode_length(tmp.size());
 682     message += tmp;
 683
 684     for (auto i : matchspies) {
 685         tmp = i->name();
 686         if (tmp.empty()) {
 687             throw Xapian::UnimplementedError("MatchSpy subclass not suitable for use with remote searches - name() method returned empty string");
 688         }
 689         message += encode_length(tmp.size());
 690         message += tmp;
 691
 692         tmp = i->serialise();
 693         message += encode_length(tmp.size());
 694         message += tmp;
 695     }
 696
 697     send_message(MSG_QUERY, message);
 698 }
 699
 700 bool
 701 RemoteDatabase::get_remote_stats(bool block,
 702                                  Xapian::Weight::Internal& out) const
 703 {
 704     if (!block && !link.ready_to_read()) return false;
 705
 706     string message;
 707     get_message(message, REPLY_STATS);
 708     unserialise_stats(message, out);
 709
 710     return true;
 711 }
 712
 713 void
 714 RemoteDatabase::send_global_stats(Xapian::doccount first,
 715                                   Xapian::doccount maxitems,
 716                                   Xapian::doccount check_at_least,
 717                                   const Xapian::Weight::Internal &stats) const
 718 {
 719     string message = encode_length(first);
 720     message += encode_length(maxitems);
 721     message += encode_length(check_at_least);
 722     message += serialise_stats(stats);
 723     send_message(MSG_GETMSET, message);
 724 }
 725
 726 Xapian::MSet
 727 RemoteDatabase::get_mset(const vector<opt_ptr_spy>& matchspies) const
 728 {
 729     string message;
 730     get_message(message, REPLY_RESULTS);
 731     const char * p = message.data();
 732     const char * p_end = p + message.size();
 733
 734     for (auto i : matchspies) {
 735         if (p == p_end)
 736             throw Xapian::NetworkError("Expected serialised matchspy");
 737         size_t len;
 738         decode_length_and_check(&p, p_end, len);
 739         string spyresults(p, len);
 740         p += len;
 741         i->merge_results(spyresults);
 742     }
 743     Xapian::MSet mset;
 744     mset.internal->unserialise(p, p_end);
 745     return mset;
 746 }
 747
 748 void
 749 RemoteDatabase::commit()
 750 {
 751     send_message(MSG_COMMIT, string());
 752
 753     // We need to wait for a response to ensure documents have been committed.
 754     string message;
 755     get_message(message, REPLY_DONE);
 756 }
 757
 758 void
 759 RemoteDatabase::cancel()
 760 {
 761     cached_stats_valid = false;
 762     mru_slot = Xapian::BAD_VALUENO;
 763
 764     send_message(MSG_CANCEL, string());
 765 }
 766
 767 Xapian::docid
 768 RemoteDatabase::add_document(const Xapian::Document & doc)
 769 {
 770     cached_stats_valid = false;
 771     mru_slot = Xapian::BAD_VALUENO;
 772
 773     send_message(MSG_ADDDOCUMENT, serialise_document(doc));
 774
 775     string message;
 776     get_message(message, REPLY_ADDDOCUMENT);
 777
 778     const char * p = message.data();
 779     const char * p_end = p + message.size();
 780     Xapian::docid did;
 781     decode_length(&p, p_end, did);
 782     return did;
 783 }
 784
 785 void
 786 RemoteDatabase::delete_document(Xapian::docid did)
 787 {
 788     cached_stats_valid = false;
 789     mru_slot = Xapian::BAD_VALUENO;
 790
 791     send_message(MSG_DELETEDOCUMENT, encode_length(did));
 792     string dummy;
 793     get_message(dummy, REPLY_DONE);
 794 }
 795
 796 void
 797 RemoteDatabase::delete_document(const std::string & unique_term)
 798 {
 799     cached_stats_valid = false;
 800     mru_slot = Xapian::BAD_VALUENO;
 801
 802     send_message(MSG_DELETEDOCUMENTTERM, unique_term);
 803 }
 804
 805 void
 806 RemoteDatabase::replace_document(Xapian::docid did,
 807                                  const Xapian::Document & doc)
 808 {
 809     cached_stats_valid = false;
 810     mru_slot = Xapian::BAD_VALUENO;
 811
 812     string message = encode_length(did);
 813     message += serialise_document(doc);
 814
 815     send_message(MSG_REPLACEDOCUMENT, message);
 816 }
 817
 818 Xapian::docid
 819 RemoteDatabase::replace_document(const std::string & unique_term,
 820                                  const Xapian::Document & doc)
 821 {
 822     cached_stats_valid = false;
 823     mru_slot = Xapian::BAD_VALUENO;
 824
 825     string message = encode_length(unique_term.size());
 826     message += unique_term;
 827     message += serialise_document(doc);
 828
 829     send_message(MSG_REPLACEDOCUMENTTERM, message);
 830
 831     get_message(message, REPLY_ADDDOCUMENT);
 832
 833     const char * p = message.data();
 834     const char * p_end = p + message.size();
 835     Xapian::docid did;
 836     decode_length(&p, p_end, did);
 837     return did;
 838 }
 839
 840 string
 841 RemoteDatabase::get_uuid() const
 842 {
 843     return uuid;
 844 }
 845
 846 string
 847 RemoteDatabase::get_metadata(const string & key) const
 848 {
 849     send_message(MSG_GETMETADATA, key);
 850     string metadata;
 851     get_message(metadata, REPLY_METADATA);
 852     return metadata;
 853 }
 854
 855 void
 856 RemoteDatabase::set_metadata(const string & key, const string & value)
 857 {
 858     string data = encode_length(key.size());
 859     data += key;
 860     data += value;
 861     send_message(MSG_SETMETADATA, data);
 862 }
 863
 864 void
 865 RemoteDatabase::add_spelling(const string & word,
 866                              Xapian::termcount freqinc) const
 867 {
 868     string data = encode_length(freqinc);
 869     data += word;
 870     send_message(MSG_ADDSPELLING, data);
 871 }
 872
 873 Xapian::termcount
 874 RemoteDatabase::remove_spelling(const string & word,
 875                                 Xapian::termcount freqdec) const
 876 {
 877     string data = encode_length(freqdec);
 878     data += word;
 879     send_message(MSG_REMOVESPELLING, data);
 880
 881     string message;
 882     get_message(message, REPLY_REMOVESPELLING);
 883     const char * p = message.data();
 884     const char * p_end = p + message.size();
 885     Xapian::termcount result;
 886     decode_length(&p, p_end, result);
 887     if (p != p_end) {
 888         throw Xapian::NetworkError("Bad REPLY_REMOVESPELLING message received", context);
 889     }
 890     return result;
 891 }
 892
 893 bool
 894 RemoteDatabase::locked() const
 895 {
 896     throw Xapian::UnimplementedError("Database::locked() not implemented for remote backend");
 897 }
 898
 899 string
 900 RemoteDatabase::get_description() const
 901 {
 902     string desc = "Remote(context=";
 903     desc += context;
 904     desc += ')';
 905     return desc;
 906 }