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