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