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