src/filter_virt_db.cpp

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