src/filter_multi.cpp

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