1 /* This file is part of Metaproxy.
2 Copyright (C) 2005-2011 Index Data
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
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
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
21 #include <metaproxy/filter.hpp>
22 #include <metaproxy/package.hpp>
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>
31 #include <metaproxy/util.hpp>
32 #include "filter_session_shared.hpp"
36 #include <yaz/otherinfo.h>
37 #include <yaz/diagbib1.h>
38 #include <yazpp/z-query.h>
39 #include <yazpp/record-cache.h>
44 namespace mp = metaproxy_1;
45 namespace yf = metaproxy_1::filter;
47 namespace metaproxy_1 {
50 // key for session.. We'll only share sessions with same InitKey
51 class SessionShared::InitKey {
53 bool operator < (const SessionShared::InitKey &k) const;
54 InitKey(Z_InitRequest *req);
55 InitKey(const InitKey &);
58 char *m_idAuthentication_buf;
59 int m_idAuthentication_size;
60 char *m_otherInfo_buf;
64 // worker thread .. for expiry of sessions
65 class SessionShared::Worker {
67 Worker(SessionShared::Rep *rep);
68 void operator() (void);
70 SessionShared::Rep *m_p;
73 class SessionShared::BackendSet {
75 std::string m_result_set_id;
76 Databases m_databases;
77 int m_result_set_size;
78 yazpp_1::Yaz_Z_Query m_query;
79 time_t m_time_last_use;
81 yazpp_1::RecordCache m_record_cache;
83 const std::string &result_set_id,
84 const Databases &databases,
85 const yazpp_1::Yaz_Z_Query &query);
87 Package &frontend_package,
88 Package &search_package,
89 const Z_APDU *apdu_req,
90 const BackendInstancePtr bp,
91 Z_Records **z_records);
93 // backend connection instance
94 class SessionShared::BackendInstance {
96 friend class BackendClass;
97 friend class BackendSet;
99 mp::Session m_session;
100 BackendSetList m_sets;
103 int m_result_set_sequence;
104 time_t m_time_last_use;
105 mp::Package * m_close_package;
108 // backends of some class (all with same InitKey)
109 class SessionShared::BackendClass : boost::noncopyable {
111 friend struct Frontend;
112 bool m_named_result_sets;
113 BackendInstanceList m_backend_list;
114 BackendInstancePtr create_backend(const Package &package);
115 void remove_backend(BackendInstancePtr b);
116 BackendInstancePtr get_backend(const Package &package);
117 void use_backend(BackendInstancePtr b);
118 void release_backend(BackendInstancePtr b);
120 yazpp_1::GDU m_init_request;
121 yazpp_1::GDU m_init_response;
122 boost::mutex m_mutex_backend_class;
124 time_t m_backend_set_ttl;
125 time_t m_backend_expiry_ttl;
126 size_t m_backend_set_max;
128 BackendClass(const yazpp_1::GDU &init_request,
134 // frontend result set
135 class SessionShared::FrontendSet {
136 Databases m_databases;
137 yazpp_1::Yaz_Z_Query m_query;
139 const Databases &get_databases();
140 const yazpp_1::Yaz_Z_Query &get_query();
142 const Databases &databases,
143 const yazpp_1::Yaz_Z_Query &query);
147 struct SessionShared::Frontend {
152 Z_Options m_init_options;
153 void search(Package &package, Z_APDU *apdu);
154 void present(Package &package, Z_APDU *apdu);
155 void scan(Package &package, Z_APDU *apdu);
157 void get_set(mp::Package &package,
158 const Z_APDU *apdu_req,
159 const Databases &databases,
160 yazpp_1::Yaz_Z_Query &query,
161 BackendInstancePtr &found_backend,
162 BackendSetPtr &found_set);
163 void override_set(BackendInstancePtr &found_backend,
164 std::string &result_set_id);
167 BackendClassPtr m_backend_class;
168 FrontendSets m_frontend_sets;
171 class SessionShared::Rep {
172 friend class SessionShared;
173 friend struct Frontend;
175 FrontendPtr get_frontend(Package &package);
176 void release_frontend(Package &package);
181 void init(Package &package, const Z_GDU *gdu,
182 FrontendPtr frontend);
184 boost::mutex m_mutex;
185 boost::condition m_cond_session_ready;
186 std::map<mp::Session, FrontendPtr> m_clients;
188 BackendClassMap m_backend_map;
189 boost::mutex m_mutex_backend_map;
190 boost::thread_group m_thrds;
194 bool m_optimize_search;
199 yf::SessionShared::FrontendSet::FrontendSet(
200 const Databases &databases,
201 const yazpp_1::Yaz_Z_Query &query)
202 : m_databases(databases), m_query(query)
206 const yf::SessionShared::Databases &
207 yf::SessionShared::FrontendSet::get_databases()
212 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
217 yf::SessionShared::InitKey::InitKey(const InitKey &k)
219 m_odr = odr_createmem(ODR_ENCODE);
221 m_idAuthentication_size = k.m_idAuthentication_size;
222 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
223 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
224 m_idAuthentication_size);
226 m_otherInfo_size = k.m_otherInfo_size;
227 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
228 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
232 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
234 m_odr = odr_createmem(ODR_ENCODE);
236 Z_IdAuthentication *t = req->idAuthentication;
237 z_IdAuthentication(m_odr, &t, 1, 0);
238 m_idAuthentication_buf =
239 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
241 Z_OtherInformation *o = req->otherInfo;
242 z_OtherInformation(m_odr, &o, 1, 0);
243 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
246 yf::SessionShared::InitKey::~InitKey()
251 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
255 c = mp::util::memcmp2(
256 (void*) m_idAuthentication_buf, m_idAuthentication_size,
257 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
263 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
264 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
272 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
274 boost::mutex::scoped_lock lock(m_mutex_backend_class);
279 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
281 BackendInstanceList::iterator it = m_backend_list.begin();
283 while (it != m_backend_list.end())
288 (*it)->m_close_package->response() = odr.create_close(
289 0, Z_Close_lackOfActivity, 0);
290 (*it)->m_close_package->session().close();
291 (*it)->m_close_package->move();
293 it = m_backend_list.erase(it);
302 yf::SessionShared::BackendInstancePtr
303 yf::SessionShared::BackendClass::get_backend(
304 const mp::Package &frontend_package)
307 boost::mutex::scoped_lock lock(m_mutex_backend_class);
309 BackendInstanceList::const_iterator it = m_backend_list.begin();
311 BackendInstancePtr backend1; // null
313 for (; it != m_backend_list.end(); it++)
315 if (!(*it)->m_in_use)
318 || (*it)->m_sequence_this < backend1->m_sequence_this)
324 use_backend(backend1);
328 return create_backend(frontend_package);
331 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
333 backend->m_in_use = true;
334 time(&backend->m_time_last_use);
335 backend->m_sequence_this = m_sequence_top++;
338 yf::SessionShared::BackendInstance::~BackendInstance()
340 delete m_close_package;
343 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
344 const mp::Package &frontend_package)
346 BackendInstancePtr bp(new BackendInstance);
347 BackendInstancePtr null;
349 bp->m_close_package =
350 new mp::Package(bp->m_session, frontend_package.origin());
351 bp->m_close_package->copy_filter(frontend_package);
353 Package init_package(bp->m_session, frontend_package.origin());
355 init_package.copy_filter(frontend_package);
357 yazpp_1::GDU actual_init_request = m_init_request;
358 Z_GDU *init_pdu = actual_init_request.get();
360 assert(init_pdu->which == Z_GDU_Z3950);
361 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
363 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
364 ODR_MASK_ZERO(req->options);
366 ODR_MASK_SET(req->options, Z_Options_search);
367 ODR_MASK_SET(req->options, Z_Options_present);
368 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
369 ODR_MASK_SET(req->options, Z_Options_scan);
371 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
372 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
373 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
375 init_package.request() = init_pdu;
379 boost::mutex::scoped_lock lock(m_mutex_backend_class);
381 m_named_result_sets = false;
382 Z_GDU *gdu = init_package.response().get();
383 if (init_package.session().is_closed())
385 /* already closed. We don't know why */
388 else if (gdu && gdu->which == Z_GDU_Z3950
389 && gdu->u.z3950->which == Z_APDU_initResponse
390 && *gdu->u.z3950->u.initResponse->result)
392 /* successful init response */
393 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
394 m_init_response = gdu->u.z3950;
395 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
397 m_named_result_sets = true;
402 /* not init or init rejected */
403 init_package.copy_filter(frontend_package);
404 init_package.session().close();
409 time(&bp->m_time_last_use);
410 bp->m_sequence_this = 0;
411 bp->m_result_set_sequence = 0;
412 m_backend_list.push_back(bp);
418 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
422 : m_named_result_sets(false), m_init_request(init_request),
423 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
424 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
427 yf::SessionShared::BackendClass::~BackendClass()
430 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
431 FrontendPtr frontend)
433 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
435 frontend->m_is_virtual = true;
436 frontend->m_init_options = *req->options;
439 boost::mutex::scoped_lock lock(m_mutex_backend_map);
440 BackendClassMap::const_iterator it;
441 it = m_backend_map.find(k);
442 if (it == m_backend_map.end())
444 BackendClassPtr b(new BackendClass(gdu->u.z3950,
448 m_backend_map[k] = b;
449 frontend->m_backend_class = b;
453 frontend->m_backend_class = it->second;
456 BackendClassPtr bc = frontend->m_backend_class;
457 BackendInstancePtr backend = bc->get_backend(package);
462 Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
463 *apdu->u.initResponse->result = 0;
464 package.response() = apdu;
465 package.session().close();
469 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
470 yazpp_1::GDU init_response = bc->m_init_response;
471 Z_GDU *response_gdu = init_response.get();
472 mp::util::transfer_referenceId(odr, gdu->u.z3950,
473 response_gdu->u.z3950);
475 Z_Options *server_options =
476 response_gdu->u.z3950->u.initResponse->options;
477 Z_Options *client_options = &frontend->m_init_options;
480 for (i = 0; i<30; i++)
481 if (!ODR_MASK_GET(client_options, i))
482 ODR_MASK_CLEAR(server_options, i);
483 package.response() = init_response;
486 bc->release_backend(backend);
489 void yf::SessionShared::BackendSet::timestamp()
491 time(&m_time_last_use);
494 yf::SessionShared::BackendSet::BackendSet(
495 const std::string &result_set_id,
496 const Databases &databases,
497 const yazpp_1::Yaz_Z_Query &query) :
498 m_result_set_id(result_set_id),
499 m_databases(databases), m_result_set_size(0), m_query(query)
504 static int get_diagnostic(Z_DefaultDiagFormat *r)
506 return *r->condition;
509 bool yf::SessionShared::BackendSet::search(
510 mp::Package &frontend_package,
511 mp::Package &search_package,
512 const Z_APDU *frontend_apdu,
513 const BackendInstancePtr bp,
514 Z_Records **z_records)
517 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
518 Z_SearchRequest *req = apdu_req->u.searchRequest;
520 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
521 req->query = m_query.get_Z_Query();
523 req->num_databaseNames = m_databases.size();
524 req->databaseNames = (char**)
525 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
526 Databases::const_iterator it = m_databases.begin();
528 for (; it != m_databases.end(); it++)
529 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
531 if (frontend_apdu->which == Z_APDU_searchRequest)
532 req->preferredRecordSyntax =
533 frontend_apdu->u.searchRequest->preferredRecordSyntax;
535 search_package.request() = apdu_req;
537 search_package.move();
539 Z_GDU *gdu = search_package.response().get();
540 if (!search_package.session().is_closed()
541 && gdu && gdu->which == Z_GDU_Z3950
542 && gdu->u.z3950->which == Z_APDU_searchResponse)
544 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
545 *z_records = b_resp->records;
546 m_result_set_size = *b_resp->resultCount;
550 if (frontend_apdu->which == Z_APDU_searchRequest)
551 f_apdu = odr.create_searchResponse(
552 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
553 else if (frontend_apdu->which == Z_APDU_presentRequest)
554 f_apdu = odr.create_presentResponse(
555 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
557 f_apdu = odr.create_close(
558 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
559 frontend_package.response() = f_apdu;
563 void yf::SessionShared::Frontend::override_set(
564 BackendInstancePtr &found_backend,
565 std::string &result_set_id)
567 BackendClassPtr bc = m_backend_class;
568 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
572 for (; it != bc->m_backend_list.end(); it++)
574 if (!(*it)->m_in_use)
576 BackendSetList::iterator set_it = (*it)->m_sets.begin();
577 for (; set_it != (*it)->m_sets.end(); set_it++)
579 if (now >= (*set_it)->m_time_last_use &&
580 now - (*set_it)->m_time_last_use > bc->m_backend_set_ttl)
583 result_set_id = (*set_it)->m_result_set_id;
584 found_backend->m_sets.erase(set_it);
590 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
591 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
593 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
596 if (bc->m_named_result_sets)
598 result_set_id = boost::io::str(
599 boost::format("%1%") %
600 found_backend->m_result_set_sequence);
601 found_backend->m_result_set_sequence++;
604 result_set_id = "default";
610 void yf::SessionShared::Frontend::get_set(mp::Package &package,
611 const Z_APDU *apdu_req,
612 const Databases &databases,
613 yazpp_1::Yaz_Z_Query &query,
614 BackendInstancePtr &found_backend,
615 BackendSetPtr &found_set)
617 bool session_restarted = false;
620 std::string result_set_id;
621 BackendClassPtr bc = m_backend_class;
623 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
625 if (m_p->m_optimize_search)
627 // look at each backend and see if we have a similar search
628 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
629 for (; it != bc->m_backend_list.end(); it++)
631 if (!(*it)->m_in_use)
633 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
634 for (; set_it != (*it)->m_sets.end(); set_it++)
636 if ((*set_it)->m_databases == databases
637 && query.match(&(*set_it)->m_query))
641 bc->use_backend(found_backend);
642 found_set->timestamp();
643 // found matching set. No need to search again
650 override_set(found_backend, result_set_id);
652 bc->use_backend(found_backend);
656 // create a new backend set (and new set)
657 found_backend = bc->create_backend(package);
663 if (apdu_req->which == Z_APDU_searchRequest)
665 f_apdu = odr.create_searchResponse(
666 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
668 else if (apdu_req->which == Z_APDU_presentRequest)
670 f_apdu = odr.create_presentResponse(
671 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
675 f_apdu = odr.create_close(
676 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
678 package.response() = f_apdu;
681 if (bc->m_named_result_sets)
683 result_set_id = boost::io::str(
684 boost::format("%1%") % found_backend->m_result_set_sequence);
685 found_backend->m_result_set_sequence++;
688 result_set_id = "default";
690 // we must search ...
691 BackendSetPtr new_set(new BackendSet(result_set_id,
693 Z_Records *z_records = 0;
695 Package search_package(found_backend->m_session, package.origin());
696 search_package.copy_filter(package);
698 if (!new_set->search(package, search_package,
699 apdu_req, found_backend, &z_records))
701 bc->remove_backend(found_backend);
702 return; // search error
708 if (z_records->which == Z_Records_NSD)
711 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
713 else if (z_records->which == Z_Records_multipleNSD)
715 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
718 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
719 Z_DiagRec_defaultFormat)
721 condition = get_diagnostic(
722 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
726 if (!session_restarted &&
727 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
729 bc->remove_backend(found_backend);
730 session_restarted = true;
731 found_backend.reset();
739 if (apdu_req->which == Z_APDU_searchRequest)
741 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
743 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
744 *f_resp->searchStatus = Z_SearchResponse_none;
745 f_resp->records = z_records;
746 package.response() = f_apdu;
748 if (apdu_req->which == Z_APDU_presentRequest)
750 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
752 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
753 f_resp->records = z_records;
754 package.response() = f_apdu;
756 bc->release_backend(found_backend);
757 return; // search error
760 if (!session_restarted && new_set->m_result_set_size < 0)
762 bc->remove_backend(found_backend);
763 session_restarted = true;
764 found_backend.reset();
769 found_set->timestamp();
770 found_backend->m_sets.push_back(found_set);
773 void yf::SessionShared::Frontend::search(mp::Package &package,
776 Z_SearchRequest *req = apdu_req->u.searchRequest;
777 FrontendSets::iterator fset_it =
778 m_frontend_sets.find(req->resultSetName);
779 if (fset_it != m_frontend_sets.end())
781 // result set already exist
782 // if replace indicator is off: we return diagnostic if
783 // result set already exist.
784 if (*req->replaceIndicator == 0)
788 odr.create_searchResponse(
790 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
792 package.response() = apdu;
796 m_frontend_sets.erase(fset_it);
799 yazpp_1::Yaz_Z_Query query;
800 query.set_Z_Query(req->query);
803 for (i = 0; i<req->num_databaseNames; i++)
804 databases.push_back(req->databaseNames[i]);
806 BackendSetPtr found_set; // null
807 BackendInstancePtr found_backend; // null
809 get_set(package, apdu_req, databases, query, found_backend, found_set);
814 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
815 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
816 *f_resp->resultCount = found_set->m_result_set_size;
817 package.response() = f_apdu;
819 FrontendSetPtr fset(new FrontendSet(databases, query));
820 m_frontend_sets[req->resultSetName] = fset;
822 m_backend_class->release_backend(found_backend);
825 void yf::SessionShared::Frontend::present(mp::Package &package,
829 Z_PresentRequest *req = apdu_req->u.presentRequest;
831 FrontendSets::iterator fset_it =
832 m_frontend_sets.find(req->resultSetId);
834 if (fset_it == m_frontend_sets.end())
837 odr.create_presentResponse(
839 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
841 package.response() = apdu;
844 FrontendSetPtr fset = fset_it->second;
846 Databases databases = fset->get_databases();
847 yazpp_1::Yaz_Z_Query query = fset->get_query();
849 BackendClassPtr bc = m_backend_class;
850 BackendSetPtr found_set; // null
851 BackendInstancePtr found_backend;
853 get_set(package, apdu_req, databases, query, found_backend, found_set);
857 Z_NamePlusRecordList *npr_res = 0;
858 if (found_set->m_record_cache.lookup(odr, &npr_res,
859 *req->resultSetStartPoint,
860 *req->numberOfRecordsRequested,
861 req->preferredRecordSyntax,
862 req->recordComposition))
864 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
865 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
867 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
868 " records in cache %p",
869 *req->resultSetStartPoint,
870 *req->numberOfRecordsRequested,
871 &found_set->m_record_cache);
873 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
874 *f_resp->nextResultSetPosition =
875 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
876 // f_resp->presentStatus assumed OK.
877 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
878 f_resp->records->which = Z_Records_DBOSD;
879 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
880 package.response() = f_apdu_res;
881 bc->release_backend(found_backend);
885 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
886 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
887 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
888 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
889 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
890 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
891 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
892 p_req->recordComposition = req->recordComposition;
894 Package present_package(found_backend->m_session, package.origin());
895 present_package.copy_filter(package);
897 present_package.request() = p_apdu;
899 present_package.move();
901 Z_GDU *gdu = present_package.response().get();
902 if (!present_package.session().is_closed()
903 && gdu && gdu->which == Z_GDU_Z3950
904 && gdu->u.z3950->which == Z_APDU_presentResponse)
906 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
907 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
908 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
910 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
911 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
912 f_resp->presentStatus= b_resp->presentStatus;
913 f_resp->records = b_resp->records;
914 f_resp->otherInfo = b_resp->otherInfo;
915 package.response() = f_apdu_res;
917 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
919 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
920 " records to cache %p",
921 *req->resultSetStartPoint,
922 *f_resp->numberOfRecordsReturned,
923 &found_set->m_record_cache);
924 found_set->m_record_cache.add(
926 b_resp->records->u.databaseOrSurDiagnostics,
927 *req->resultSetStartPoint,
928 *f_resp->numberOfRecordsReturned);
930 bc->release_backend(found_backend);
934 bc->remove_backend(found_backend);
936 odr.create_presentResponse(
937 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
938 package.response() = f_apdu_res;
942 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
945 BackendClassPtr bc = m_backend_class;
946 BackendInstancePtr backend = bc->get_backend(frontend_package);
950 Z_APDU *apdu = odr.create_scanResponse(
951 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
952 frontend_package.response() = apdu;
956 Package scan_package(backend->m_session, frontend_package.origin());
957 scan_package.copy_filter(frontend_package);
958 scan_package.request() = apdu_req;
960 frontend_package.response() = scan_package.response();
961 if (scan_package.session().is_closed())
963 frontend_package.session().close();
964 bc->remove_backend(backend);
967 bc->release_backend(backend);
971 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
975 void yf::SessionShared::Worker::operator() (void)
980 void yf::SessionShared::BackendClass::expire_class()
984 boost::mutex::scoped_lock lock(m_mutex_backend_class);
985 BackendInstanceList::iterator bit = m_backend_list.begin();
986 while (bit != m_backend_list.end())
988 time_t last_use = (*bit)->m_time_last_use;
990 if ((*bit)->m_in_use)
994 else if ((now >= last_use && now - last_use > m_backend_expiry_ttl)
998 (*bit)->m_close_package->response() = odr.create_close(
999 0, Z_Close_lackOfActivity, 0);
1000 (*bit)->m_close_package->session().close();
1001 (*bit)->m_close_package->move();
1003 bit = m_backend_list.erase(bit);
1012 void yf::SessionShared::Rep::expire()
1017 boost::xtime_get(&xt, boost::TIME_UTC);
1019 boost::thread::sleep(xt);
1021 BackendClassMap::const_iterator b_it = m_backend_map.begin();
1022 for (; b_it != m_backend_map.end(); b_it++)
1023 b_it->second->expire_class();
1027 yf::SessionShared::Rep::Rep()
1029 m_resultset_ttl = 30;
1030 m_resultset_max = 10;
1032 m_optimize_search = true;
1035 void yf::SessionShared::Rep::start()
1037 yf::SessionShared::Worker w(this);
1038 m_thrds.add_thread(new boost::thread(w));
1041 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1045 yf::SessionShared::~SessionShared() {
1048 void yf::SessionShared::start() const
1053 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1057 yf::SessionShared::Frontend::~Frontend()
1061 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1063 boost::mutex::scoped_lock lock(m_mutex);
1065 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1069 it = m_clients.find(package.session());
1070 if (it == m_clients.end())
1073 if (!it->second->m_in_use)
1075 it->second->m_in_use = true;
1078 m_cond_session_ready.wait(lock);
1080 FrontendPtr f(new Frontend(this));
1081 m_clients[package.session()] = f;
1086 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1088 boost::mutex::scoped_lock lock(m_mutex);
1089 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1091 it = m_clients.find(package.session());
1092 if (it != m_clients.end())
1094 if (package.session().is_closed())
1096 m_clients.erase(it);
1100 it->second->m_in_use = false;
1102 m_cond_session_ready.notify_all();
1107 void yf::SessionShared::process(mp::Package &package) const
1109 FrontendPtr f = m_p->get_frontend(package);
1111 Z_GDU *gdu = package.request().get();
1113 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1114 Z_APDU_initRequest && !f->m_is_virtual)
1116 m_p->init(package, gdu, f);
1118 else if (!f->m_is_virtual)
1120 else if (gdu && gdu->which == Z_GDU_Z3950)
1122 Z_APDU *apdu = gdu->u.z3950;
1123 if (apdu->which == Z_APDU_initRequest)
1127 package.response() = odr.create_close(
1129 Z_Close_protocolError,
1132 package.session().close();
1134 else if (apdu->which == Z_APDU_close)
1138 package.response() = odr.create_close(
1140 Z_Close_peerAbort, "received close from client");
1141 package.session().close();
1143 else if (apdu->which == Z_APDU_searchRequest)
1145 f->search(package, apdu);
1147 else if (apdu->which == Z_APDU_presentRequest)
1149 f->present(package, apdu);
1151 else if (apdu->which == Z_APDU_scanRequest)
1153 f->scan(package, apdu);
1159 package.response() = odr.create_close(
1160 apdu, Z_Close_protocolError,
1161 "unsupported APDU in filter_session_shared");
1163 package.session().close();
1166 m_p->release_frontend(package);
1169 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only,
1172 for (ptr = ptr->children; ptr; ptr = ptr->next)
1174 if (ptr->type != XML_ELEMENT_NODE)
1176 if (!strcmp((const char *) ptr->name, "resultset"))
1178 const struct _xmlAttr *attr;
1179 for (attr = ptr->properties; attr; attr = attr->next)
1181 if (!strcmp((const char *) attr->name, "ttl"))
1182 m_p->m_resultset_ttl =
1183 mp::xml::get_int(attr->children, 30);
1184 else if (!strcmp((const char *) attr->name, "max"))
1186 m_p->m_resultset_max =
1187 mp::xml::get_int(attr->children, 10);
1189 else if (!strcmp((const char *) attr->name, "optimizesearch"))
1191 m_p->m_optimize_search =
1192 mp::xml::get_bool(attr->children, true);
1195 throw mp::filter::FilterException(
1196 "Bad attribute " + std::string((const char *)
1200 else if (!strcmp((const char *) ptr->name, "session"))
1202 const struct _xmlAttr *attr;
1203 for (attr = ptr->properties; attr; attr = attr->next)
1205 if (!strcmp((const char *) attr->name, "ttl"))
1206 m_p->m_session_ttl =
1207 mp::xml::get_int(attr->children, 120);
1209 throw mp::filter::FilterException(
1210 "Bad attribute " + std::string((const char *)
1216 throw mp::filter::FilterException("Bad element "
1217 + std::string((const char *)
1223 static mp::filter::Base* filter_creator()
1225 return new mp::filter::SessionShared;
1229 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1239 * c-file-style: "Stroustrup"
1240 * indent-tabs-mode: nil
1242 * vim: shiftwidth=4 tabstop=8 expandtab