src/filter_z3950_client.cpp

   1 /* This file is part of Metaproxy.
   2    Copyright (C) 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 "config.hpp"
  20
  21 #include "filter_z3950_client.hpp"
  22 #include <metaproxy/package.hpp>
  23 #include <metaproxy/util.hpp>
  24
  25 #include <map>
  26 #include <stdexcept>
  27 #include <list>
  28 #include <iostream>
  29
  30 #include <boost/thread/mutex.hpp>
  31 #include <boost/thread/condition.hpp>
  32 #include <boost/thread/xtime.hpp>
  33
  34 #include <yaz/zgdu.h>
  35 #include <yaz/log.h>
  36 #include <yaz/otherinfo.h>
  37 #include <yaz/diagbib1.h>
  38 #include <yaz/oid_db.h>
  39
  40 #include <yazpp/socket-manager.h>
  41 #include <yazpp/pdu-assoc.h>
  42 #include <yazpp/z-assoc.h>
  43
  44 namespace mp = metaproxy_1;
  45 namespace yf = mp::filter;
  46
  47 namespace metaproxy_1 {
  48     namespace filter {
  49         class Z3950Client::Assoc : public yazpp_1::Z_Assoc{
  50             friend class Rep;
  51             Assoc(yazpp_1::SocketManager *socket_manager,
  52                   yazpp_1::IPDU_Observable *PDU_Observable,
  53                   std::string host, int timeout);
  54             ~Assoc();
  55             void connectNotify();
  56             void failNotify();
  57             void timeoutNotify();
  58             void recv_GDU(Z_GDU *gdu, int len);
  59             void fixup_nsd(ODR odr, Z_Records *records);
  60             void fixup_nsd(ODR odr, Z_DefaultDiagFormat *nsd);
  61             void fixup_init(ODR odr, Z_InitResponse *initrs);
  62             yazpp_1::IPDU_Observer* sessionNotify(
  63                 yazpp_1::IPDU_Observable *the_PDU_Observable,
  64                 int fd);
  65
  66             yazpp_1::SocketManager *m_socket_manager;
  67             yazpp_1::IPDU_Observable *m_PDU_Observable;
  68             Package *m_package;
  69             bool m_in_use;
  70             bool m_waiting;
  71             bool m_destroyed;
  72             bool m_connected;
  73             bool m_has_closed;
  74             int m_queue_len;
  75             int m_time_elapsed;
  76             int m_time_max;
  77             int m_time_connect_max;
  78             std::string m_host;
  79         };
  80
  81         class Z3950Client::Rep {
  82         public:
  83             // number of seconds to wait before we give up request
  84             int m_timeout_sec;
  85             int m_max_sockets;
  86             bool m_force_close;
  87             bool m_client_ip;
  88             std::string m_default_target;
  89             std::string m_force_target;
  90             boost::mutex m_mutex;
  91             boost::condition m_cond_session_ready;
  92             std::map<mp::Session,Z3950Client::Assoc *> m_clients;
  93             Z3950Client::Assoc *get_assoc(Package &package);
  94             void send_and_receive(Package &package,
  95                                   yf::Z3950Client::Assoc *c);
  96             void release_assoc(Package &package);
  97         };
  98     }
  99 }
 100
 101 using namespace mp;
 102
 103 yf::Z3950Client::Assoc::Assoc(yazpp_1::SocketManager *socket_manager,
 104                               yazpp_1::IPDU_Observable *PDU_Observable,
 105                               std::string host, int timeout_sec)
 106     :  Z_Assoc(PDU_Observable),
 107        m_socket_manager(socket_manager), m_PDU_Observable(PDU_Observable),
 108        m_package(0), m_in_use(true), m_waiting(false),
 109        m_destroyed(false), m_connected(false), m_has_closed(false),
 110        m_queue_len(1),
 111        m_time_elapsed(0), m_time_max(timeout_sec),  m_time_connect_max(10),
 112        m_host(host)
 113 {
 114     // std::cout << "create assoc " << this << "\n";
 115 }
 116
 117 yf::Z3950Client::Assoc::~Assoc()
 118 {
 119     // std::cout << "destroy assoc " << this << "\n";
 120 }
 121
 122 void yf::Z3950Client::Assoc::connectNotify()
 123 {
 124     m_waiting = false;
 125
 126     m_connected = true;
 127 }
 128
 129 void yf::Z3950Client::Assoc::failNotify()
 130 {
 131     m_waiting = false;
 132
 133     mp::odr odr;
 134
 135     if (m_package)
 136     {
 137         Z_GDU *gdu = m_package->request().get();
 138         Z_APDU *apdu = 0;
 139         if (gdu && gdu->which == Z_GDU_Z3950)
 140             apdu = gdu->u.z3950;
 141
 142         m_package->response() = odr.create_close(apdu, Z_Close_peerAbort, 0);
 143         m_package->session().close();
 144     }
 145 }
 146
 147 void yf::Z3950Client::Assoc::timeoutNotify()
 148 {
 149     m_time_elapsed++;
 150     if ((m_connected && m_time_elapsed >= m_time_max)
 151         || (!m_connected && m_time_elapsed >= m_time_connect_max))
 152     {
 153         m_waiting = false;
 154
 155         mp::odr odr;
 156
 157         if (m_package)
 158         {
 159             Z_GDU *gdu = m_package->request().get();
 160             Z_APDU *apdu = 0;
 161             if (gdu && gdu->which == Z_GDU_Z3950)
 162                 apdu = gdu->u.z3950;
 163
 164             if (m_connected)
 165                 m_package->response() =
 166                     odr.create_close(apdu, Z_Close_lackOfActivity, 0);
 167             else
 168                 m_package->response() =
 169                     odr.create_close(apdu, Z_Close_peerAbort, 0);
 170
 171             m_package->session().close();
 172         }
 173     }
 174 }
 175
 176 void yf::Z3950Client::Assoc::fixup_nsd(ODR odr, Z_DefaultDiagFormat *nsd)
 177 {
 178     std::string addinfo;
 179
 180     // should really check for nsd->which.. But union has two members
 181     // containing almost same data
 182     const char *v2Addinfo = nsd->u.v2Addinfo;
 183     //  Z_InternationalString *v3Addinfo;
 184     if (v2Addinfo && *v2Addinfo)
 185     {
 186         addinfo.assign(nsd->u.v2Addinfo);
 187         addinfo += " ";
 188     }
 189     addinfo += "(backend=" + m_host + ")";
 190     nsd->u.v2Addinfo = odr_strdup(odr, addinfo.c_str());
 191 }
 192
 193 void yf::Z3950Client::Assoc::fixup_nsd(ODR odr, Z_Records *records)
 194 {
 195     if (records && records->which == Z_Records_NSD)
 196     {
 197         fixup_nsd(odr, records->u.nonSurrogateDiagnostic);
 198     }
 199     if (records && records->which == Z_Records_multipleNSD)
 200     {
 201         Z_DiagRecs *drecs = records->u.multipleNonSurDiagnostics;
 202         int i;
 203         for (i = 0; i < drecs->num_diagRecs; i++)
 204         {
 205             Z_DiagRec *dr = drecs->diagRecs[i];
 206
 207             if (dr->which == Z_DiagRec_defaultFormat)
 208                 fixup_nsd(odr, dr->u.defaultFormat);
 209         }
 210     }
 211 }
 212
 213 void yf::Z3950Client::Assoc::fixup_init(ODR odr, Z_InitResponse *initrs)
 214 {
 215     Z_External *uif = initrs->userInformationField;
 216
 217     if (uif && uif->which == Z_External_userInfo1)
 218     {
 219         Z_OtherInformation *ui = uif->u.userInfo1;
 220         int i;
 221         for (i = 0; i < ui->num_elements; i++)
 222         {
 223             Z_OtherInformationUnit *unit = ui->list[i];
 224             if (unit->which == Z_OtherInfo_externallyDefinedInfo &&
 225                 unit->information.externallyDefinedInfo &&
 226                 unit->information.externallyDefinedInfo->which ==
 227                 Z_External_diag1)
 228             {
 229                 Z_DiagnosticFormat *diag =
 230                     unit->information.externallyDefinedInfo->u.diag1;
 231                 int j;
 232                 for (j = 0; j < diag->num; j++)
 233                 {
 234                     Z_DiagnosticFormat_s *ds = diag->elements[j];
 235                     if (ds->which == Z_DiagnosticFormat_s_defaultDiagRec)
 236                     {
 237                         Z_DefaultDiagFormat *r = ds->u.defaultDiagRec;
 238                         char *oaddinfo = r->u.v2Addinfo;
 239                         char *naddinfo = (char *) odr_malloc(
 240                             odr,
 241                             (oaddinfo ? strlen(oaddinfo) : 0) + 20 +
 242                             m_host.length());
 243                         *naddinfo = '\0';
 244                         if (oaddinfo && *oaddinfo)
 245                         {
 246                             strcat(naddinfo, oaddinfo);
 247                             strcat(naddinfo, " ");
 248                         }
 249                         strcat(naddinfo, "(backend=");
 250                         strcat(naddinfo, m_host.c_str());
 251                         strcat(naddinfo, ")");
 252
 253                         r->u.v2Addinfo = naddinfo;
 254                     }
 255                 }
 256             }
 257         }
 258     }
 259 }
 260
 261 void yf::Z3950Client::Assoc::recv_GDU(Z_GDU *gdu, int len)
 262 {
 263     m_waiting = false;
 264
 265     if (m_package)
 266     {
 267         mp::odr odr; // must be in scope for response() = assignment
 268         if (gdu && gdu->which == Z_GDU_Z3950)
 269         {
 270             Z_APDU *apdu = gdu->u.z3950;
 271             switch (apdu->which)
 272             {
 273             case Z_APDU_searchResponse:
 274                 fixup_nsd(odr, apdu->u.searchResponse->records);
 275                 break;
 276             case Z_APDU_presentResponse:
 277                 fixup_nsd(odr, apdu->u.presentResponse->records);
 278                 break;
 279             case Z_APDU_initResponse:
 280                 fixup_init(odr, apdu->u.initResponse);
 281                 break;
 282             }
 283         }
 284         m_package->response() = gdu;
 285     }
 286 }
 287
 288 yazpp_1::IPDU_Observer *yf::Z3950Client::Assoc::sessionNotify(
 289     yazpp_1::IPDU_Observable *the_PDU_Observable,
 290     int fd)
 291 {
 292     return 0;
 293 }
 294
 295
 296 yf::Z3950Client::Z3950Client() :  m_p(new yf::Z3950Client::Rep)
 297 {
 298     m_p->m_timeout_sec = 30;
 299     m_p->m_max_sockets = 0;
 300     m_p->m_force_close = false;
 301     m_p->m_client_ip = false;
 302 }
 303
 304 yf::Z3950Client::~Z3950Client() {
 305 }
 306
 307 yf::Z3950Client::Assoc *yf::Z3950Client::Rep::get_assoc(Package &package)
 308 {
 309     // only one thread messes with the clients list at a time
 310     boost::mutex::scoped_lock lock(m_mutex);
 311
 312     std::map<mp::Session,yf::Z3950Client::Assoc *>::iterator it;
 313
 314     Z_GDU *gdu = package.request().get();
 315
 316     int max_sockets = package.origin().get_max_sockets();
 317     if (max_sockets == 0)
 318         max_sockets = m_max_sockets;
 319
 320     it = m_clients.find(package.session());
 321     if (it != m_clients.end())
 322     {
 323         it->second->m_queue_len++;
 324         while (true)
 325         {
 326 #if 0
 327             // double init .. NOT working yet
 328             if (gdu && gdu->which == Z_GDU_Z3950 &&
 329                 gdu->u.z3950->which == Z_APDU_initRequest)
 330             {
 331                 yazpp_1::SocketManager *s = it->second->m_socket_manager;
 332                 delete it->second;  // destroy Z_Assoc
 333                 delete s;    // then manager
 334                 m_clients.erase(it);
 335                 break;
 336             }
 337 #endif
 338             if (!it->second->m_in_use)
 339             {
 340                 it->second->m_in_use = true;
 341                 return it->second;
 342             }
 343             m_cond_session_ready.wait(lock);
 344         }
 345     }
 346     if (!gdu || gdu->which != Z_GDU_Z3950)
 347     {
 348         package.move();
 349         return 0;
 350     }
 351     // new Z39.50 session ..
 352     Z_APDU *apdu = gdu->u.z3950;
 353     // check that it is init. If not, close
 354     if (apdu->which != Z_APDU_initRequest)
 355     {
 356         mp::odr odr;
 357
 358         package.response() = odr.create_close(apdu,
 359                                               Z_Close_protocolError,
 360                                               "First PDU was not an "
 361                                               "Initialize Request");
 362         package.session().close();
 363         return 0;
 364     }
 365     std::string target = m_force_target;
 366     if (!target.length())
 367     {
 368         target = m_default_target;
 369         std::list<std::string> vhosts;
 370         mp::util::remove_vhost_otherinfo(&apdu->u.initRequest->otherInfo,
 371                                              vhosts);
 372         size_t no_vhosts = vhosts.size();
 373         if (no_vhosts == 1)
 374         {
 375             std::list<std::string>::const_iterator v_it = vhosts.begin();
 376             target = *v_it;
 377         }
 378         else if (no_vhosts == 0)
 379         {
 380             if (!target.length())
 381             {
 382                 // no default target. So we don't know where to connect
 383                 mp::odr odr;
 384                 package.response() = odr.create_initResponse(
 385                     apdu,
 386                     YAZ_BIB1_INIT_NEGOTIATION_OPTION_REQUIRED,
 387                     "z3950_client: No vhost given");
 388
 389                 package.session().close();
 390                 return 0;
 391             }
 392         }
 393         else if (no_vhosts > 1)
 394         {
 395             mp::odr odr;
 396             package.response() = odr.create_initResponse(
 397                 apdu,
 398                 YAZ_BIB1_COMBI_OF_SPECIFIED_DATABASES_UNSUPP,
 399                 "z3950_client: Can not cope with multiple vhosts");
 400             package.session().close();
 401             return 0;
 402         }
 403     }
 404
 405     // see if we have reached max number of clients (max-sockets)
 406
 407     while (max_sockets)
 408     {
 409         int no_not_in_use = 0;
 410         int number = 0;
 411         it = m_clients.begin();
 412         for (; it != m_clients.end(); it++)
 413         {
 414             yf::Z3950Client::Assoc *as = it->second;
 415             if (!strcmp(as->m_host.c_str(), target.c_str()))
 416             {
 417                 number++;
 418                 if (!as->m_in_use)
 419                     no_not_in_use++;
 420             }
 421         }
 422         yaz_log(YLOG_LOG, "Found %d/%d connections for %s", number, max_sockets,
 423                 target.c_str());
 424         if (number < max_sockets)
 425             break;
 426         if (no_not_in_use == 0) // all in use..
 427         {
 428             mp::odr odr;
 429
 430             package.response() = odr.create_initResponse(
 431                 apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
 432                 "z3950_client: max sessions");
 433             package.session().close();
 434             return 0;
 435         }
 436         boost::xtime xt;
 437         xtime_get(&xt,
 438 #if BOOST_VERSION >= 105000
 439                 boost::TIME_UTC_
 440 #else
 441                 boost::TIME_UTC
 442 #endif
 443                 );
 444
 445         xt.sec += 15;
 446         if (!m_cond_session_ready.timed_wait(lock, xt))
 447         {
 448             mp::odr odr;
 449
 450             package.response() = odr.create_initResponse(
 451                 apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
 452                 "z3950_client: max sessions");
 453             package.session().close();
 454             return 0;
 455         }
 456     }
 457
 458     yazpp_1::SocketManager *sm = new yazpp_1::SocketManager;
 459     yazpp_1::PDU_Assoc *pdu_as = new yazpp_1::PDU_Assoc(sm);
 460     yf::Z3950Client::Assoc *as = new yf::Z3950Client::Assoc(sm, pdu_as,
 461                                                             target.c_str(),
 462                                                             m_timeout_sec);
 463     m_clients[package.session()] = as;
 464     return as;
 465 }
 466
 467 void yf::Z3950Client::Rep::send_and_receive(Package &package,
 468                                             yf::Z3950Client::Assoc *c)
 469 {
 470     if (c->m_destroyed)
 471         return;
 472
 473     c->m_package = &package;
 474
 475     if (package.session().is_closed() && c->m_connected && !c->m_has_closed
 476         && m_force_close)
 477     {
 478         mp::odr odr;
 479
 480         package.request() = odr.create_close(
 481             0, Z_Close_finished, "z3950_client");
 482         c->m_package = 0; // don't inspect response
 483     }
 484     Z_GDU *gdu = package.request().get();
 485
 486     if (!gdu || gdu->which != Z_GDU_Z3950)
 487         return;
 488
 489     if (gdu->u.z3950->which == Z_APDU_close)
 490         c->m_has_closed = true;
 491
 492     Z_APDU *apdu = gdu->u.z3950;
 493
 494     // prepare connect
 495     c->m_time_elapsed = 0;
 496     c->m_waiting = true;
 497     if (!c->m_connected)
 498     {
 499         if (c->client(c->m_host.c_str()))
 500         {
 501             mp::odr odr;
 502             package.response() =
 503                 odr.create_close(gdu->u.z3950, Z_Close_peerAbort, 0);
 504             package.session().close();
 505             return;
 506         }
 507         c->timeout(1);  // so timeoutNotify gets called once per second
 508
 509
 510         while (!c->m_destroyed && c->m_waiting
 511                && c->m_socket_manager->processEvent() > 0)
 512             ;
 513     }
 514     if (!c->m_connected)
 515     {
 516         return;
 517     }
 518     mp::odr odr;
 519     if (m_client_ip)
 520     {
 521         std::string peer_name2 = package.origin().get_address();
 522         if (apdu->which == Z_APDU_initRequest && peer_name2.length())
 523         {
 524             Z_OtherInformation **oi = &apdu->u.initRequest->otherInfo;
 525             char *peer_name1 =
 526                 yaz_oi_get_string_oid(oi, yaz_oid_userinfo_client_ip, 1, 1);
 527             std::string pcomb;
 528             if (peer_name1)
 529             {
 530                 pcomb.append(peer_name1);
 531                 pcomb.append(", ");
 532             }
 533             pcomb.append(peer_name2);
 534             yaz_oi_set_string_oid(&apdu->u.initRequest->otherInfo,
 535                                   odr, yaz_oid_userinfo_client_ip,
 536                                   1, pcomb.c_str());
 537         }
 538     }
 539     // prepare response
 540     c->m_time_elapsed = 0;
 541     c->m_waiting = true;
 542
 543     // relay the package  ..
 544     int len;
 545     c->send_GDU(gdu, &len);
 546
 547     switch (gdu->u.z3950->which)
 548     {
 549     case Z_APDU_triggerResourceControlRequest:
 550         // request only..
 551         break;
 552     default:
 553         // for the rest: wait for a response PDU
 554         while (!c->m_destroyed && c->m_waiting
 555                && c->m_socket_manager->processEvent() > 0)
 556             ;
 557         break;
 558     }
 559 }
 560
 561 void yf::Z3950Client::Rep::release_assoc(Package &package)
 562 {
 563     boost::mutex::scoped_lock lock(m_mutex);
 564     std::map<mp::Session,yf::Z3950Client::Assoc *>::iterator it;
 565
 566     it = m_clients.find(package.session());
 567     if (it != m_clients.end())
 568     {
 569         it->second->m_in_use = false;
 570         it->second->m_queue_len--;
 571
 572         if (package.session().is_closed())
 573         {
 574             // destroy hint (send_and_receive)
 575             it->second->m_destroyed = true;
 576             if (it->second->m_queue_len == 0)
 577             {
 578                 yazpp_1::SocketManager *s = it->second->m_socket_manager;
 579                 delete it->second;  // destroy Z_Assoc
 580                 delete s;    // then manager
 581                 m_clients.erase(it);
 582             }
 583         }
 584         m_cond_session_ready.notify_all();
 585     }
 586 }
 587
 588 void yf::Z3950Client::process(Package &package) const
 589 {
 590     yf::Z3950Client::Assoc *c = m_p->get_assoc(package);
 591     if (c)
 592     {
 593         m_p->send_and_receive(package, c);
 594         m_p->release_assoc(package);
 595     }
 596 }
 597
 598 void yf::Z3950Client::configure(const xmlNode *ptr, bool test_only,
 599                                 const char *path)
 600 {
 601     for (ptr = ptr->children; ptr; ptr = ptr->next)
 602     {
 603         if (ptr->type != XML_ELEMENT_NODE)
 604             continue;
 605         if (!strcmp((const char *) ptr->name, "timeout"))
 606         {
 607             m_p->m_timeout_sec = mp::xml::get_int(ptr, 30);
 608         }
 609         else if (!strcmp((const char *) ptr->name, "default_target"))
 610         {
 611             m_p->m_default_target = mp::xml::get_text(ptr);
 612         }
 613         else if (!strcmp((const char *) ptr->name, "force_target"))
 614         {
 615             m_p->m_force_target = mp::xml::get_text(ptr);
 616         }
 617         else if (!strcmp((const char *) ptr->name, "max-sockets"))
 618         {
 619             m_p->m_max_sockets = mp::xml::get_int(ptr, 0);
 620         }
 621         else if (!strcmp((const char *) ptr->name, "force_close"))
 622         {
 623             m_p->m_force_close = mp::xml::get_bool(ptr, 0);
 624         }
 625         else if (!strcmp((const char *) ptr->name, "client_ip"))
 626         {
 627             m_p->m_client_ip = mp::xml::get_bool(ptr, 0);
 628         }
 629         else
 630         {
 631             throw mp::filter::FilterException("Bad element "
 632                                                + std::string((const char *)
 633                                                              ptr->name));
 634         }
 635     }
 636 }
 637
 638 static mp::filter::Base* filter_creator()
 639 {
 640     return new mp::filter::Z3950Client;
 641 }
 642
 643 extern "C" {
 644     struct metaproxy_1_filter_struct metaproxy_1_filter_z3950_client = {
 645         0,
 646         "z3950_client",
 647         filter_creator
 648     };
 649 }
 650
 651 /*
 652  * Local variables:
 653  * c-basic-offset: 4
 654  * c-file-style: "Stroustrup"
 655  * indent-tabs-mode: nil
 656  * End:
 657  * vim: shiftwidth=4 tabstop=8 expandtab
 658  */
 659