src/filter_virt_db.cpp

   1 /* $Id: filter_virt_db.cpp,v 1.29 2006-01-17 13:34:51 adam Exp $
   2    Copyright (c) 2005, Index Data.
   3
   4 %LICENSE%
   5  */
   6
   7 #include "config.hpp"
   8
   9 #include "filter.hpp"
  10 #include "package.hpp"
  11
  12 #include <boost/thread/mutex.hpp>
  13 #include <boost/thread/condition.hpp>
  14 #include <boost/shared_ptr.hpp>
  15
  16 #include "util.hpp"
  17 #include "filter_virt_db.hpp"
  18
  19 #include <yaz/zgdu.h>
  20 #include <yaz/otherinfo.h>
  21 #include <yaz/diagbib1.h>
  22
  23 #include <map>
  24 #include <iostream>
  25
  26 namespace yf = yp2::filter;
  27
  28 namespace yp2 {
  29     namespace filter {
  30
  31         struct Virt_db::Set {
  32             Set(BackendPtr b, std::string setname);
  33             Set();
  34             ~Set();
  35
  36             BackendPtr m_backend;
  37             std::string m_setname;
  38         };
  39         struct Virt_db::Map {
  40             Map(std::list<std::string> targets, std::string route);
  41             Map();
  42             std::list<std::string> m_targets;
  43             std::string m_route;
  44         };
  45         struct Virt_db::Backend {
  46             yp2::Session m_backend_session;
  47             std::list<std::string> m_frontend_databases;
  48             std::list<std::string> m_targets;
  49             std::string m_route;
  50             bool m_named_result_sets;
  51             int m_number_of_sets;
  52         };
  53         struct Virt_db::Frontend {
  54             Frontend(Rep *rep);
  55             ~Frontend();
  56             yp2::Session m_session;
  57             bool m_is_virtual;
  58             bool m_in_use;
  59             std::list<BackendPtr> m_backend_list;
  60             std::map<std::string,Virt_db::Set> m_sets;
  61
  62             void search(Package &package, Z_APDU *apdu);
  63             void present(Package &package, Z_APDU *apdu);
  64             void scan(Package &package, Z_APDU *apdu);
  65
  66             void close(Package &package);
  67             typedef std::map<std::string,Virt_db::Set>::iterator Sets_it;
  68
  69             BackendPtr lookup_backend_from_databases(
  70                 std::list<std::string> databases);
  71             BackendPtr create_backend_from_databases(
  72                 std::list<std::string> databases,
  73                 std::string &failing_database);
  74
  75             BackendPtr init_backend(std::list<std::string> database,
  76                                     Package &package,
  77                                     int &error_code, std::string &addinfo);
  78             Rep *m_p;
  79         };
  80         class Virt_db::Rep {
  81             friend class Virt_db;
  82             friend class Frontend;
  83
  84             FrontendPtr get_frontend(Package &package);
  85             void release_frontend(Package &package);
  86         private:
  87             boost::mutex m_sessions_mutex;
  88             std::map<std::string, Virt_db::Map>m_maps;
  89
  90             typedef std::map<std::string,Virt_db::Set>::iterator Sets_it;
  91
  92             boost::mutex m_mutex;
  93             boost::condition m_cond_session_ready;
  94             std::map<yp2::Session, FrontendPtr> m_clients;
  95         };
  96     }
  97 }
  98
  99 using namespace yp2;
 100
 101 yf::Virt_db::BackendPtr yf::Virt_db::Frontend::lookup_backend_from_databases(
 102     std::list<std::string> databases)
 103 {
 104     std::list<BackendPtr>::const_iterator map_it;
 105     map_it = m_backend_list.begin();
 106     for (; map_it != m_backend_list.end(); map_it++)
 107         if ((*map_it)->m_frontend_databases == databases)
 108             return *map_it;
 109     BackendPtr null;
 110     return null;
 111 }
 112
 113 yf::Virt_db::BackendPtr yf::Virt_db::Frontend::create_backend_from_databases(
 114     std::list<std::string> databases, std::string &failing_database)
 115 {
 116     BackendPtr b(new Backend);
 117     std::list<std::string>::const_iterator db_it = databases.begin();
 118
 119     b->m_number_of_sets = 0;
 120     b->m_frontend_databases = databases;
 121     b->m_named_result_sets = false;
 122
 123     std::map<std::string,bool> targets_dedup;
 124     for (; db_it != databases.end(); db_it++)
 125     {
 126         std::map<std::string, Virt_db::Map>::iterator map_it;
 127         map_it = m_p->m_maps.find(*db_it);
 128         if (map_it == m_p->m_maps.end())  // database not found
 129         {
 130             failing_database = *db_it;
 131             BackendPtr ptr;
 132             return ptr;
 133         }
 134         std::list<std::string>::const_iterator t_it =
 135             map_it->second.m_targets.begin();
 136         for (; t_it != map_it->second.m_targets.end(); t_it++)
 137             targets_dedup[*t_it] = true;
 138         b->m_route = map_it->second.m_route;
 139     }
 140     std::map<std::string,bool>::const_iterator tm_it = targets_dedup.begin();
 141     for (; tm_it != targets_dedup.end(); tm_it++)
 142         b->m_targets.push_back(tm_it->first);
 143
 144     return b;
 145 }
 146
 147 yf::Virt_db::BackendPtr yf::Virt_db::Frontend::init_backend(
 148     std::list<std::string> databases, Package &package,
 149     int &error_code, std::string &addinfo)
 150 {
 151     std::string failing_database;
 152     BackendPtr b = create_backend_from_databases(databases, failing_database);
 153     if (!b)
 154     {
 155         error_code = YAZ_BIB1_DATABASE_UNAVAILABLE;
 156         addinfo = failing_database;
 157         return b;
 158     }
 159     Package init_package(b->m_backend_session, package.origin());
 160     init_package.copy_filter(package);
 161
 162     yp2::odr odr;
 163
 164     Z_APDU *init_apdu = zget_APDU(odr, Z_APDU_initRequest);
 165
 166     std::list<std::string>::const_iterator t_it = b->m_targets.begin();
 167     int cat = 1;
 168     for (; t_it != b->m_targets.end(); t_it++, cat++)
 169     {
 170         yaz_oi_set_string_oidval(&init_apdu->u.initRequest->otherInfo, odr,
 171                                  VAL_PROXY, cat, t_it->c_str());
 172     }
 173
 174     Z_InitRequest *req = init_apdu->u.initRequest;
 175
 176     ODR_MASK_SET(req->options, Z_Options_search);
 177     ODR_MASK_SET(req->options, Z_Options_present);
 178     ODR_MASK_SET(req->options, Z_Options_namedResultSets);
 179     ODR_MASK_SET(req->options, Z_Options_scan);
 180
 181     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
 182     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
 183     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
 184
 185     init_package.request() = init_apdu;
 186
 187     init_package.move(b->m_route);  // sending init
 188
 189     if (init_package.session().is_closed())
 190     {
 191         error_code = YAZ_BIB1_DATABASE_UNAVAILABLE;
 192         // addinfo = database;
 193         BackendPtr null;
 194         return null;
 195     }
 196     Z_GDU *gdu = init_package.response().get();
 197     // we hope to get an init response
 198     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 199         Z_APDU_initResponse)
 200     {
 201         if (ODR_MASK_GET(gdu->u.z3950->u.initResponse->options,
 202                          Z_Options_namedResultSets))
 203         {
 204             b->m_named_result_sets = true;
 205         }
 206     }
 207     else
 208     {
 209         error_code = YAZ_BIB1_DATABASE_UNAVAILABLE;
 210         // addinfo = database;
 211         BackendPtr null;
 212         return null;
 213     }
 214     m_backend_list.push_back(b);
 215     return b;
 216 }
 217
 218 void yf::Virt_db::Frontend::search(Package &package, Z_APDU *apdu_req)
 219 {
 220     Z_SearchRequest *req = apdu_req->u.searchRequest;
 221     std::string vhost;
 222     std::string resultSetId = req->resultSetName;
 223     yp2::odr odr;
 224
 225     std::list<std::string> databases;
 226     int i;
 227     for (i = 0; i<req->num_databaseNames; i++)
 228         databases.push_back(req->databaseNames[i]);
 229
 230     BackendPtr b; // null for now
 231     Sets_it sets_it = m_sets.find(req->resultSetName);
 232     if (sets_it != m_sets.end())
 233     {
 234         // result set already exist
 235         // if replace indicator is off: we return diagnostic if
 236         // result set already exist.
 237         if (*req->replaceIndicator == 0)
 238         {
 239             Z_APDU *apdu =
 240                 odr.create_searchResponse(
 241                     apdu_req,
 242                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
 243                     0);
 244             package.response() = apdu;
 245
 246             return;
 247         }
 248         sets_it->second.m_backend->m_number_of_sets--;
 249
 250         // pick up any existing backend with a database match
 251         std::list<BackendPtr>::const_iterator map_it;
 252         map_it = m_backend_list.begin();
 253         for (; map_it != m_backend_list.end(); map_it++)
 254         {
 255             BackendPtr tmp = *map_it;
 256             if (tmp->m_frontend_databases == databases)
 257                 break;
 258         }
 259         if (map_it != m_backend_list.end())
 260             b = *map_it;
 261     }
 262     else
 263     {
 264         // new result set.
 265
 266         // pick up any existing database with named result sets ..
 267         // or one which has no result sets.. yet.
 268         std::list<BackendPtr>::const_iterator map_it;
 269         map_it = m_backend_list.begin();
 270         for (; map_it != m_backend_list.end(); map_it++)
 271         {
 272             BackendPtr tmp = *map_it;
 273             if (tmp->m_frontend_databases == databases &&
 274                 (tmp->m_named_result_sets ||
 275                  tmp->m_number_of_sets == 0))
 276                 break;
 277         }
 278         if (map_it != m_backend_list.end())
 279             b = *map_it;
 280     }
 281     if (!b)  // no backend yet. Must create a new one
 282     {
 283         int error_code;
 284         std::string addinfo;
 285         b = init_backend(databases, package, error_code, addinfo);
 286         if (!b)
 287         {
 288             // did not get a backend (unavailable somehow?)
 289
 290             Z_APDU *apdu =
 291                 odr.create_searchResponse(
 292                     apdu_req, error_code, addinfo.c_str());
 293             package.response() = apdu;
 294             return;
 295         }
 296     }
 297     m_sets.erase(req->resultSetName);
 298     // sending search to backend
 299     Package search_package(b->m_backend_session, package.origin());
 300
 301     search_package.copy_filter(package);
 302
 303     std::string backend_setname;
 304     if (b->m_named_result_sets)
 305     {
 306         backend_setname = std::string(req->resultSetName);
 307     }
 308     else
 309     {
 310         backend_setname = "default";
 311         req->resultSetName = odr_strdup(odr, backend_setname.c_str());
 312     }
 313
 314     // pick first targets spec and move the databases from it ..
 315     std::list<std::string>::const_iterator t_it = b->m_targets.begin();
 316     if (t_it != b->m_targets.end())
 317     {
 318         if (!yp2::util::set_databases_from_zurl(odr, *t_it,
 319                                                 &req->num_databaseNames,
 320                                                 &req->databaseNames));
 321     }
 322
 323     *req->replaceIndicator = 1;
 324
 325     search_package.request() = yazpp_1::GDU(apdu_req);
 326
 327     search_package.move(b->m_route);
 328
 329     if (search_package.session().is_closed())
 330     {
 331         package.response() = search_package.response();
 332         package.session().close();
 333         return;
 334     }
 335     package.response() = search_package.response();
 336
 337     b->m_number_of_sets++;
 338
 339     m_sets[resultSetId] = Virt_db::Set(b, backend_setname);
 340 }
 341
 342 yf::Virt_db::Frontend::Frontend(Rep *rep)
 343 {
 344     m_p = rep;
 345     m_is_virtual = false;
 346 }
 347
 348 void yf::Virt_db::Frontend::close(Package &package)
 349 {
 350     std::list<BackendPtr>::const_iterator b_it;
 351
 352     for (b_it = m_backend_list.begin(); b_it != m_backend_list.end(); b_it++)
 353     {
 354         (*b_it)->m_backend_session.close();
 355         Package close_package((*b_it)->m_backend_session, package.origin());
 356         close_package.copy_filter(package);
 357         close_package.move((*b_it)->m_route);
 358     }
 359     m_backend_list.clear();
 360 }
 361
 362 yf::Virt_db::Frontend::~Frontend()
 363 {
 364 }
 365
 366 yf::Virt_db::FrontendPtr yf::Virt_db::Rep::get_frontend(Package &package)
 367 {
 368     boost::mutex::scoped_lock lock(m_mutex);
 369
 370     std::map<yp2::Session,yf::Virt_db::FrontendPtr>::iterator it;
 371
 372     while(true)
 373     {
 374         it = m_clients.find(package.session());
 375         if (it == m_clients.end())
 376             break;
 377
 378         if (!it->second->m_in_use)
 379         {
 380             it->second->m_in_use = true;
 381             return it->second;
 382         }
 383         m_cond_session_ready.wait(lock);
 384     }
 385     FrontendPtr f(new Frontend(this));
 386     m_clients[package.session()] = f;
 387     f->m_in_use = true;
 388     return f;
 389 }
 390
 391 void yf::Virt_db::Rep::release_frontend(Package &package)
 392 {
 393     boost::mutex::scoped_lock lock(m_mutex);
 394     std::map<yp2::Session,yf::Virt_db::FrontendPtr>::iterator it;
 395
 396     it = m_clients.find(package.session());
 397     if (it != m_clients.end())
 398     {
 399         if (package.session().is_closed())
 400         {
 401             it->second->close(package);
 402             m_clients.erase(it);
 403         }
 404         else
 405         {
 406             it->second->m_in_use = false;
 407         }
 408         m_cond_session_ready.notify_all();
 409     }
 410 }
 411
 412 yf::Virt_db::Set::Set(BackendPtr b, std::string setname)
 413     :  m_backend(b), m_setname(setname)
 414 {
 415 }
 416
 417
 418 yf::Virt_db::Set::Set()
 419 {
 420 }
 421
 422
 423 yf::Virt_db::Set::~Set()
 424 {
 425 }
 426
 427 yf::Virt_db::Map::Map(std::list<std::string> targets, std::string route)
 428     : m_targets(targets), m_route(route)
 429 {
 430 }
 431
 432 yf::Virt_db::Map::Map()
 433 {
 434 }
 435
 436 yf::Virt_db::Virt_db() : m_p(new Virt_db::Rep)
 437 {
 438 }
 439
 440 yf::Virt_db::~Virt_db() {
 441 }
 442
 443 void yf::Virt_db::Frontend::present(Package &package, Z_APDU *apdu_req)
 444 {
 445     Z_PresentRequest *req = apdu_req->u.presentRequest;
 446     std::string resultSetId = req->resultSetId;
 447     yp2::odr odr;
 448
 449     Sets_it sets_it = m_sets.find(resultSetId);
 450     if (sets_it == m_sets.end())
 451     {
 452         Z_APDU *apdu =
 453             odr.create_presentResponse(
 454                 apdu_req,
 455                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
 456                 resultSetId.c_str());
 457         package.response() = apdu;
 458         return;
 459     }
 460     Session *id =
 461         new yp2::Session(sets_it->second.m_backend->m_backend_session);
 462
 463     // sending present to backend
 464     Package present_package(*id, package.origin());
 465     present_package.copy_filter(package);
 466
 467     req->resultSetId = odr_strdup(odr, sets_it->second.m_setname.c_str());
 468
 469     present_package.request() = yazpp_1::GDU(apdu_req);
 470
 471     present_package.move();
 472
 473     if (present_package.session().is_closed())
 474     {
 475         package.response() = present_package.response();
 476         package.session().close();
 477         return;
 478     }
 479     else
 480     {
 481         package.response() = present_package.response();
 482     }
 483     delete id;
 484 }
 485
 486 void yf::Virt_db::Frontend::scan(Package &package, Z_APDU *apdu_req)
 487 {
 488     Z_ScanRequest *req = apdu_req->u.scanRequest;
 489     std::string vhost;
 490     yp2::odr odr;
 491
 492     std::list<std::string> databases;
 493     int i;
 494     for (i = 0; i<req->num_databaseNames; i++)
 495         databases.push_back(req->databaseNames[i]);
 496
 497     BackendPtr b;
 498     // pick up any existing backend with a database match
 499     std::list<BackendPtr>::const_iterator map_it;
 500     map_it = m_backend_list.begin();
 501     for (; map_it != m_backend_list.end(); map_it++)
 502     {
 503         BackendPtr tmp = *map_it;
 504         if (tmp->m_frontend_databases == databases)
 505             break;
 506     }
 507     if (map_it != m_backend_list.end())
 508         b = *map_it;
 509     if (!b)  // no backend yet. Must create a new one
 510     {
 511         int error_code;
 512         std::string addinfo;
 513         b = init_backend(databases, package, error_code, addinfo);
 514         if (!b)
 515         {
 516             // did not get a backend (unavailable somehow?)
 517             Z_APDU *apdu =
 518                 odr.create_scanResponse(
 519                     apdu_req, error_code, addinfo.c_str());
 520             package.response() = apdu;
 521
 522             return;
 523         }
 524     }
 525     // sending scan to backend
 526     Package scan_package(b->m_backend_session, package.origin());
 527
 528     scan_package.copy_filter(package);
 529
 530     // pick first targets spec and move the databases from it ..
 531     std::list<std::string>::const_iterator t_it = b->m_targets.begin();
 532     if (t_it != b->m_targets.end())
 533     {
 534         if (!yp2::util::set_databases_from_zurl(odr, *t_it,
 535                                                 &req->num_databaseNames,
 536                                                 &req->databaseNames));
 537     }
 538     scan_package.request() = yazpp_1::GDU(apdu_req);
 539
 540     scan_package.move(b->m_route);
 541
 542     if (scan_package.session().is_closed())
 543     {
 544         package.response() = scan_package.response();
 545         package.session().close();
 546         return;
 547     }
 548     package.response() = scan_package.response();
 549 }
 550
 551
 552 void yf::Virt_db::add_map_db2targets(std::string db,
 553                                      std::list<std::string> targets,
 554                                      std::string route)
 555 {
 556     m_p->m_maps[db] = Virt_db::Map(targets, route);
 557 }
 558
 559
 560 void yf::Virt_db::add_map_db2target(std::string db,
 561                                     std::string target,
 562                                     std::string route)
 563 {
 564     std::list<std::string> targets;
 565     targets.push_back(target);
 566
 567     m_p->m_maps[db] = Virt_db::Map(targets, route);
 568 }
 569
 570 void yf::Virt_db::process(Package &package) const
 571 {
 572     FrontendPtr f = m_p->get_frontend(package);
 573
 574     Z_GDU *gdu = package.request().get();
 575
 576     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
 577         Z_APDU_initRequest && !f->m_is_virtual)
 578     {
 579         Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 580
 581         const char *vhost =
 582             yaz_oi_get_string_oidval(&req->otherInfo, VAL_PROXY, 1, 0);
 583         if (!vhost)
 584         {
 585             yp2::odr odr;
 586             Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
 587             Z_InitResponse *resp = apdu->u.initResponse;
 588
 589             int i;
 590             static const int masks[] = {
 591                 Z_Options_search,
 592                 Z_Options_present,
 593                 Z_Options_namedResultSets,
 594                 Z_Options_scan,
 595                 -1
 596             };
 597             for (i = 0; masks[i] != -1; i++)
 598                 if (ODR_MASK_GET(req->options, masks[i]))
 599                     ODR_MASK_SET(resp->options, masks[i]);
 600
 601             static const int versions[] = {
 602                 Z_ProtocolVersion_1,
 603                 Z_ProtocolVersion_2,
 604                 Z_ProtocolVersion_3,
 605                 -1
 606             };
 607             for (i = 0; versions[i] != -1; i++)
 608                 if (ODR_MASK_GET(req->protocolVersion, versions[i]))
 609                     ODR_MASK_SET(resp->protocolVersion, versions[i]);
 610                 else
 611                     break;
 612
 613             package.response() = apdu;
 614             f->m_is_virtual = true;
 615         }
 616         else
 617             package.move();
 618     }
 619     else if (!f->m_is_virtual)
 620         package.move();
 621     else if (gdu && gdu->which == Z_GDU_Z3950)
 622     {
 623         Z_APDU *apdu = gdu->u.z3950;
 624         if (apdu->which == Z_APDU_initRequest)
 625         {
 626             yp2::odr odr;
 627
 628             package.response() = odr.create_close(
 629                 apdu,
 630                 Z_Close_protocolError,
 631                 "double init");
 632
 633             package.session().close();
 634         }
 635         else if (apdu->which == Z_APDU_searchRequest)
 636         {
 637             f->search(package, apdu);
 638         }
 639         else if (apdu->which == Z_APDU_presentRequest)
 640         {
 641             f->present(package, apdu);
 642         }
 643         else if (apdu->which == Z_APDU_scanRequest)
 644         {
 645             f->scan(package, apdu);
 646         }
 647         else
 648         {
 649             yp2::odr odr;
 650
 651             package.response() = odr.create_close(
 652                 apdu, Z_Close_protocolError,
 653                 "unsupported APDU in filter_virt_db");
 654
 655             package.session().close();
 656         }
 657     }
 658     m_p->release_frontend(package);
 659 }
 660
 661
 662 void yp2::filter::Virt_db::configure(const xmlNode * ptr)
 663 {
 664     for (ptr = ptr->children; ptr; ptr = ptr->next)
 665     {
 666         if (ptr->type != XML_ELEMENT_NODE)
 667             continue;
 668         if (!strcmp((const char *) ptr->name, "virtual"))
 669         {
 670             std::string database;
 671             std::list<std::string> targets;
 672             xmlNode *v_node = ptr->children;
 673             for (; v_node; v_node = v_node->next)
 674             {
 675                 if (v_node->type != XML_ELEMENT_NODE)
 676                     continue;
 677
 678                 if (yp2::xml::is_element_yp2(v_node, "database"))
 679                     database = yp2::xml::get_text(v_node);
 680                 else if (yp2::xml::is_element_yp2(v_node, "target"))
 681                     targets.push_back(yp2::xml::get_text(v_node));
 682                 else
 683                     throw yp2::filter::FilterException
 684                         ("Bad element "
 685                          + std::string((const char *) v_node->name)
 686                          + " in virtual section"
 687                             );
 688             }
 689             std::string route = yp2::xml::get_route(ptr);
 690             add_map_db2targets(database, targets, route);
 691         }
 692         else
 693         {
 694             throw yp2::filter::FilterException
 695                 ("Bad element "
 696                  + std::string((const char *) ptr->name)
 697                  + " in virt_db filter");
 698         }
 699     }
 700 }
 701
 702 static yp2::filter::Base* filter_creator()
 703 {
 704     return new yp2::filter::Virt_db;
 705 }
 706
 707 extern "C" {
 708     struct yp2_filter_struct yp2_filter_virt_db = {
 709         0,
 710         "virt_db",
 711         filter_creator
 712     };
 713 }
 714
 715
 716 /*
 717  * Local variables:
 718  * c-basic-offset: 4
 719  * indent-tabs-mode: nil
 720  * c-file-style: "stroustrup"
 721  * End:
 722  * vim: shiftwidth=4 tabstop=8 expandtab
 723  */