src/filter_multi.cpp

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