src/filter_multi.cpp

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