src/filter_session_shared.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 <metaproxy/filter.hpp>
  22 #include <metaproxy/package.hpp>
  23
  24 #include <boost/thread/mutex.hpp>
  25 #include <boost/thread/condition.hpp>
  26 #include <boost/thread/thread.hpp>
  27 #include <boost/thread/xtime.hpp>
  28 #include <boost/shared_ptr.hpp>
  29 #include <boost/format.hpp>
  30
  31 #include <metaproxy/util.hpp>
  32 #include "filter_session_shared.hpp"
  33
  34 #include <yaz/copy_types.h>
  35 #include <yaz/log.h>
  36 #include <yaz/zgdu.h>
  37 #include <yaz/otherinfo.h>
  38 #include <yaz/diagbib1.h>
  39 #include <yazpp/z-query.h>
  40 #include <yazpp/record-cache.h>
  41 #include <map>
  42 #include <iostream>
  43 #include <time.h>
  44 #include <limits.h>
  45
  46 namespace mp = metaproxy_1;
  47 namespace yf = metaproxy_1::filter;
  48
  49 namespace metaproxy_1 {
  50
  51     namespace filter {
  52         // key for session.. We'll only share sessions with same InitKey
  53         class SessionShared::InitKey {
  54         public:
  55             bool operator < (const SessionShared::InitKey &k) const;
  56             InitKey(Z_InitRequest *req);
  57             InitKey(const InitKey &);
  58             ~InitKey();
  59         private:
  60             char *m_idAuthentication_buf;
  61             int m_idAuthentication_size;
  62             char *m_otherInfo_buf;
  63             int m_otherInfo_size;
  64             ODR m_odr;
  65         };
  66         // worker thread .. for expiry of sessions
  67         class SessionShared::Worker {
  68         public:
  69             Worker(SessionShared::Rep *rep);
  70             void operator() (void);
  71         private:
  72             SessionShared::Rep *m_p;
  73         };
  74         // backend result set
  75         class SessionShared::BackendSet {
  76         public:
  77             std::string m_result_set_id;
  78             Databases m_databases;
  79             Odr_int m_result_set_size;
  80             yazpp_1::Yaz_Z_Query m_query;
  81             time_t m_time_last_use;
  82             void timestamp();
  83             yazpp_1::RecordCache m_record_cache;
  84
  85             Z_OtherInformation *additionalSearchInfoRequest;
  86             Z_OtherInformation *additionalSearchInfoResponse;
  87             NMEM mem_additionalSearchInfo;
  88             BackendSet(
  89                 const std::string &result_set_id,
  90                 const Databases &databases,
  91                 const yazpp_1::Yaz_Z_Query &query,
  92                 Z_OtherInformation *additionalSearchInfoRequest);
  93             ~BackendSet();
  94             bool search(
  95                 Package &frontend_package,
  96                 Package &search_package,
  97                 const Z_APDU *apdu_req,
  98                 const BackendInstancePtr bp,
  99                 Z_Records **z_records);
 100         };
 101         // backend connection instance
 102         class SessionShared::BackendInstance {
 103             friend class Rep;
 104             friend class BackendClass;
 105             friend class BackendSet;
 106         public:
 107             mp::Session m_session;
 108             BackendSetList m_sets;
 109             bool m_in_use;
 110             int m_sequence_this;
 111             int m_result_set_sequence;
 112             time_t m_time_last_use;
 113             mp::Package * m_close_package;
 114             ~BackendInstance();
 115             void timestamp();
 116         };
 117         // backends of some class (all with same InitKey)
 118         class SessionShared::BackendClass : boost::noncopyable {
 119             friend class Rep;
 120             friend struct Frontend;
 121             bool m_named_result_sets;
 122             BackendInstanceList m_backend_list;
 123             BackendInstancePtr create_backend(const Package &package);
 124             void remove_backend(BackendInstancePtr b);
 125             BackendInstancePtr get_backend(const Package &package);
 126             void use_backend(BackendInstancePtr b);
 127             void release_backend(BackendInstancePtr b);
 128             bool expire_instances();
 129             yazpp_1::GDU m_init_request;
 130             yazpp_1::GDU m_init_response;
 131             boost::mutex m_mutex_backend_class;
 132             int m_sequence_top;
 133             time_t m_backend_set_ttl;
 134             time_t m_backend_expiry_ttl;
 135             size_t m_backend_set_max;
 136             Odr_int m_preferredMessageSize;
 137             Odr_int m_maximumRecordSize;
 138         public:
 139             BackendClass(const yazpp_1::GDU &init_request,
 140                          int resultset_ttl,
 141                          int resultset_max,
 142                          int session_ttl,
 143                          Odr_int preferredRecordSize,
 144                          Odr_int maximumRecordSize);
 145             ~BackendClass();
 146         };
 147         // frontend result set
 148         class SessionShared::FrontendSet {
 149             Databases m_databases;
 150             yazpp_1::Yaz_Z_Query m_query;
 151         public:
 152             const Databases &get_databases();
 153             const yazpp_1::Yaz_Z_Query &get_query();
 154             FrontendSet(
 155                 const Databases &databases,
 156                 const yazpp_1::Yaz_Z_Query &query);
 157             FrontendSet();
 158         };
 159         // frontend session
 160         struct SessionShared::Frontend {
 161             Frontend(Rep *rep);
 162             ~Frontend();
 163             bool m_is_virtual;
 164             bool m_in_use;
 165             Z_Options m_init_options;
 166             void search(Package &package, Z_APDU *apdu);
 167             void present(Package &package, Z_APDU *apdu);
 168             void scan(Package &package, Z_APDU *apdu);
 169
 170             int result_set_ref(ODR o,
 171                                const Databases &databases,
 172                                Z_RPNStructure *s, std::string &rset);
 173             void get_set(mp::Package &package,
 174                          const Z_APDU *apdu_req,
 175                          const Databases &databases,
 176                          yazpp_1::Yaz_Z_Query &query,
 177                          BackendInstancePtr &found_backend,
 178                          BackendSetPtr &found_set);
 179             void override_set(BackendInstancePtr &found_backend,
 180                               std::string &result_set_id,
 181                               const Databases &databases,
 182                               bool out_of_sessions);
 183
 184             Rep *m_p;
 185             BackendClassPtr m_backend_class;
 186             FrontendSets m_frontend_sets;
 187         };
 188         // representation
 189         class SessionShared::Rep {
 190             friend class SessionShared;
 191             friend struct Frontend;
 192
 193             FrontendPtr get_frontend(Package &package);
 194             void release_frontend(Package &package);
 195             Rep();
 196         public:
 197             ~Rep();
 198             void expire();
 199         private:
 200             void expire_classes();
 201             void stat();
 202             void init(Package &package, const Z_GDU *gdu,
 203                       FrontendPtr frontend);
 204             void start();
 205             boost::mutex m_mutex;
 206             boost::condition m_cond_session_ready;
 207             boost::condition m_cond_expire_ready;
 208             std::map<mp::Session, FrontendPtr> m_clients;
 209
 210             BackendClassMap m_backend_map;
 211             boost::mutex m_mutex_backend_map;
 212             boost::thread_group m_thrds;
 213             int m_resultset_ttl;
 214             int m_resultset_max;
 215             int m_session_ttl;
 216             bool m_optimize_search;
 217             bool m_restart;
 218             int m_session_max;
 219             Odr_int m_preferredMessageSize;
 220             Odr_int m_maximumRecordSize;
 221             bool close_down;
 222         };
 223     }
 224 }
 225
 226 yf::SessionShared::FrontendSet::FrontendSet(
 227     const Databases &databases,
 228     const yazpp_1::Yaz_Z_Query &query)
 229     : m_databases(databases), m_query(query)
 230 {
 231 }
 232
 233 const yf::SessionShared::Databases &
 234 yf::SessionShared::FrontendSet::get_databases()
 235 {
 236     return m_databases;
 237 }
 238
 239 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
 240 {
 241     return m_query;
 242 }
 243
 244 yf::SessionShared::InitKey::InitKey(const InitKey &k)
 245 {
 246     m_odr = odr_createmem(ODR_ENCODE);
 247
 248     m_idAuthentication_size =  k.m_idAuthentication_size;
 249     m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
 250     memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
 251            m_idAuthentication_size);
 252
 253     m_otherInfo_size =  k.m_otherInfo_size;
 254     m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
 255     memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
 256            m_otherInfo_size);
 257 }
 258
 259 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
 260 {
 261     m_odr = odr_createmem(ODR_ENCODE);
 262
 263     Z_IdAuthentication *t = req->idAuthentication;
 264     z_IdAuthentication(m_odr, &t, 1, 0);
 265     m_idAuthentication_buf =
 266         odr_getbuf(m_odr, &m_idAuthentication_size, 0);
 267
 268     Z_OtherInformation *o = req->otherInfo;
 269     z_OtherInformation(m_odr, &o, 1, 0);
 270     m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
 271 }
 272
 273 yf::SessionShared::InitKey::~InitKey()
 274 {
 275     odr_destroy(m_odr);
 276 }
 277
 278 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
 279     const
 280 {
 281     int c;
 282     c = mp::util::memcmp2(
 283         (void*) m_idAuthentication_buf, m_idAuthentication_size,
 284         (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
 285     if (c < 0)
 286         return true;
 287     else if (c > 0)
 288         return false;
 289
 290     c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
 291                           (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
 292     if (c < 0)
 293         return true;
 294     else if (c > 0)
 295         return false;
 296     return false;
 297 }
 298
 299 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
 300 {
 301     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 302     b->m_in_use = false;
 303 }
 304
 305
 306 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
 307 {
 308     {
 309         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 310         BackendInstanceList::iterator it = m_backend_list.begin();
 311         for (;;)
 312         {
 313             if (it == m_backend_list.end())
 314                 return;
 315             if (*it == b)
 316             {
 317                 it = m_backend_list.erase(it);
 318                 break;
 319             }
 320             it++;
 321         }
 322     }
 323     mp::odr odr;
 324     b->m_close_package->response() = odr.create_close(
 325         0, Z_Close_lackOfActivity, 0);
 326     b->m_close_package->session().close();
 327     b->m_close_package->move();
 328 }
 329
 330
 331 yf::SessionShared::BackendInstancePtr
 332 yf::SessionShared::BackendClass::get_backend(
 333     const mp::Package &frontend_package)
 334 {
 335     {
 336         boost::mutex::scoped_lock lock(m_mutex_backend_class);
 337
 338         BackendInstanceList::const_iterator it = m_backend_list.begin();
 339
 340         BackendInstancePtr backend1; // null
 341
 342         for (; it != m_backend_list.end(); it++)
 343         {
 344             if (!(*it)->m_in_use)
 345             {
 346                 if (!backend1
 347                     || (*it)->m_sequence_this < backend1->m_sequence_this)
 348                     backend1 = *it;
 349             }
 350         }
 351         if (backend1)
 352         {
 353             use_backend(backend1);
 354             return backend1;
 355         }
 356     }
 357     return create_backend(frontend_package);
 358 }
 359
 360 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
 361 {
 362     backend->m_in_use = true;
 363     backend->m_sequence_this = m_sequence_top++;
 364 }
 365
 366 void yf::SessionShared::BackendInstance::timestamp()
 367 {
 368     assert(m_in_use);
 369     time(&m_time_last_use);
 370 }
 371
 372 yf::SessionShared::BackendInstance::~BackendInstance()
 373 {
 374     delete m_close_package;
 375 }
 376
 377 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
 378     const mp::Package &frontend_package)
 379 {
 380     BackendInstancePtr bp(new BackendInstance);
 381     BackendInstancePtr null;
 382
 383     bp->m_close_package =
 384         new mp::Package(bp->m_session, frontend_package.origin());
 385     bp->m_close_package->copy_filter(frontend_package);
 386
 387     Package init_package(bp->m_session, frontend_package.origin());
 388
 389     init_package.copy_filter(frontend_package);
 390
 391     yazpp_1::GDU actual_init_request = m_init_request;
 392     Z_GDU *init_pdu = actual_init_request.get();
 393
 394     assert(init_pdu->which == Z_GDU_Z3950);
 395     assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
 396
 397     Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
 398     ODR_MASK_ZERO(req->options);
 399
 400     ODR_MASK_SET(req->options, Z_Options_search);
 401     ODR_MASK_SET(req->options, Z_Options_present);
 402     ODR_MASK_SET(req->options, Z_Options_namedResultSets);
 403     ODR_MASK_SET(req->options, Z_Options_scan);
 404
 405     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
 406     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
 407     ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
 408
 409     if (m_preferredMessageSize)
 410         *req->preferredMessageSize = m_preferredMessageSize;
 411     if (m_maximumRecordSize)
 412         *req->maximumRecordSize = m_maximumRecordSize;
 413
 414     init_package.request() = init_pdu;
 415
 416     init_package.move();
 417
 418     boost::mutex::scoped_lock lock(m_mutex_backend_class);
 419
 420     m_named_result_sets = false;
 421     Z_GDU *gdu = init_package.response().get();
 422
 423     if (gdu && gdu->which == Z_GDU_Z3950
 424         && gdu->u.z3950->which == Z_APDU_initResponse)
 425     {
 426         Z_InitResponse *res = gdu->u.z3950->u.initResponse;
 427         m_init_response = gdu->u.z3950;
 428         if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
 429         {
 430             m_named_result_sets = true;
 431         }
 432         if (*gdu->u.z3950->u.initResponse->result
 433             && !init_package.session().is_closed())
 434         {
 435             bp->m_in_use = true;
 436             time(&bp->m_time_last_use);
 437             bp->m_sequence_this = 0;
 438             bp->m_result_set_sequence = 0;
 439             m_backend_list.push_back(bp);
 440             return bp;
 441         }
 442     }
 443     else
 444     {
 445         yazpp_1::GDU empty_gdu;
 446         m_init_response = empty_gdu;
 447     }
 448
 449     if (!init_package.session().is_closed())
 450     {
 451         init_package.copy_filter(frontend_package);
 452         init_package.session().close();
 453         init_package.move();
 454     }
 455     return null;
 456 }
 457
 458
 459 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
 460                                               int resultset_ttl,
 461                                               int resultset_max,
 462                                               int session_ttl,
 463                                               Odr_int preferredMessageSize,
 464                                               Odr_int maximumRecordSize)
 465     : m_named_result_sets(false), m_init_request(init_request),
 466       m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
 467       m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max),
 468       m_preferredMessageSize(preferredMessageSize),
 469       m_maximumRecordSize(maximumRecordSize)
 470 {}
 471
 472 yf::SessionShared::BackendClass::~BackendClass()
 473 {}
 474
 475 void yf::SessionShared::Rep::stat()
 476 {
 477     int no_classes = 0;
 478     int no_instances = 0;
 479     BackendClassMap::const_iterator it;
 480     {
 481         boost::mutex::scoped_lock lock(m_mutex_backend_map);
 482         for (it = m_backend_map.begin(); it != m_backend_map.end(); it++)
 483         {
 484             BackendClassPtr bc = it->second;
 485             no_classes++;
 486             BackendInstanceList::iterator bit = bc->m_backend_list.begin();
 487             for (; bit != bc->m_backend_list.end(); bit++)
 488                 no_instances++;
 489         }
 490     }
 491 }
 492
 493 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
 494                                   FrontendPtr frontend)
 495 {
 496     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 497
 498     frontend->m_is_virtual = true;
 499     frontend->m_init_options = *req->options;
 500     InitKey k(req);
 501     {
 502         boost::mutex::scoped_lock lock(m_mutex_backend_map);
 503         BackendClassMap::const_iterator it;
 504         it = m_backend_map.find(k);
 505         if (it == m_backend_map.end())
 506         {
 507             BackendClassPtr b(new BackendClass(gdu->u.z3950,
 508                                                m_resultset_ttl,
 509                                                m_resultset_max,
 510                                                m_session_ttl,
 511                                                m_preferredMessageSize,
 512                                                m_maximumRecordSize));
 513             m_backend_map[k] = b;
 514             frontend->m_backend_class = b;
 515         }
 516         else
 517         {
 518             frontend->m_backend_class = it->second;
 519         }
 520     }
 521     BackendClassPtr bc = frontend->m_backend_class;
 522     mp::odr odr;
 523
 524     // we only need to get init response from "first" target in
 525     // backend class - the assumption being that init response is
 526     // same for all
 527     if (bc->m_backend_list.size() == 0)
 528     {
 529         BackendInstancePtr backend = bc->create_backend(package);
 530         if (backend)
 531             bc->release_backend(backend);
 532     }
 533
 534     yazpp_1::GDU init_response;
 535     {
 536         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 537
 538         init_response = bc->m_init_response;
 539     }
 540
 541     if (init_response.get())
 542     {
 543         Z_GDU *response_gdu = init_response.get();
 544         mp::util::transfer_referenceId(odr, gdu->u.z3950,
 545                                        response_gdu->u.z3950);
 546         Z_InitResponse *init_res = response_gdu->u.z3950->u.initResponse;
 547         Z_Options *server_options = init_res->options;
 548         Z_Options *client_options = &frontend->m_init_options;
 549         int i;
 550         for (i = 0; i < 30; i++)
 551             if (!ODR_MASK_GET(client_options, i))
 552                 ODR_MASK_CLEAR(server_options, i);
 553
 554         if (!m_preferredMessageSize ||
 555             *init_res->preferredMessageSize > *req->preferredMessageSize)
 556             *init_res->preferredMessageSize = *req->preferredMessageSize;
 557
 558         if (!m_maximumRecordSize ||
 559             *init_res->maximumRecordSize > *req->maximumRecordSize)
 560             *init_res->maximumRecordSize = *req->maximumRecordSize;
 561
 562         package.response() = init_response;
 563         if (!*response_gdu->u.z3950->u.initResponse->result)
 564             package.session().close();
 565     }
 566     else
 567     {
 568         Z_APDU *apdu =
 569             odr.create_initResponse(
 570                 gdu->u.z3950, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
 571                 "session_shared: target closed connection during init");
 572         *apdu->u.initResponse->result = 0;
 573         package.response() = apdu;
 574         package.session().close();
 575     }
 576 }
 577
 578 void yf::SessionShared::BackendSet::timestamp()
 579 {
 580     time(&m_time_last_use);
 581 }
 582
 583 yf::SessionShared::BackendSet::BackendSet(
 584     const std::string &result_set_id,
 585     const Databases &databases,
 586     const yazpp_1::Yaz_Z_Query &query,
 587     Z_OtherInformation *additionalSearchInfo) :
 588     m_result_set_id(result_set_id),
 589     m_databases(databases), m_result_set_size(0), m_query(query)
 590 {
 591     timestamp();
 592     mem_additionalSearchInfo = nmem_create();
 593     additionalSearchInfoResponse = 0;
 594     additionalSearchInfoRequest =
 595         yaz_clone_z_OtherInformation(additionalSearchInfo,
 596                                      mem_additionalSearchInfo);
 597 }
 598
 599 yf::SessionShared::BackendSet::~BackendSet()
 600 {
 601     nmem_destroy(mem_additionalSearchInfo);
 602 }
 603
 604 static int get_diagnostic(Z_DefaultDiagFormat *r)
 605 {
 606     return *r->condition;
 607 }
 608
 609 bool yf::SessionShared::BackendSet::search(
 610     mp::Package &frontend_package,
 611     mp::Package &search_package,
 612     const Z_APDU *frontend_apdu,
 613     const BackendInstancePtr bp,
 614     Z_Records **z_records)
 615 {
 616     mp::odr odr;
 617     Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
 618     Z_SearchRequest *req = apdu_req->u.searchRequest;
 619
 620     req->additionalSearchInfo = additionalSearchInfoRequest;
 621     req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
 622     req->query = m_query.get_Z_Query();
 623
 624     req->num_databaseNames = m_databases.size();
 625     req->databaseNames = (char**)
 626         odr_malloc(odr, req->num_databaseNames * sizeof(char *));
 627     Databases::const_iterator it = m_databases.begin();
 628     size_t i = 0;
 629     for (; it != m_databases.end(); it++)
 630         req->databaseNames[i++] = odr_strdup(odr, it->c_str());
 631
 632     if (frontend_apdu->which == Z_APDU_searchRequest)
 633         req->preferredRecordSyntax =
 634             frontend_apdu->u.searchRequest->preferredRecordSyntax;
 635
 636     search_package.request() = apdu_req;
 637
 638     search_package.move();
 639
 640     Z_GDU *gdu = search_package.response().get();
 641     if (!search_package.session().is_closed()
 642         && gdu && gdu->which == Z_GDU_Z3950
 643         && gdu->u.z3950->which == Z_APDU_searchResponse)
 644     {
 645         Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
 646         *z_records = b_resp->records;
 647         m_result_set_size = *b_resp->resultCount;
 648
 649         additionalSearchInfoResponse = yaz_clone_z_OtherInformation(
 650             b_resp->additionalSearchInfo, mem_additionalSearchInfo);
 651         return true;
 652     }
 653     Z_APDU *f_apdu = 0;
 654     const char *addinfo = "session_shared: "
 655         "target closed connection during search";
 656     if (frontend_apdu->which == Z_APDU_searchRequest)
 657         f_apdu = odr.create_searchResponse(
 658             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 659     else if (frontend_apdu->which == Z_APDU_presentRequest)
 660         f_apdu = odr.create_presentResponse(
 661             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 662     else
 663         f_apdu = odr.create_close(
 664             frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 665     frontend_package.response() = f_apdu;
 666     return false;
 667 }
 668
 669 void yf::SessionShared::Frontend::override_set(
 670     BackendInstancePtr &found_backend,
 671     std::string &result_set_id,
 672     const Databases &databases,
 673     bool out_of_sessions)
 674 {
 675     BackendClassPtr bc = m_backend_class;
 676     BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 677     time_t now;
 678     time(&now);
 679
 680     size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
 681     for (; it != bc->m_backend_list.end(); it++)
 682     {
 683         if (!(*it)->m_in_use)
 684         {
 685             BackendSetList::iterator set_it = (*it)->m_sets.begin();
 686             for (; set_it != (*it)->m_sets.end(); set_it++)
 687             {
 688                 if ((max_sets > 1 || (*set_it)->m_databases == databases)
 689                     &&
 690                     (out_of_sessions ||
 691                      now < (*set_it)->m_time_last_use ||
 692                      now - (*set_it)->m_time_last_use >= bc->m_backend_set_ttl))
 693                 {
 694                     found_backend = *it;
 695                     result_set_id = (*set_it)->m_result_set_id;
 696                     found_backend->m_sets.erase(set_it);
 697                     return;
 698                 }
 699             }
 700         }
 701     }
 702     for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
 703     {
 704         if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
 705         {
 706             found_backend = *it;
 707             if (bc->m_named_result_sets)
 708             {
 709                 result_set_id = boost::io::str(
 710                     boost::format("%1%") %
 711                     found_backend->m_result_set_sequence);
 712                 found_backend->m_result_set_sequence++;
 713             }
 714             else
 715                 result_set_id = "default";
 716             return;
 717         }
 718     }
 719 }
 720
 721 void yf::SessionShared::Frontend::get_set(mp::Package &package,
 722                                           const Z_APDU *apdu_req,
 723                                           const Databases &databases,
 724                                           yazpp_1::Yaz_Z_Query &query,
 725                                           BackendInstancePtr &found_backend,
 726                                           BackendSetPtr &found_set)
 727 {
 728     bool session_restarted = false;
 729     Z_OtherInformation *additionalSearchInfo = 0;
 730
 731     if (apdu_req->which == Z_APDU_searchRequest)
 732         additionalSearchInfo = apdu_req->u.searchRequest->additionalSearchInfo;
 733
 734 restart:
 735     std::string result_set_id;
 736     bool out_of_sessions = false;
 737     BackendClassPtr bc = m_backend_class;
 738     {
 739         boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
 740
 741         if ((int) bc->m_backend_list.size() >= m_p->m_session_max)
 742             out_of_sessions = true;
 743
 744         if (m_p->m_optimize_search)
 745         {
 746             // look at each backend and see if we have a similar search
 747             BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
 748             for (; it != bc->m_backend_list.end(); it++)
 749             {
 750                 if (!(*it)->m_in_use)
 751                 {
 752                     BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
 753                     for (; set_it != (*it)->m_sets.end(); set_it++)
 754                     {
 755                         // for real present request we don't care
 756                         // if additionalSearchInfo matches: same records
 757                         if ((*set_it)->m_databases == databases
 758                             && query.match(&(*set_it)->m_query)
 759                             && (apdu_req->which != Z_APDU_searchRequest ||
 760                                 yaz_compare_z_OtherInformation(
 761                                     additionalSearchInfo,
 762                                 (*set_it)->additionalSearchInfoRequest)))
 763                         {
 764                             found_set = *set_it;
 765                             found_backend = *it;
 766                             bc->use_backend(found_backend);
 767                             // found matching set. No need to search again
 768                             return;
 769                         }
 770                     }
 771                 }
 772             }
 773         }
 774         override_set(found_backend, result_set_id, databases, out_of_sessions);
 775         if (found_backend)
 776             bc->use_backend(found_backend);
 777     }
 778     if (!found_backend)
 779     {
 780         // create a new backend set (and new set) if we're not out of sessions
 781         if (!out_of_sessions)
 782             found_backend = bc->create_backend(package);
 783
 784         if (!found_backend)
 785         {
 786             Z_APDU *f_apdu = 0;
 787             mp::odr odr;
 788             const char *addinfo = 0;
 789
 790             if (out_of_sessions)
 791                 addinfo = "session_shared: all sessions in use";
 792             else
 793                 addinfo = "session_shared: could not create backend";
 794             if (apdu_req->which == Z_APDU_searchRequest)
 795             {
 796                 f_apdu = odr.create_searchResponse(
 797                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 798             }
 799             else if (apdu_req->which == Z_APDU_presentRequest)
 800             {
 801                 f_apdu = odr.create_presentResponse(
 802                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 803             }
 804             else
 805             {
 806                 f_apdu = odr.create_close(
 807                     apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, addinfo);
 808             }
 809             package.response() = f_apdu;
 810             return;
 811         }
 812         if (bc->m_named_result_sets)
 813         {
 814             result_set_id = boost::io::str(
 815                 boost::format("%1%") % found_backend->m_result_set_sequence);
 816             found_backend->m_result_set_sequence++;
 817         }
 818         else
 819             result_set_id = "default";
 820     }
 821     found_backend->timestamp();
 822
 823     // we must search ...
 824     BackendSetPtr new_set(new BackendSet(result_set_id,
 825                                          databases, query,
 826                                          additionalSearchInfo));
 827     Z_Records *z_records = 0;
 828
 829     Package search_package(found_backend->m_session, package.origin());
 830     search_package.copy_filter(package);
 831
 832     if (!new_set->search(package, search_package,
 833                          apdu_req, found_backend, &z_records))
 834     {
 835         bc->remove_backend(found_backend);
 836         return; // search error
 837     }
 838
 839     if (z_records)
 840     {
 841         int condition = 0;
 842         if (z_records->which == Z_Records_NSD)
 843         {
 844             condition =
 845                 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
 846         }
 847         else if (z_records->which == Z_Records_multipleNSD)
 848         {
 849             if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
 850                 &&
 851
 852                 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
 853                 Z_DiagRec_defaultFormat)
 854             {
 855                 condition = get_diagnostic(
 856                     z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
 857
 858             }
 859         }
 860         if (m_p->m_restart && !session_restarted &&
 861             condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
 862         {
 863             package.log("session_shared", YLOG_LOG, "restart");
 864             bc->remove_backend(found_backend);
 865             session_restarted = true;
 866             found_backend.reset();
 867             goto restart;
 868
 869         }
 870
 871         if (condition)
 872         {
 873             mp::odr odr;
 874             if (apdu_req->which == Z_APDU_searchRequest)
 875             {
 876                 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
 877                                                            0, 0);
 878                 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 879                 *f_resp->searchStatus = Z_SearchResponse_none;
 880                 f_resp->records = z_records;
 881                 package.response() = f_apdu;
 882             }
 883             if (apdu_req->which == Z_APDU_presentRequest)
 884             {
 885                 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
 886                                                             0, 0);
 887                 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
 888                 f_resp->records = z_records;
 889                 package.response() = f_apdu;
 890             }
 891             bc->release_backend(found_backend);
 892             return; // search error
 893         }
 894     }
 895     if (m_p->m_restart && !session_restarted && new_set->m_result_set_size < 0)
 896     {
 897         package.log("session_shared", YLOG_LOG, "restart");
 898         bc->remove_backend(found_backend);
 899         session_restarted = true;
 900         found_backend.reset();
 901         goto restart;
 902     }
 903
 904     found_set = new_set;
 905     found_set->timestamp();
 906     found_backend->m_sets.push_back(found_set);
 907 }
 908
 909 int yf::SessionShared::Frontend::result_set_ref(ODR o,
 910                                                 const Databases &databases,
 911                                                 Z_RPNStructure *s,
 912                                                 std::string &rset)
 913 {
 914     int ret = 0;
 915     switch (s->which)
 916     {
 917     case Z_RPNStructure_simple:
 918         if (s->u.simple->which == Z_Operand_resultSetId)
 919         {
 920             const char *id = s->u.simple->u.resultSetId;
 921             rset = id;
 922
 923             FrontendSets::iterator fset_it = m_frontend_sets.find(id);
 924             if (fset_it == m_frontend_sets.end())
 925             {
 926                 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
 927             }
 928             else if (fset_it->second->get_databases() != databases)
 929             {
 930                 ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
 931             }
 932             else
 933             {
 934                 yazpp_1::Yaz_Z_Query query = fset_it->second->get_query();
 935                 Z_Query *q = yaz_copy_Z_Query(query.get_Z_Query(), o);
 936                 if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
 937                 {
 938                     s->which = q->u.type_1->RPNStructure->which;
 939                     s->u.simple = q->u.type_1->RPNStructure->u.simple;
 940                 }
 941                 else
 942                 {
 943                     ret = YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST;
 944                 }
 945             }
 946         }
 947         break;
 948     case Z_RPNStructure_complex:
 949         ret = result_set_ref(o, databases, s->u.complex->s1, rset);
 950         if (!ret)
 951             ret = result_set_ref(o, databases, s->u.complex->s2, rset);
 952         break;
 953     }
 954     return ret;
 955 }
 956
 957 void yf::SessionShared::Frontend::search(mp::Package &package,
 958                                          Z_APDU *apdu_req)
 959 {
 960     Z_SearchRequest *req = apdu_req->u.searchRequest;
 961     FrontendSets::iterator fset_it =
 962         m_frontend_sets.find(req->resultSetName);
 963     if (fset_it != m_frontend_sets.end())
 964     {
 965         // result set already exist
 966         // if replace indicator is off: we return diagnostic if
 967         // result set already exist.
 968         if (*req->replaceIndicator == 0)
 969         {
 970             mp::odr odr;
 971             Z_APDU *apdu =
 972                 odr.create_searchResponse(
 973                     apdu_req,
 974                     YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
 975                     0);
 976             package.response() = apdu;
 977             return;
 978         }
 979         m_frontend_sets.erase(fset_it);
 980     }
 981
 982     Databases databases;
 983     int i;
 984     for (i = 0; i < req->num_databaseNames; i++)
 985         databases.push_back(req->databaseNames[i]);
 986
 987
 988     yazpp_1::Yaz_Z_Query query;
 989     query.set_Z_Query(req->query);
 990
 991     Z_Query *q = query.get_Z_Query();
 992     if (q->which == Z_Query_type_1 || q->which == Z_Query_type_101)
 993     {
 994         mp::odr odr;
 995         std::string rset;
 996         int diag = result_set_ref(odr, databases, q->u.type_1->RPNStructure,
 997                                   rset);
 998         if (diag)
 999         {
1000             Z_APDU *apdu =
1001                 odr.create_searchResponse(
1002                     apdu_req,
1003                     diag,
1004                     rset.c_str());
1005             package.response() = apdu;
1006             return;
1007         }
1008         query.set_Z_Query(q);
1009     }
1010
1011     BackendSetPtr found_set; // null
1012     BackendInstancePtr found_backend; // null
1013
1014     get_set(package, apdu_req, databases, query, found_backend, found_set);
1015     if (!found_set)
1016         return;
1017
1018     mp::odr odr;
1019     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
1020     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
1021     *f_resp->resultCount = found_set->m_result_set_size;
1022     f_resp->additionalSearchInfo = found_set->additionalSearchInfoResponse;
1023     package.response() = f_apdu;
1024
1025     FrontendSetPtr fset(new FrontendSet(databases, query));
1026     m_frontend_sets[req->resultSetName] = fset;
1027
1028     m_backend_class->release_backend(found_backend);
1029 }
1030
1031 void yf::SessionShared::Frontend::present(mp::Package &package,
1032                                           Z_APDU *apdu_req)
1033 {
1034     mp::odr odr;
1035     Z_PresentRequest *req = apdu_req->u.presentRequest;
1036
1037     FrontendSets::iterator fset_it =
1038         m_frontend_sets.find(req->resultSetId);
1039
1040     if (fset_it == m_frontend_sets.end())
1041     {
1042         Z_APDU *apdu =
1043             odr.create_presentResponse(
1044                 apdu_req,
1045                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
1046                 req->resultSetId);
1047         package.response() = apdu;
1048         return;
1049     }
1050     FrontendSetPtr fset = fset_it->second;
1051
1052     Databases databases = fset->get_databases();
1053     yazpp_1::Yaz_Z_Query query = fset->get_query();
1054
1055     BackendClassPtr bc = m_backend_class;
1056     BackendSetPtr found_set; // null
1057     BackendInstancePtr found_backend;
1058
1059     get_set(package, apdu_req, databases, query, found_backend, found_set);
1060     if (!found_set)
1061         return;
1062
1063     Z_NamePlusRecordList *npr_res = 0;
1064     // record_cache.lookup types are int's. Avoid non-fitting values
1065     if (*req->resultSetStartPoint > 0
1066         && *req->resultSetStartPoint < INT_MAX
1067         && *req->numberOfRecordsRequested > 0
1068         && *req->numberOfRecordsRequested < INT_MAX
1069         && found_set->m_record_cache.lookup(odr, &npr_res,
1070                                             *req->resultSetStartPoint,
1071                                             *req->numberOfRecordsRequested,
1072                                             req->preferredRecordSyntax,
1073                                             req->recordComposition))
1074     {
1075         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1076         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1077
1078         yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1079                 " records in cache %p",
1080                 *req->resultSetStartPoint,
1081                 *req->numberOfRecordsRequested,
1082                 &found_set->m_record_cache);
1083
1084         *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
1085         *f_resp->nextResultSetPosition =
1086             *req->resultSetStartPoint + *req->numberOfRecordsRequested;
1087         // f_resp->presentStatus assumed OK.
1088         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
1089         f_resp->records->which = Z_Records_DBOSD;
1090         f_resp->records->u.databaseOrSurDiagnostics = npr_res;
1091         package.response() = f_apdu_res;
1092         bc->release_backend(found_backend);
1093         return;
1094     }
1095
1096     found_backend->timestamp();
1097
1098     Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
1099     Z_PresentRequest *p_req = p_apdu->u.presentRequest;
1100     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1101     p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
1102     *p_req->resultSetStartPoint = *req->resultSetStartPoint;
1103     *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
1104     p_req->preferredRecordSyntax = req->preferredRecordSyntax;
1105     p_req->recordComposition = req->recordComposition;
1106
1107     Package present_package(found_backend->m_session, package.origin());
1108     present_package.copy_filter(package);
1109
1110     present_package.request() = p_apdu;
1111
1112     present_package.move();
1113
1114     Z_GDU *gdu = present_package.response().get();
1115     if (!present_package.session().is_closed()
1116         && gdu && gdu->which == Z_GDU_Z3950
1117         && gdu->u.z3950->which == Z_APDU_presentResponse)
1118     {
1119         Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
1120         Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
1121         Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
1122
1123         f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
1124         f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
1125         f_resp->presentStatus= b_resp->presentStatus;
1126         f_resp->records = b_resp->records;
1127         f_resp->otherInfo = b_resp->otherInfo;
1128         package.response() = f_apdu_res;
1129
1130         if (b_resp->records && b_resp->records->which ==  Z_Records_DBOSD)
1131         {
1132             Z_NamePlusRecordList *npr =
1133                 b_resp->records->u.databaseOrSurDiagnostics;
1134             // record_cache.add types are int's. Avoid non-fitting values
1135             if (*req->resultSetStartPoint > 0
1136                 && npr->num_records + *req->resultSetStartPoint < INT_MAX)
1137             {
1138 #if 0
1139                 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
1140                         " records to cache %p",
1141                         *req->resultSetStartPoint,
1142                         *f_resp->numberOfRecordsReturned,
1143                         &found_set->m_record_cache);
1144 #endif
1145                 found_set->m_record_cache.add(
1146                     odr, npr, *req->resultSetStartPoint,
1147                     p_req->recordComposition);
1148             }
1149         }
1150         bc->release_backend(found_backend);
1151     }
1152     else
1153     {
1154         bc->remove_backend(found_backend);
1155         Z_APDU *f_apdu_res =
1156             odr.create_presentResponse(
1157                 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1158                 "session_shared: target closed connection during present");
1159         package.response() = f_apdu_res;
1160     }
1161 }
1162
1163 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
1164                                        Z_APDU *apdu_req)
1165 {
1166     BackendClassPtr bc = m_backend_class;
1167     BackendInstancePtr backend = bc->get_backend(frontend_package);
1168     if (!backend)
1169     {
1170         mp::odr odr;
1171         Z_APDU *apdu = odr.create_scanResponse(
1172             apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR,
1173             "session_shared: could not create backend");
1174         frontend_package.response() = apdu;
1175     }
1176     else
1177     {
1178         Package scan_package(backend->m_session, frontend_package.origin());
1179         backend->timestamp();
1180         scan_package.copy_filter(frontend_package);
1181         scan_package.request() = apdu_req;
1182         scan_package.move();
1183         frontend_package.response() = scan_package.response();
1184         if (scan_package.session().is_closed())
1185         {
1186             frontend_package.session().close();
1187             bc->remove_backend(backend);
1188         }
1189         else
1190             bc->release_backend(backend);
1191     }
1192 }
1193
1194 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
1195 {
1196 }
1197
1198 void yf::SessionShared::Worker::operator() (void)
1199 {
1200     m_p->expire();
1201 }
1202
1203 bool yf::SessionShared::BackendClass::expire_instances()
1204 {
1205     time_t now;
1206     time(&now);
1207     boost::mutex::scoped_lock lock(m_mutex_backend_class);
1208     BackendInstanceList::iterator bit = m_backend_list.begin();
1209     while (bit != m_backend_list.end())
1210     {
1211         time_t last_use = (*bit)->m_time_last_use;
1212
1213         if ((*bit)->m_in_use)
1214         {
1215             bit++;
1216         }
1217         else if (now < last_use || now - last_use > m_backend_expiry_ttl)
1218         {
1219             mp::odr odr;
1220             (*bit)->m_close_package->response() = odr.create_close(
1221                 0, Z_Close_lackOfActivity, 0);
1222             (*bit)->m_close_package->session().close();
1223             (*bit)->m_close_package->move();
1224
1225             bit = m_backend_list.erase(bit);
1226         }
1227         else
1228         {
1229             bit++;
1230         }
1231     }
1232     if (m_backend_list.empty())
1233         return true;
1234     return false;
1235 }
1236
1237 void yf::SessionShared::Rep::expire_classes()
1238 {
1239     boost::mutex::scoped_lock lock(m_mutex_backend_map);
1240     BackendClassMap::iterator b_it = m_backend_map.begin();
1241     while (b_it != m_backend_map.end())
1242     {
1243         if (b_it->second->expire_instances())
1244         {
1245             m_backend_map.erase(b_it);
1246             b_it = m_backend_map.begin();
1247         }
1248         else
1249             b_it++;
1250     }
1251 }
1252
1253 void yf::SessionShared::Rep::expire()
1254 {
1255     while (true)
1256     {
1257         boost::xtime xt;
1258         boost::xtime_get(&xt,
1259 #if BOOST_VERSION >= 105000
1260                 boost::TIME_UTC_
1261 #else
1262                 boost::TIME_UTC
1263 #endif
1264                   );
1265         xt.sec += m_session_ttl / 3;
1266         {
1267             boost::mutex::scoped_lock lock(m_mutex);
1268             m_cond_expire_ready.timed_wait(lock, xt);
1269             if (close_down)
1270                 break;
1271         }
1272         stat();
1273         expire_classes();
1274     }
1275 }
1276
1277 yf::SessionShared::Rep::Rep()
1278 {
1279     m_resultset_ttl = 30;
1280     m_resultset_max = 10;
1281     m_session_ttl = 90;
1282     m_optimize_search = true;
1283     m_restart = false;
1284     m_session_max = 100;
1285     m_preferredMessageSize = 0;
1286     m_maximumRecordSize = 0;
1287     close_down = false;
1288 }
1289
1290 yf::SessionShared::Rep::~Rep()
1291 {
1292     {
1293         boost::mutex::scoped_lock lock(m_mutex);
1294         close_down = true;
1295         m_cond_expire_ready.notify_all();
1296     }
1297     m_thrds.join_all();
1298 }
1299
1300 void yf::SessionShared::Rep::start()
1301 {
1302     yf::SessionShared::Worker w(this);
1303     m_thrds.add_thread(new boost::thread(w));
1304 }
1305
1306 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1307 {
1308 }
1309
1310 yf::SessionShared::~SessionShared() {
1311 }
1312
1313 void yf::SessionShared::start() const
1314 {
1315     m_p->start();
1316 }
1317
1318 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1319 {
1320 }
1321
1322 yf::SessionShared::Frontend::~Frontend()
1323 {
1324 }
1325
1326 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1327 {
1328     boost::mutex::scoped_lock lock(m_mutex);
1329
1330     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1331
1332     while(true)
1333     {
1334         it = m_clients.find(package.session());
1335         if (it == m_clients.end())
1336             break;
1337
1338         if (!it->second->m_in_use)
1339         {
1340             it->second->m_in_use = true;
1341             return it->second;
1342         }
1343         m_cond_session_ready.wait(lock);
1344     }
1345     FrontendPtr f(new Frontend(this));
1346     m_clients[package.session()] = f;
1347     f->m_in_use = true;
1348     return f;
1349 }
1350
1351 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1352 {
1353     boost::mutex::scoped_lock lock(m_mutex);
1354     std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1355
1356     it = m_clients.find(package.session());
1357     if (it != m_clients.end())
1358     {
1359         if (package.session().is_closed())
1360         {
1361             m_clients.erase(it);
1362         }
1363         else
1364         {
1365             it->second->m_in_use = false;
1366         }
1367         m_cond_session_ready.notify_all();
1368     }
1369 }
1370
1371
1372 void yf::SessionShared::process(mp::Package &package) const
1373 {
1374     FrontendPtr f = m_p->get_frontend(package);
1375
1376     Z_GDU *gdu = package.request().get();
1377
1378     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1379         Z_APDU_initRequest && !f->m_is_virtual)
1380     {
1381         m_p->init(package, gdu, f);
1382     }
1383     else if (!f->m_is_virtual)
1384         package.move();
1385     else if (gdu && gdu->which == Z_GDU_Z3950)
1386     {
1387         Z_APDU *apdu = gdu->u.z3950;
1388         if (apdu->which == Z_APDU_initRequest)
1389         {
1390             mp::odr odr;
1391
1392             package.response() = odr.create_close(
1393                 apdu,
1394                 Z_Close_protocolError,
1395                 "double init");
1396
1397             package.session().close();
1398         }
1399         else if (apdu->which == Z_APDU_close)
1400         {
1401             mp::odr odr;
1402
1403             package.response() = odr.create_close(
1404                 apdu,
1405                 Z_Close_peerAbort, "received close from client");
1406             package.session().close();
1407         }
1408         else if (apdu->which == Z_APDU_searchRequest)
1409         {
1410             f->search(package, apdu);
1411         }
1412         else if (apdu->which == Z_APDU_presentRequest)
1413         {
1414             f->present(package, apdu);
1415         }
1416         else if (apdu->which == Z_APDU_scanRequest)
1417         {
1418             f->scan(package, apdu);
1419         }
1420         else
1421         {
1422             mp::odr odr;
1423
1424             package.response() = odr.create_close(
1425                 apdu, Z_Close_protocolError,
1426                 "unsupported APDU in filter_session_shared");
1427
1428             package.session().close();
1429         }
1430     }
1431     m_p->release_frontend(package);
1432 }
1433
1434 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1435                                   const char *path)
1436 {
1437     for (ptr = ptr->children; ptr; ptr = ptr->next)
1438     {
1439         if (ptr->type != XML_ELEMENT_NODE)
1440             continue;
1441         if (!strcmp((const char *) ptr->name, "resultset"))
1442         {
1443             const struct _xmlAttr *attr;
1444             for (attr = ptr->properties; attr; attr = attr->next)
1445             {
1446                 if (!strcmp((const char *) attr->name, "ttl"))
1447                     m_p->m_resultset_ttl =
1448                         mp::xml::get_int(attr->children, 30);
1449                 else if (!strcmp((const char *) attr->name, "max"))
1450                 {
1451                     m_p->m_resultset_max =
1452                         mp::xml::get_int(attr->children, 10);
1453                 }
1454                 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1455                 {
1456                     m_p->m_optimize_search =
1457                         mp::xml::get_bool(attr->children, true);
1458                 }
1459                 else if (!strcmp((const char *) attr->name, "restart"))
1460                 {
1461                     m_p->m_restart = mp::xml::get_bool(attr->children, true);
1462                 }
1463                 else
1464                     throw mp::filter::FilterException(
1465                         "Bad attribute " + std::string((const char *)
1466                                                        attr->name));
1467             }
1468         }
1469         else if (!strcmp((const char *) ptr->name, "session"))
1470         {
1471             const struct _xmlAttr *attr;
1472             for (attr = ptr->properties; attr; attr = attr->next)
1473             {
1474                 if (!strcmp((const char *) attr->name, "ttl"))
1475                     m_p->m_session_ttl =
1476                         mp::xml::get_int(attr->children, 90);
1477                 else if (!strcmp((const char *) attr->name, "max"))
1478                     m_p->m_session_max =
1479                         mp::xml::get_int(attr->children, 100);
1480                 else
1481                     throw mp::filter::FilterException(
1482                         "Bad attribute " + std::string((const char *)
1483                                                        attr->name));
1484             }
1485         }
1486         else if (!strcmp((const char *) ptr->name, "init"))
1487         {
1488             const struct _xmlAttr *attr;
1489             for (attr = ptr->properties; attr; attr = attr->next)
1490             {
1491                 if (!strcmp((const char *) attr->name, "maximum-record-size"))
1492                     m_p->m_maximumRecordSize =
1493                         mp::xml::get_int(attr->children, 0);
1494                 else if (!strcmp((const char *) attr->name,
1495                                  "preferred-message-size"))
1496                     m_p->m_preferredMessageSize =
1497                         mp::xml::get_int(attr->children, 0);
1498                 else
1499                     throw mp::filter::FilterException(
1500                         "Bad attribute " + std::string((const char *)
1501                                                        attr->name));
1502             }
1503         }
1504         else
1505         {
1506             throw mp::filter::FilterException("Bad element "
1507                                                + std::string((const char *)
1508                                                              ptr->name));
1509         }
1510     }
1511 }
1512
1513 static mp::filter::Base* filter_creator()
1514 {
1515     return new mp::filter::SessionShared;
1516 }
1517
1518 extern "C" {
1519     struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1520         0,
1521         "session_shared",
1522         filter_creator
1523     };
1524 }
1525
1526 /*
1527  * Local variables:
1528  * c-basic-offset: 4
1529  * c-file-style: "Stroustrup"
1530  * indent-tabs-mode: nil
1531  * End:
1532  * vim: shiftwidth=4 tabstop=8 expandtab
1533  */
1534