src/filter_multi.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 2005-2011 Index Data
   3
   4 Metaproxy is free software; you can redistribute it and/or modify it under
   5 the terms of the GNU General Public License as published by the Free
   6 Software Foundation; either version 2, or (at your option) any later
   7 version.
   8
   9 Metaproxy is distributed in the hope that it will be useful, but WITHOUT ANY
  10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  12 for more details.
  13
  14 You should have received a copy of the GNU General Public License
  15 along with this program; if not, write to the Free Software
  16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  17 */
  18
  19 #include <yaz/log.h>
  20
  21 #include "config.hpp"
  22
  23 #include <metaproxy/filter.hpp>
  24 #include <metaproxy/package.hpp>
  25
  26 #include <boost/thread/thread.hpp>
  27 #include <boost/thread/mutex.hpp>
  28 #include <boost/thread/condition.hpp>
  29 #include <boost/shared_ptr.hpp>
  30
  31 #include <metaproxy/util.hpp>
  32 #include "filter_multi.hpp"
  33
  34 #include <yaz/zgdu.h>
  35 #include <yaz/otherinfo.h>
  36 #include <yaz/diagbib1.h>
  37 #include <yaz/match_glob.h>
  38
  39 #include <vector>
  40 #include <algorithm>
  41 #include <map>
  42 #include <iostream>
  43
  44 namespace mp = metaproxy_1;
  45 namespace yf = mp::filter;
  46
  47 namespace metaproxy_1 {
  48     namespace filter {
  49         enum multi_merge_type {
  50             round_robin,
  51             serve_order
  52         };
  53         struct Multi::BackendSet {
  54             BackendPtr m_backend;
  55             int m_count;
  56             bool operator < (const BackendSet &k) const;
  57             bool operator == (const BackendSet &k) const;
  58         };
  59         struct Multi::ScanTermInfo {
  60             std::string m_norm_term;
  61             std::string m_display_term;
  62             int m_count;
  63             bool operator < (const ScanTermInfo &) const;
  64             bool operator == (const ScanTermInfo &) const;
  65             Z_Entry *get_entry(ODR odr);
  66         };
  67         struct Multi::FrontendSet {
  68             class PresentJob {
  69             public:
  70                 BackendPtr m_backend;
  71                 int m_pos; // position for backend (1=first, 2=second,..
  72                 int m_start; // present request start
  73                 PresentJob(BackendPtr ptr, int pos) :
  74                     m_backend(ptr), m_pos(pos), m_start(0) {};
  75             };
  76             FrontendSet(std::string setname);
  77             FrontendSet();
  78             ~FrontendSet();
  79
  80             void round_robin(int pos, int number, std::list<PresentJob> &job);
  81             void serve_order(int pos, int number, std::list<PresentJob> &job);
  82
  83             std::list<BackendSet> m_backend_sets;
  84             std::string m_setname;
  85         };
  86         struct Multi::Backend {
  87             PackagePtr m_package;
  88             std::string m_backend_database;
  89             std::string m_vhost;
  90             std::string m_route;
  91             void operator() (void);  // thread operation
  92         };
  93         struct Multi::Frontend {
  94             Frontend(Rep *rep);
  95             ~Frontend();
  96             bool m_is_multi;
  97             bool m_in_use;
  98             std::list<BackendPtr> m_backend_list;
  99             std::map<std::string,Multi::FrontendSet> m_sets;
 100
 101             void multi_move(std::list<BackendPtr> &blist);
 102             void init(Package &package, Z_GDU *gdu);
 103             void close(Package &package);
 104             void search(Package &package, Z_APDU *apdu);
 105             void present(Package &package, Z_APDU *apdu);
 106             void scan1(Package &package, Z_APDU *apdu);
 107             void scan2(Package &package, Z_APDU *apdu);
 108             Rep *m_p;
 109         };
 110         class Multi::Map {
 111             std::string m_target_pattern;
 112             std::string m_route;
 113         public:
 114             Map(std::string pattern, std::string route) :
 115                 m_target_pattern(pattern), m_route(route) {};
 116             bool match(const std::string target, std::string *ret) const {
 117                 if (yaz_match_glob(m_target_pattern.c_str(), target.c_str()))
 118                 {
 119                     *ret = m_route;
 120                     return true;
 121                 }
 122                 return false;
 123             };
 124         };
 125         class Multi::Rep {
 126             friend class Multi;
 127             friend struct Frontend;
 128
 129             Rep();
 130             FrontendPtr get_frontend(Package &package);
 131             void release_frontend(Package &package);
 132         private:
 133             std::list<Multi::Map> m_route_patterns;
 134             boost::mutex m_mutex;
 135             boost::condition m_cond_session_ready;
 136             std::map<mp::Session, FrontendPtr> m_clients;
 137             bool m_hide_unavailable;
 138             multi_merge_type m_merge_type;
 139         };
 140     }
 141 }
 142
 143 yf::Multi::Rep::Rep()
 144 {
 145     m_hide_unavailable = false;
 146     m_merge_type = round_robin;
 147 }
 148
 149 bool yf::Multi::BackendSet::operator < (const BackendSet &k) const
 150 {
 151     return m_count < k.m_count;
 152 }
 153
 154 yf::Multi::Frontend::Frontend(Rep *rep)
 155 {
 156     m_p = rep;
 157     m_is_multi = false;
 158 }
 159
 160 yf::Multi::Frontend::~Frontend()
 161 {
 162 }
 163
 164 yf::Multi::FrontendPtr yf::Multi::Rep::get_frontend(mp::Package &package)
 165 {
 166     boost::mutex::scoped_lock lock(m_mutex);
 167
 168     std::map<mp::Session,yf::Multi::FrontendPtr>::iterator it;
 169
 170     while(true)
 171     {
 172         it = m_clients.find(package.session());
 173         if (it == m_clients.end())
 174             break;
 175
 176         if (!it->second->m_in_use)
 177         {
 178             it->second->m_in_use = true;
 179             return it->second;
 180         }
 181         m_cond_session_ready.wait(lock);
 182     }
 183     FrontendPtr f(new Frontend(this));
 184     m_clients[package.session()] = f;
 185     f->m_in_use = true;
 186     return f;
 187 }
 188
 189 void yf::Multi::Rep::release_frontend(mp::Package &package)
 190 {
 191     boost::mutex::scoped_lock lock(m_mutex);
 192     std::map<mp::Session,yf::Multi::FrontendPtr>::iterator it;
 193
 194     it = m_clients.find(package.session());
 195     if (it != m_clients.end())
 196     {
 197         if (package.session().is_closed())
 198         {
 199             it->second->close(package);
 200             m_clients.erase(it);
 201         }
 202         else
 203         {
 204             it->second->m_in_use = false;
 205         }
 206         m_cond_session_ready.notify_all();
 207     }
 208 }
 209
 210 yf::Multi::FrontendSet::FrontendSet(std::string setname)
 211     :  m_setname(setname)
 212 {
 213 }
 214
 215
 216 yf::Multi::FrontendSet::FrontendSet()
 217 {
 218 }
 219
 220
 221 yf::Multi::FrontendSet::~FrontendSet()
 222 {
 223 }
 224
 225 yf::Multi::Multi() : m_p(new Multi::Rep)
 226 {
 227 }
 228
 229 yf::Multi::~Multi() {
 230 }
 231
 232
 233 void yf::Multi::Backend::operator() (void)
 234 {
 235     m_package->move(m_route);
 236 }
 237
 238
 239 void yf::Multi::Frontend::close(mp::Package &package)
 240 {
 241     std::list<BackendPtr>::const_iterator bit;
 242     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 243     {
 244         BackendPtr b = *bit;
 245
 246         b->m_package->copy_filter(package);
 247         b->m_package->request() = (Z_GDU *) 0;
 248         b->m_package->session().close();
 249         b->m_package->move(b->m_route);
 250     }
 251 }
 252
 253 void yf::Multi::Frontend::multi_move(std::list<BackendPtr> &blist)
 254 {
 255     std::list<BackendPtr>::const_iterator bit;
 256     boost::thread_group g;
 257     for (bit = blist.begin(); bit != blist.end(); bit++)
 258     {
 259         g.add_thread(new boost::thread(**bit));
 260     }
 261     g.join_all();
 262 }
 263
 264 void yf::Multi::FrontendSet::serve_order(int start, int number,
 265                                          std::list<PresentJob> &jobs)
 266 {
 267     int i;
 268     for (i = 0; i < number; i++)
 269     {
 270         std::list<BackendSet>::const_iterator bsit;
 271         int voffset = 0;
 272         int offset = start + i - 1;
 273         for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end();
 274              bsit++)
 275         {
 276             if (offset >= voffset && offset < voffset + bsit->m_count)
 277             {
 278                 PresentJob job(bsit->m_backend, offset - voffset + 1);
 279                 jobs.push_back(job);
 280                 break;
 281             }
 282             voffset += bsit->m_count;
 283         }
 284     }
 285 }
 286
 287 void yf::Multi::FrontendSet::round_robin(int start, int number,
 288                                          std::list<PresentJob> &jobs)
 289 {
 290     std::list<int> pos;
 291     std::list<BackendSet>::const_iterator bsit;
 292     for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end(); bsit++)
 293     {
 294         pos.push_back(1);
 295     }
 296
 297     int p = 1;
 298 #if 1
 299     // optimization step!
 300     int omin = 0;
 301     while(true)
 302     {
 303         int min = 0;
 304         int no_left = 0;
 305         // find min count for each set which is > omin
 306         for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end(); bsit++)
 307         {
 308             if (bsit->m_count > omin)
 309             {
 310                 if (no_left == 0 || bsit->m_count < min)
 311                     min = bsit->m_count;
 312                 no_left++;
 313             }
 314         }
 315         if (no_left == 0) // if nothing greater than omin, bail out.
 316             break;
 317         int skip = no_left * min;
 318         if (p + skip > start)  // step gets us "into" present range?
 319         {
 320             // Yes. skip until start.. Rounding off is deliberate!
 321             min = (start-p) / no_left;
 322             p += no_left * min;
 323
 324             // update positions in each set..
 325             std::list<int>::iterator psit = pos.begin();
 326             for (psit = pos.begin(); psit != pos.end(); psit++)
 327                 *psit += min;
 328             break;
 329         }
 330         // skip on each set.. before "present range"..
 331         p = p + skip;
 332
 333         std::list<int>::iterator psit = pos.begin();
 334         for (psit = pos.begin(); psit != pos.end(); psit++)
 335             *psit += min;
 336
 337         omin = min; // update so we consider next class (with higher count)
 338     }
 339 #endif
 340     int fetched = 0;
 341     bool more = true;
 342     while (more)
 343     {
 344         more = false;
 345         std::list<int>::iterator psit = pos.begin();
 346         bsit = m_backend_sets.begin();
 347
 348         for (; bsit != m_backend_sets.end(); psit++,bsit++)
 349         {
 350             if (fetched >= number)
 351             {
 352                 more = false;
 353                 break;
 354             }
 355             if (*psit <= bsit->m_count)
 356             {
 357                 if (p >= start)
 358                 {
 359                     PresentJob job(bsit->m_backend, *psit);
 360                     jobs.push_back(job);
 361                     fetched++;
 362                 }
 363                 (*psit)++;
 364                 p++;
 365                 more = true;
 366             }
 367         }
 368     }
 369 }
 370
 371 void yf::Multi::Frontend::init(mp::Package &package, Z_GDU *gdu)
 372 {
 373     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 374
 375     std::list<std::string> targets;
 376
 377     mp::util::get_vhost_otherinfo(req->otherInfo, targets);
 378
 379     if (targets.size() < 1)
 380     {
 381         package.move();
 382         return;
 383     }
 384
 385     std::list<std::string>::const_iterator t_it = targets.begin();
 386     for (; t_it != targets.end(); t_it++)
 387     {
 388         Session s;
 389         Backend *b = new Backend;
 390         b->m_vhost = *t_it;
 391
 392         std::list<Multi::Map>::const_iterator it =
 393             m_p->m_route_patterns.begin();
 394         while (it != m_p->m_route_patterns.end()) {
 395             if (it->match(*t_it, &b->m_route))
 396                 break;
 397             it++;
 398         }
 399         // b->m_route = m_p->m_target_route[*t_it];
 400         // b->m_route unset
 401         b->m_package = PackagePtr(new Package(s, package.origin()));
 402
 403         m_backend_list.push_back(BackendPtr(b));
 404     }
 405     m_is_multi = true;
 406
 407     // create init request
 408     std::list<BackendPtr>::iterator bit;
 409     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 410     {
 411         mp::odr odr;
 412         BackendPtr b = *bit;
 413         Z_APDU *init_apdu = zget_APDU(odr, Z_APDU_initRequest);
 414
 415         std::list<std::string>vhost_one;
 416         vhost_one.push_back(b->m_vhost);
 417         mp::util::set_vhost_otherinfo(&init_apdu->u.initRequest->otherInfo,
 418                                        odr, vhost_one);
 419
 420
 421         Z_InitRequest *breq = init_apdu->u.initRequest;
 422
 423         breq->idAuthentication = req->idAuthentication;
 424
 425         *breq->preferredMessageSize = *req->preferredMessageSize;
 426         *breq->maximumRecordSize = *req->maximumRecordSize;
 427
 428         ODR_MASK_SET(breq->options, Z_Options_search);
 429         ODR_MASK_SET(breq->options, Z_Options_present);
 430         ODR_MASK_SET(breq->options, Z_Options_namedResultSets);
 431         ODR_MASK_SET(breq->options, Z_Options_scan);
 432
 433         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_1);
 434         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_2);
 435         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_3);
 436
 437         b->m_package->request() = init_apdu;
 438
 439         b->m_package->copy_filter(package);
 440     }
 441     multi_move(m_backend_list);
 442
 443     // create the frontend init response based on each backend init response
 444     mp::odr odr;
 445
 446     Z_APDU *f_apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
 447     Z_InitResponse *f_resp = f_apdu->u.initResponse;
 448
 449     ODR_MASK_SET(f_resp->options, Z_Options_search);
 450     ODR_MASK_SET(f_resp->options, Z_Options_present);
 451     ODR_MASK_SET(f_resp->options, Z_Options_namedResultSets);
 452     ODR_MASK_SET(f_resp->options, Z_Options_scan);
 453
 454     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_1);
 455     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_2);
 456     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_3);
 457
 458     int no_failed = 0;
 459     int no_succeeded = 0;
 460
 461     Odr_int preferredMessageSize = *req->preferredMessageSize;
 462     Odr_int maximumRecordSize = *req->maximumRecordSize;
 463     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); )
 464     {
 465         PackagePtr p = (*bit)->m_package;
 466
 467         if (p->session().is_closed())
 468         {
 469             // failed. Remove from list and increment number of failed
 470             no_failed++;
 471             bit = m_backend_list.erase(bit);
 472             continue;
 473         }
 474         Z_GDU *gdu = p->response().get();
 475         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 476             Z_APDU_initResponse)
 477         {
 478             int i;
 479             Z_APDU *b_apdu = gdu->u.z3950;
 480             Z_InitResponse *b_resp = b_apdu->u.initResponse;
 481
 482             // common options for all backends
 483             for (i = 0; i <= Z_Options_stringSchema; i++)
 484             {
 485                 if (!ODR_MASK_GET(b_resp->options, i))
 486                     ODR_MASK_CLEAR(f_resp->options, i);
 487             }
 488             // common protocol version
 489             for (i = 0; i <= Z_ProtocolVersion_3; i++)
 490                 if (!ODR_MASK_GET(b_resp->protocolVersion, i))
 491                     ODR_MASK_CLEAR(f_resp->protocolVersion, i);
 492             if (*b_resp->result)
 493             {
 494                 no_succeeded++;
 495                 if (preferredMessageSize > *b_resp->preferredMessageSize)
 496                     preferredMessageSize = *b_resp->preferredMessageSize;
 497                 if (maximumRecordSize > *b_resp->maximumRecordSize)
 498                     maximumRecordSize = *b_resp->maximumRecordSize;
 499             }
 500             else
 501                 no_failed++;
 502         }
 503         else
 504             no_failed++;
 505         bit++;
 506     }
 507     *f_resp->preferredMessageSize = preferredMessageSize;
 508     *f_resp->maximumRecordSize = maximumRecordSize;
 509
 510     if (m_p->m_hide_unavailable)
 511     {
 512         if (no_succeeded == 0)
 513         {
 514             *f_resp->result = 0;
 515             package.session().close();
 516         }
 517     }
 518     else
 519     {
 520         if (no_failed)
 521         {
 522             *f_resp->result = 0;
 523             package.session().close();
 524         }
 525     }
 526     package.response() = f_apdu;
 527 }
 528
 529 void yf::Multi::Frontend::search(mp::Package &package, Z_APDU *apdu_req)
 530 {
 531     // create search request
 532     Z_SearchRequest *req = apdu_req->u.searchRequest;
 533
 534     // save these for later
 535     int smallSetUpperBound = *req->smallSetUpperBound;
 536     int largeSetLowerBound = *req->largeSetLowerBound;
 537     int mediumSetPresentNumber = *req->mediumSetPresentNumber;
 538
 539     // they are altered now - to disable piggyback
 540     *req->smallSetUpperBound = 0;
 541     *req->largeSetLowerBound = 1;
 542     *req->mediumSetPresentNumber = 1;
 543
 544     int default_num_db = req->num_databaseNames;
 545     char **default_db = req->databaseNames;
 546
 547     std::list<BackendPtr>::const_iterator bit;
 548     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 549     {
 550         PackagePtr p = (*bit)->m_package;
 551         mp::odr odr;
 552
 553         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
 554                                                 &req->num_databaseNames,
 555                                                 &req->databaseNames))
 556         {
 557             req->num_databaseNames = default_num_db;
 558             req->databaseNames = default_db;
 559         }
 560         p->request() = apdu_req;
 561         p->copy_filter(package);
 562     }
 563     multi_move(m_backend_list);
 564
 565     // look at each response
 566     FrontendSet resultSet(std::string(req->resultSetName));
 567
 568     int result_set_size = 0;
 569     Z_Records *z_records_diag = 0;  // no diagnostics (yet)
 570     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 571     {
 572         PackagePtr p = (*bit)->m_package;
 573
 574         if (p->session().is_closed()) // if any backend closes, close frontend
 575             package.session().close();
 576
 577         Z_GDU *gdu = p->response().get();
 578         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 579             Z_APDU_searchResponse)
 580         {
 581             Z_APDU *b_apdu = gdu->u.z3950;
 582             Z_SearchResponse *b_resp = b_apdu->u.searchResponse;
 583
 584             // see we get any errors (AKA diagnstics)
 585             if (b_resp->records)
 586             {
 587                 if (b_resp->records->which == Z_Records_NSD
 588                     || b_resp->records->which == Z_Records_multipleNSD)
 589                     z_records_diag = b_resp->records;
 590                 // we may set this multiple times (TOO BAD!)
 591             }
 592             BackendSet backendSet;
 593             backendSet.m_backend = *bit;
 594             backendSet.m_count = *b_resp->resultCount;
 595             result_set_size += *b_resp->resultCount;
 596             resultSet.m_backend_sets.push_back(backendSet);
 597         }
 598         else
 599         {
 600             // if any target does not return search response - return that
 601             package.response() = p->response();
 602             return;
 603         }
 604     }
 605
 606     mp::odr odr;
 607     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
 608     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 609
 610     *f_resp->resultCount = result_set_size;
 611     if (z_records_diag)
 612     {
 613         // search error
 614         f_resp->records = z_records_diag;
 615         package.response() = f_apdu;
 616         return;
 617     }
 618     // assume OK
 619     m_sets[resultSet.m_setname] = resultSet;
 620
 621     int number;
 622     mp::util::piggyback(smallSetUpperBound,
 623                          largeSetLowerBound,
 624                          mediumSetPresentNumber,
 625                          result_set_size,
 626                          number);
 627     Package pp(package.session(), package.origin());
 628     if (number > 0)
 629     {
 630         pp.copy_filter(package);
 631         Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
 632         Z_PresentRequest *p_req = p_apdu->u.presentRequest;
 633         p_req->preferredRecordSyntax = req->preferredRecordSyntax;
 634         p_req->resultSetId = req->resultSetName;
 635         *p_req->resultSetStartPoint = 1;
 636         *p_req->numberOfRecordsRequested = number;
 637         pp.request() = p_apdu;
 638         present(pp, p_apdu);
 639
 640         if (pp.session().is_closed())
 641             package.session().close();
 642
 643         Z_GDU *gdu = pp.response().get();
 644         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 645             Z_APDU_presentResponse)
 646         {
 647             Z_PresentResponse *p_res = gdu->u.z3950->u.presentResponse;
 648             f_resp->records = p_res->records;
 649             *f_resp->numberOfRecordsReturned =
 650                 *p_res->numberOfRecordsReturned;
 651             *f_resp->nextResultSetPosition =
 652                 *p_res->nextResultSetPosition;
 653         }
 654         else
 655         {
 656             package.response() = pp.response();
 657             return;
 658         }
 659     }
 660     package.response() = f_apdu; // in this scope because of p
 661 }
 662
 663 void yf::Multi::Frontend::present(mp::Package &package, Z_APDU *apdu_req)
 664 {
 665     // create present request
 666     Z_PresentRequest *req = apdu_req->u.presentRequest;
 667
 668     Sets_it it;
 669     it = m_sets.find(std::string(req->resultSetId));
 670     if (it == m_sets.end())
 671     {
 672         mp::odr odr;
 673         Z_APDU *apdu =
 674             odr.create_presentResponse(
 675                 apdu_req,
 676                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
 677                 req->resultSetId);
 678         package.response() = apdu;
 679         return;
 680     }
 681     std::list<Multi::FrontendSet::PresentJob> jobs;
 682     int start = *req->resultSetStartPoint;
 683     int number = *req->numberOfRecordsRequested;
 684
 685     if (m_p->m_merge_type == round_robin)
 686         it->second.round_robin(start, number, jobs);
 687     else if (m_p->m_merge_type == serve_order)
 688         it->second.serve_order(start, number, jobs);
 689
 690     if (0)
 691     {
 692         std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
 693         for (jit = jobs.begin(); jit != jobs.end(); jit++)
 694         {
 695             yaz_log(YLOG_LOG, "job pos=%d", jit->m_pos);
 696         }
 697     }
 698
 699     std::list<BackendPtr> present_backend_list;
 700
 701     std::list<BackendSet>::const_iterator bsit;
 702     bsit = it->second.m_backend_sets.begin();
 703     for (; bsit != it->second.m_backend_sets.end(); bsit++)
 704     {
 705         int start = -1;
 706         int end = -1;
 707         {
 708             std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
 709             for (jit = jobs.begin(); jit != jobs.end(); jit++)
 710             {
 711                 if (jit->m_backend == bsit->m_backend)
 712                 {
 713                     if (start == -1 || jit->m_pos < start)
 714                         start = jit->m_pos;
 715                     if (end == -1 || jit->m_pos > end)
 716                         end = jit->m_pos;
 717                 }
 718             }
 719         }
 720         if (start != -1)
 721         {
 722             std::list<Multi::FrontendSet::PresentJob>::iterator jit;
 723             for (jit = jobs.begin(); jit != jobs.end(); jit++)
 724             {
 725                 if (jit->m_backend == bsit->m_backend)
 726                 {
 727                     if (jit->m_pos >= start && jit->m_pos <= end)
 728                         jit->m_start = start;
 729                 }
 730             }
 731
 732             PackagePtr p = bsit->m_backend->m_package;
 733
 734             *req->resultSetStartPoint = start;
 735             *req->numberOfRecordsRequested = end - start + 1;
 736
 737             p->request() = apdu_req;
 738             p->copy_filter(package);
 739
 740             present_backend_list.push_back(bsit->m_backend);
 741         }
 742     }
 743     multi_move(present_backend_list);
 744
 745     // look at each response
 746     Z_Records *z_records_diag = 0;
 747
 748     std::list<BackendPtr>::const_iterator pbit = present_backend_list.begin();
 749     for (; pbit != present_backend_list.end(); pbit++)
 750     {
 751         PackagePtr p = (*pbit)->m_package;
 752
 753         if (p->session().is_closed()) // if any backend closes, close frontend
 754             package.session().close();
 755
 756         Z_GDU *gdu = p->response().get();
 757         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 758             Z_APDU_presentResponse)
 759         {
 760             Z_APDU *b_apdu = gdu->u.z3950;
 761             Z_PresentResponse *b_resp = b_apdu->u.presentResponse;
 762
 763             // see we get any errors (AKA diagnstics)
 764             if (b_resp->records)
 765             {
 766                 if (b_resp->records->which != Z_Records_DBOSD)
 767                     z_records_diag = b_resp->records;
 768                 // we may set this multiple times (TOO BAD!)
 769             }
 770         }
 771         else
 772         {
 773             // if any target does not return present response - return that
 774             package.response() = p->response();
 775             return;
 776         }
 777     }
 778
 779     mp::odr odr;
 780     Z_APDU *f_apdu = odr.create_presentResponse(apdu_req, 0, 0);
 781     Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 782
 783     if (z_records_diag)
 784     {
 785         f_resp->records = z_records_diag;
 786         *f_resp->presentStatus = Z_PresentStatus_failure;
 787     }
 788     else if (number < 0 || (size_t) number > jobs.size())
 789     {
 790         f_apdu =
 791             odr.create_presentResponse(
 792                 apdu_req,
 793                 YAZ_BIB1_PRESENT_REQUEST_OUT_OF_RANGE,
 794                 0);
 795     }
 796     else
 797     {
 798         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
 799         Z_Records * records = f_resp->records;
 800         records->which = Z_Records_DBOSD;
 801         records->u.databaseOrSurDiagnostics =
 802             (Z_NamePlusRecordList *)
 803             odr_malloc(odr, sizeof(Z_NamePlusRecordList));
 804         Z_NamePlusRecordList *nprl = records->u.databaseOrSurDiagnostics;
 805         nprl->num_records = jobs.size();
 806         nprl->records = (Z_NamePlusRecord**)
 807             odr_malloc(odr, sizeof(Z_NamePlusRecord *) * nprl->num_records);
 808         int i = 0;
 809         std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
 810         for (jit = jobs.begin(); jit != jobs.end(); jit++, i++)
 811         {
 812             PackagePtr p = jit->m_backend->m_package;
 813
 814             Z_GDU *gdu = p->response().get();
 815             Z_APDU *b_apdu = gdu->u.z3950;
 816             Z_PresentResponse *b_resp = b_apdu->u.presentResponse;
 817
 818             nprl->records[i] = (Z_NamePlusRecord*)
 819                 odr_malloc(odr, sizeof(Z_NamePlusRecord));
 820             int inside_pos = jit->m_pos - jit->m_start;
 821             if (inside_pos >= b_resp->records->
 822                 u.databaseOrSurDiagnostics->num_records)
 823                 break;
 824             *nprl->records[i] = *b_resp->records->
 825                 u.databaseOrSurDiagnostics->records[inside_pos];
 826             nprl->records[i]->databaseName =
 827                     odr_strdup(odr, jit->m_backend->m_vhost.c_str());
 828         }
 829         nprl->num_records = i; // usually same as jobs.size();
 830         *f_resp->nextResultSetPosition = start + i;
 831         *f_resp->numberOfRecordsReturned = i;
 832     }
 833     package.response() = f_apdu;
 834 }
 835
 836 void yf::Multi::Frontend::scan1(mp::Package &package, Z_APDU *apdu_req)
 837 {
 838     if (m_backend_list.size() > 1)
 839     {
 840         mp::odr odr;
 841         Z_APDU *f_apdu =
 842             odr.create_scanResponse(
 843                 apdu_req, YAZ_BIB1_COMBI_OF_SPECIFIED_DATABASES_UNSUPP, 0);
 844         package.response() = f_apdu;
 845         return;
 846     }
 847     Z_ScanRequest *req = apdu_req->u.scanRequest;
 848
 849     int default_num_db = req->num_databaseNames;
 850     char **default_db = req->databaseNames;
 851
 852     std::list<BackendPtr>::const_iterator bit;
 853     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 854     {
 855         PackagePtr p = (*bit)->m_package;
 856         mp::odr odr;
 857
 858         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
 859                                                 &req->num_databaseNames,
 860                                                 &req->databaseNames))
 861         {
 862             req->num_databaseNames = default_num_db;
 863             req->databaseNames = default_db;
 864         }
 865         p->request() = apdu_req;
 866         p->copy_filter(package);
 867     }
 868     multi_move(m_backend_list);
 869
 870     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 871     {
 872         PackagePtr p = (*bit)->m_package;
 873
 874         if (p->session().is_closed()) // if any backend closes, close frontend
 875             package.session().close();
 876
 877         Z_GDU *gdu = p->response().get();
 878         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 879             Z_APDU_scanResponse)
 880         {
 881             package.response() = p->response();
 882             break;
 883         }
 884         else
 885         {
 886             // if any target does not return scan response - return that
 887             package.response() = p->response();
 888             return;
 889         }
 890     }
 891 }
 892
 893 bool yf::Multi::ScanTermInfo::operator < (const ScanTermInfo &k) const
 894 {
 895     return m_norm_term < k.m_norm_term;
 896 }
 897
 898 bool yf::Multi::ScanTermInfo::operator == (const ScanTermInfo &k) const
 899 {
 900     return m_norm_term == k.m_norm_term;
 901 }
 902
 903 Z_Entry *yf::Multi::ScanTermInfo::get_entry(ODR odr)
 904 {
 905     Z_Entry *e = (Z_Entry *)odr_malloc(odr, sizeof(*e));
 906     e->which = Z_Entry_termInfo;
 907     Z_TermInfo *t;
 908     t = e->u.termInfo = (Z_TermInfo *) odr_malloc(odr, sizeof(*t));
 909     t->suggestedAttributes = 0;
 910     t->displayTerm = 0;
 911     t->alternativeTerm = 0;
 912     t->byAttributes = 0;
 913     t->otherTermInfo = 0;
 914     t->globalOccurrences = odr_intdup(odr, m_count);
 915     t->term = (Z_Term *)
 916         odr_malloc(odr, sizeof(*t->term));
 917     t->term->which = Z_Term_general;
 918     Odr_oct *o;
 919     t->term->u.general = o = (Odr_oct *)odr_malloc(odr, sizeof(Odr_oct));
 920
 921     o->len = o->size = m_norm_term.size();
 922     o->buf = (unsigned char *) odr_malloc(odr, o->len);
 923     memcpy(o->buf, m_norm_term.c_str(), o->len);
 924     return e;
 925 }
 926
 927 void yf::Multi::Frontend::scan2(mp::Package &package, Z_APDU *apdu_req)
 928 {
 929     Z_ScanRequest *req = apdu_req->u.scanRequest;
 930
 931     int default_num_db = req->num_databaseNames;
 932     char **default_db = req->databaseNames;
 933
 934     std::list<BackendPtr>::const_iterator bit;
 935     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 936     {
 937         PackagePtr p = (*bit)->m_package;
 938         mp::odr odr;
 939
 940         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
 941                                                 &req->num_databaseNames,
 942                                                 &req->databaseNames))
 943         {
 944             req->num_databaseNames = default_num_db;
 945             req->databaseNames = default_db;
 946         }
 947         p->request() = apdu_req;
 948         p->copy_filter(package);
 949     }
 950     multi_move(m_backend_list);
 951
 952     ScanTermInfoList entries_before;
 953     ScanTermInfoList entries_after;
 954     int no_before = 0;
 955     int no_after = 0;
 956
 957     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
 958     {
 959         PackagePtr p = (*bit)->m_package;
 960
 961         if (p->session().is_closed()) // if any backend closes, close frontend
 962             package.session().close();
 963
 964         Z_GDU *gdu = p->response().get();
 965         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 966             Z_APDU_scanResponse)
 967         {
 968             Z_ScanResponse *res = gdu->u.z3950->u.scanResponse;
 969
 970             if (res->entries && res->entries->nonsurrogateDiagnostics)
 971             {
 972                 // failure
 973                 mp::odr odr;
 974                 Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 1, 0);
 975                 Z_ScanResponse *f_res = f_apdu->u.scanResponse;
 976
 977                 f_res->entries->nonsurrogateDiagnostics =
 978                     res->entries->nonsurrogateDiagnostics;
 979                 f_res->entries->num_nonsurrogateDiagnostics =
 980                     res->entries->num_nonsurrogateDiagnostics;
 981
 982                 package.response() = f_apdu;
 983                 return;
 984             }
 985
 986             if (res->entries && res->entries->entries)
 987             {
 988                 Z_Entry **entries = res->entries->entries;
 989                 int num_entries = res->entries->num_entries;
 990                 int position = 1;
 991                 if (req->preferredPositionInResponse)
 992                     position = *req->preferredPositionInResponse;
 993                 if (res->positionOfTerm)
 994                     position = *res->positionOfTerm;
 995
 996                 // before
 997                 int i;
 998                 for (i = 0; i<position-1 && i<num_entries; i++)
 999                 {
1000                     Z_Entry *ent = entries[i];
1001
1002                     if (ent->which == Z_Entry_termInfo)
1003                     {
1004                         ScanTermInfo my;
1005
1006                         Odr_int *occur = ent->u.termInfo->globalOccurrences;
1007                         my.m_count = occur ? *occur : 0;
1008
1009                         if (ent->u.termInfo->term->which == Z_Term_general)
1010                         {
1011                             my.m_norm_term = std::string(
1012                                 (const char *)
1013                                 ent->u.termInfo->term->u.general->buf,
1014                                 ent->u.termInfo->term->u.general->len);
1015                         }
1016                         if (my.m_norm_term.length())
1017                         {
1018                             ScanTermInfoList::iterator it =
1019                                 entries_before.begin();
1020                             while (it != entries_before.end() && my <*it)
1021                                 it++;
1022                             if (my == *it)
1023                             {
1024                                 it->m_count += my.m_count;
1025                             }
1026                             else
1027                             {
1028                                 entries_before.insert(it, my);
1029                                 no_before++;
1030                             }
1031                         }
1032                     }
1033                 }
1034                 // after
1035                 if (position <= 0)
1036                     i = 0;
1037                 else
1038                     i = position-1;
1039                 for ( ; i<num_entries; i++)
1040                 {
1041                     Z_Entry *ent = entries[i];
1042
1043                     if (ent->which == Z_Entry_termInfo)
1044                     {
1045                         ScanTermInfo my;
1046
1047                         Odr_int *occur = ent->u.termInfo->globalOccurrences;
1048                         my.m_count = occur ? *occur : 0;
1049
1050                         if (ent->u.termInfo->term->which == Z_Term_general)
1051                         {
1052                             my.m_norm_term = std::string(
1053                                 (const char *)
1054                                 ent->u.termInfo->term->u.general->buf,
1055                                 ent->u.termInfo->term->u.general->len);
1056                         }
1057                         if (my.m_norm_term.length())
1058                         {
1059                             ScanTermInfoList::iterator it =
1060                                 entries_after.begin();
1061                             while (it != entries_after.end() && *it < my)
1062                                 it++;
1063                             if (my == *it)
1064                             {
1065                                 it->m_count += my.m_count;
1066                             }
1067                             else
1068                             {
1069                                 entries_after.insert(it, my);
1070                                 no_after++;
1071                             }
1072                         }
1073                     }
1074                 }
1075
1076             }
1077         }
1078         else
1079         {
1080             // if any target does not return scan response - return that
1081             package.response() = p->response();
1082             return;
1083         }
1084     }
1085
1086     if (false)
1087     {
1088         std::cout << "BEFORE\n";
1089         ScanTermInfoList::iterator it = entries_before.begin();
1090         for(; it != entries_before.end(); it++)
1091         {
1092             std::cout << " " << it->m_norm_term << " " << it->m_count << "\n";
1093         }
1094
1095         std::cout << "AFTER\n";
1096         it = entries_after.begin();
1097         for(; it != entries_after.end(); it++)
1098         {
1099             std::cout << " " << it->m_norm_term << " " << it->m_count << "\n";
1100         }
1101     }
1102
1103     if (false)
1104     {
1105         mp::odr odr;
1106         Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 1, "not implemented");
1107         package.response() = f_apdu;
1108     }
1109     else
1110     {
1111         mp::odr odr;
1112         Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 0, 0);
1113         Z_ScanResponse *resp = f_apdu->u.scanResponse;
1114
1115         int number_returned = *req->numberOfTermsRequested;
1116         int position_returned = *req->preferredPositionInResponse;
1117
1118         resp->entries->num_entries = number_returned;
1119         resp->entries->entries = (Z_Entry**)
1120             odr_malloc(odr, sizeof(Z_Entry*) * number_returned);
1121         int i;
1122
1123         int lbefore = entries_before.size();
1124         if (lbefore < position_returned-1)
1125             position_returned = lbefore+1;
1126
1127         ScanTermInfoList::iterator it = entries_before.begin();
1128         for (i = 0; i<position_returned-1 && it != entries_before.end(); i++, it++)
1129         {
1130             resp->entries->entries[position_returned-2-i] = it->get_entry(odr);
1131         }
1132
1133         it = entries_after.begin();
1134
1135         if (position_returned <= 0)
1136             i = 0;
1137         else
1138             i = position_returned-1;
1139         for (; i<number_returned && it != entries_after.end(); i++, it++)
1140         {
1141             resp->entries->entries[i] = it->get_entry(odr);
1142         }
1143
1144         number_returned = i;
1145
1146         resp->positionOfTerm = odr_intdup(odr, position_returned);
1147         resp->numberOfEntriesReturned = odr_intdup(odr, number_returned);
1148         resp->entries->num_entries = number_returned;
1149
1150         package.response() = f_apdu;
1151     }
1152 }
1153
1154
1155 void yf::Multi::process(mp::Package &package) const
1156 {
1157     FrontendPtr f = m_p->get_frontend(package);
1158
1159     Z_GDU *gdu = package.request().get();
1160
1161     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1162         Z_APDU_initRequest && !f->m_is_multi)
1163     {
1164         f->init(package, gdu);
1165     }
1166     else if (!f->m_is_multi)
1167         package.move();
1168     else if (gdu && gdu->which == Z_GDU_Z3950)
1169     {
1170         Z_APDU *apdu = gdu->u.z3950;
1171         if (apdu->which == Z_APDU_initRequest)
1172         {
1173             mp::odr odr;
1174
1175             package.response() = odr.create_close(
1176                 apdu,
1177                 Z_Close_protocolError,
1178                 "double init");
1179
1180             package.session().close();
1181         }
1182         else if (apdu->which == Z_APDU_searchRequest)
1183         {
1184             f->search(package, apdu);
1185         }
1186         else if (apdu->which == Z_APDU_presentRequest)
1187         {
1188             f->present(package, apdu);
1189         }
1190         else if (apdu->which == Z_APDU_scanRequest)
1191         {
1192             f->scan2(package, apdu);
1193         }
1194         else
1195         {
1196             mp::odr odr;
1197
1198             package.response() = odr.create_close(
1199                 apdu, Z_Close_protocolError,
1200                 "unsupported APDU in filter multi");
1201
1202             package.session().close();
1203         }
1204     }
1205     m_p->release_frontend(package);
1206 }
1207
1208 void mp::filter::Multi::configure(const xmlNode * ptr, bool test_only)
1209 {
1210     for (ptr = ptr->children; ptr; ptr = ptr->next)
1211     {
1212         if (ptr->type != XML_ELEMENT_NODE)
1213             continue;
1214         if (!strcmp((const char *) ptr->name, "target"))
1215         {
1216             std::string route = mp::xml::get_route(ptr);
1217             std::string target = mp::xml::get_text(ptr);
1218             m_p->m_route_patterns.push_back(Multi::Map(target, route));
1219         }
1220         else if (!strcmp((const char *) ptr->name, "hideunavailable"))
1221         {
1222             m_p->m_hide_unavailable = true;
1223         }
1224         else if (!strcmp((const char *) ptr->name, "mergetype"))
1225         {
1226             std::string mergetype = mp::xml::get_text(ptr);
1227             if (mergetype == "roundrobin")
1228                 m_p->m_merge_type = round_robin;
1229             else if (mergetype == "serveorder")
1230                 m_p->m_merge_type = serve_order;
1231             else
1232                 throw mp::filter::FilterException
1233                     ("Bad mergetype "  + mergetype + " in multi filter");
1234
1235         }
1236         else
1237         {
1238             throw mp::filter::FilterException
1239                 ("Bad element "
1240                  + std::string((const char *) ptr->name)
1241                  + " in multi filter");
1242         }
1243     }
1244 }
1245
1246 static mp::filter::Base* filter_creator()
1247 {
1248     return new mp::filter::Multi;
1249 }
1250
1251 extern "C" {
1252     struct metaproxy_1_filter_struct metaproxy_1_filter_multi = {
1253         0,
1254         "multi",
1255         filter_creator
1256     };
1257 }
1258
1259
1260 /*
1261  * Local variables:
1262  * c-basic-offset: 4
1263  * c-file-style: "Stroustrup"
1264  * indent-tabs-mode: nil
1265  * End:
1266  * vim: shiftwidth=4 tabstop=8 expandtab
1267  */
1268