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