1 /* This file is part of Metaproxy.
2 Copyright (C) 2005-2010 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 const Z_APDU *apdu_req,
89 const BackendInstancePtr bp,
90 Z_Records **z_records);
92 // backend connection instance
93 class SessionShared::BackendInstance {
95 friend class BackendClass;
96 friend class BackendSet;
98 mp::Session m_session;
99 BackendSetList m_sets;
102 int m_result_set_sequence;
103 time_t m_time_last_use;
104 mp::Package * m_close_package;
107 // backends of some class (all with same InitKey)
108 class SessionShared::BackendClass : boost::noncopyable {
110 friend struct Frontend;
111 bool m_named_result_sets;
112 BackendInstanceList m_backend_list;
113 BackendInstancePtr create_backend(const Package &package);
114 void remove_backend(BackendInstancePtr b);
115 BackendInstancePtr get_backend(const Package &package);
116 void use_backend(BackendInstancePtr b);
117 void release_backend(BackendInstancePtr b);
119 yazpp_1::GDU m_init_request;
120 yazpp_1::GDU m_init_response;
121 boost::mutex m_mutex_backend_class;
123 time_t m_backend_set_ttl;
124 time_t m_backend_expiry_ttl;
125 size_t m_backend_set_max;
127 BackendClass(const yazpp_1::GDU &init_request,
133 // frontend result set
134 class SessionShared::FrontendSet {
135 Databases m_databases;
136 yazpp_1::Yaz_Z_Query m_query;
138 const Databases &get_databases();
139 const yazpp_1::Yaz_Z_Query &get_query();
141 const Databases &databases,
142 const yazpp_1::Yaz_Z_Query &query);
146 struct SessionShared::Frontend {
151 Z_Options m_init_options;
152 void search(Package &package, Z_APDU *apdu);
153 void present(Package &package, Z_APDU *apdu);
154 void scan(Package &package, Z_APDU *apdu);
156 void get_set(mp::Package &package,
157 const Z_APDU *apdu_req,
158 const Databases &databases,
159 yazpp_1::Yaz_Z_Query &query,
160 BackendInstancePtr &found_backend,
161 BackendSetPtr &found_set);
162 void override_set(BackendInstancePtr &found_backend,
163 std::string &result_set_id);
166 BackendClassPtr m_backend_class;
167 FrontendSets m_frontend_sets;
170 class SessionShared::Rep {
171 friend class SessionShared;
172 friend struct Frontend;
174 FrontendPtr get_frontend(Package &package);
175 void release_frontend(Package &package);
180 void init(Package &package, const Z_GDU *gdu,
181 FrontendPtr frontend);
182 boost::mutex m_mutex;
183 boost::condition m_cond_session_ready;
184 std::map<mp::Session, FrontendPtr> m_clients;
186 BackendClassMap m_backend_map;
187 boost::mutex m_mutex_backend_map;
188 boost::thread_group m_thrds;
196 yf::SessionShared::FrontendSet::FrontendSet(
197 const Databases &databases,
198 const yazpp_1::Yaz_Z_Query &query)
199 : m_databases(databases), m_query(query)
203 const yf::SessionShared::Databases &
204 yf::SessionShared::FrontendSet::get_databases()
209 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
214 yf::SessionShared::InitKey::InitKey(const InitKey &k)
216 m_odr = odr_createmem(ODR_ENCODE);
218 m_idAuthentication_size = k.m_idAuthentication_size;
219 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
220 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
221 m_idAuthentication_size);
223 m_otherInfo_size = k.m_otherInfo_size;
224 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
225 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
229 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
231 m_odr = odr_createmem(ODR_ENCODE);
233 Z_IdAuthentication *t = req->idAuthentication;
234 z_IdAuthentication(m_odr, &t, 1, 0);
235 m_idAuthentication_buf =
236 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
238 Z_OtherInformation *o = req->otherInfo;
239 z_OtherInformation(m_odr, &o, 1, 0);
240 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
243 yf::SessionShared::InitKey::~InitKey()
248 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
252 c = mp::util::memcmp2(
253 (void*) m_idAuthentication_buf, m_idAuthentication_size,
254 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
260 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
261 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
269 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
271 boost::mutex::scoped_lock lock(m_mutex_backend_class);
276 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
278 BackendInstanceList::iterator it = m_backend_list.begin();
280 while (it != m_backend_list.end())
285 (*it)->m_close_package->response() = odr.create_close(
286 0, Z_Close_lackOfActivity, 0);
287 (*it)->m_close_package->session().close();
288 (*it)->m_close_package->move();
290 it = m_backend_list.erase(it);
299 yf::SessionShared::BackendInstancePtr
300 yf::SessionShared::BackendClass::get_backend(
301 const mp::Package &frontend_package)
304 boost::mutex::scoped_lock lock(m_mutex_backend_class);
306 BackendInstanceList::const_iterator it = m_backend_list.begin();
308 BackendInstancePtr backend1; // null
310 for (; it != m_backend_list.end(); it++)
312 if (!(*it)->m_in_use)
315 || (*it)->m_sequence_this < backend1->m_sequence_this)
321 use_backend(backend1);
325 return create_backend(frontend_package);
328 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
330 backend->m_in_use = true;
331 time(&backend->m_time_last_use);
332 backend->m_sequence_this = m_sequence_top++;
335 yf::SessionShared::BackendInstance::~BackendInstance()
337 delete m_close_package;
340 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
341 const mp::Package &frontend_package)
343 BackendInstancePtr bp(new BackendInstance);
344 BackendInstancePtr null;
346 bp->m_close_package =
347 new mp::Package(bp->m_session, frontend_package.origin());
348 bp->m_close_package->copy_filter(frontend_package);
350 Package init_package(bp->m_session, frontend_package.origin());
352 init_package.copy_filter(frontend_package);
354 yazpp_1::GDU actual_init_request = m_init_request;
355 Z_GDU *init_pdu = actual_init_request.get();
357 assert(init_pdu->which == Z_GDU_Z3950);
358 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
360 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
361 ODR_MASK_ZERO(req->options);
363 ODR_MASK_SET(req->options, Z_Options_search);
364 ODR_MASK_SET(req->options, Z_Options_present);
365 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
366 ODR_MASK_SET(req->options, Z_Options_scan);
368 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
369 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
370 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
372 init_package.request() = init_pdu;
376 boost::mutex::scoped_lock lock(m_mutex_backend_class);
378 m_named_result_sets = false;
379 Z_GDU *gdu = init_package.response().get();
380 if (!init_package.session().is_closed()
381 && gdu && gdu->which == Z_GDU_Z3950
382 && gdu->u.z3950->which == Z_APDU_initResponse)
384 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
387 m_init_response = gdu->u.z3950;
388 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
390 m_named_result_sets = true;
395 // did not receive an init response or closed
399 time(&bp->m_time_last_use);
400 bp->m_sequence_this = 0;
401 bp->m_result_set_sequence = 0;
402 m_backend_list.push_back(bp);
408 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
412 : m_named_result_sets(false), m_init_request(init_request),
413 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
414 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
417 yf::SessionShared::BackendClass::~BackendClass()
420 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
421 FrontendPtr frontend)
423 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
425 frontend->m_is_virtual = true;
426 frontend->m_init_options = *req->options;
429 boost::mutex::scoped_lock lock(m_mutex_backend_map);
430 BackendClassMap::const_iterator it;
431 it = m_backend_map.find(k);
432 if (it == m_backend_map.end())
434 BackendClassPtr b(new BackendClass(gdu->u.z3950,
438 m_backend_map[k] = b;
439 frontend->m_backend_class = b;
443 frontend->m_backend_class = it->second;
446 BackendClassPtr bc = frontend->m_backend_class;
447 BackendInstancePtr backend = bc->get_backend(package);
452 Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
453 *apdu->u.initResponse->result = 0;
454 package.response() = apdu;
455 package.session().close();
459 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
460 yazpp_1::GDU init_response = bc->m_init_response;
461 Z_GDU *response_gdu = init_response.get();
462 mp::util::transfer_referenceId(odr, gdu->u.z3950,
463 response_gdu->u.z3950);
465 Z_Options *server_options =
466 response_gdu->u.z3950->u.initResponse->options;
467 Z_Options *client_options = &frontend->m_init_options;
470 for (i = 0; i<30; i++)
471 if (!ODR_MASK_GET(client_options, i))
472 ODR_MASK_CLEAR(server_options, i);
473 package.response() = init_response;
476 bc->release_backend(backend);
479 void yf::SessionShared::BackendSet::timestamp()
481 time(&m_time_last_use);
484 yf::SessionShared::BackendSet::BackendSet(
485 const std::string &result_set_id,
486 const Databases &databases,
487 const yazpp_1::Yaz_Z_Query &query) :
488 m_result_set_id(result_set_id),
489 m_databases(databases), m_result_set_size(0), m_query(query)
494 static int get_diagnostic(Z_DefaultDiagFormat *r)
496 return *r->condition;
499 bool yf::SessionShared::BackendSet::search(
500 mp::Package &frontend_package,
501 const Z_APDU *frontend_apdu,
502 const BackendInstancePtr bp,
503 Z_Records **z_records)
505 Package search_package(bp->m_session, frontend_package.origin());
507 search_package.copy_filter(frontend_package);
510 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
511 Z_SearchRequest *req = apdu_req->u.searchRequest;
513 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
514 req->query = m_query.get_Z_Query();
516 req->num_databaseNames = m_databases.size();
517 req->databaseNames = (char**)
518 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
519 Databases::const_iterator it = m_databases.begin();
521 for (; it != m_databases.end(); it++)
522 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
524 search_package.request() = apdu_req;
526 search_package.move();
528 Z_GDU *gdu = search_package.response().get();
529 if (!search_package.session().is_closed()
530 && gdu && gdu->which == Z_GDU_Z3950
531 && gdu->u.z3950->which == Z_APDU_searchResponse)
533 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
534 *z_records = b_resp->records;
535 m_result_set_size = *b_resp->resultCount;
539 if (frontend_apdu->which == Z_APDU_searchRequest)
540 f_apdu = odr.create_searchResponse(
541 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
542 else if (frontend_apdu->which == Z_APDU_presentRequest)
543 f_apdu = odr.create_presentResponse(
544 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
546 f_apdu = odr.create_close(
547 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
548 frontend_package.response() = f_apdu;
552 void yf::SessionShared::Frontend::override_set(
553 BackendInstancePtr &found_backend,
554 std::string &result_set_id)
556 BackendClassPtr bc = m_backend_class;
557 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
561 for (; it != bc->m_backend_list.end(); it++)
563 if (!(*it)->m_in_use)
565 BackendSetList::iterator set_it = (*it)->m_sets.begin();
566 for (; set_it != (*it)->m_sets.end(); set_it++)
568 if (now >= (*set_it)->m_time_last_use &&
569 now - (*set_it)->m_time_last_use > bc->m_backend_set_ttl)
572 result_set_id = (*set_it)->m_result_set_id;
573 found_backend->m_sets.erase(set_it);
579 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
580 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
582 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
585 if (bc->m_named_result_sets)
587 result_set_id = boost::io::str(
588 boost::format("%1%") %
589 found_backend->m_result_set_sequence);
590 found_backend->m_result_set_sequence++;
593 result_set_id = "default";
599 void yf::SessionShared::Frontend::get_set(mp::Package &package,
600 const Z_APDU *apdu_req,
601 const Databases &databases,
602 yazpp_1::Yaz_Z_Query &query,
603 BackendInstancePtr &found_backend,
604 BackendSetPtr &found_set)
606 bool session_restarted = false;
609 std::string result_set_id;
610 BackendClassPtr bc = m_backend_class;
612 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
614 // look at each backend and see if we have a similar search
615 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
617 for (; it != bc->m_backend_list.end(); it++)
619 if (!(*it)->m_in_use)
621 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
622 for (; set_it != (*it)->m_sets.end(); set_it++)
624 if ((*set_it)->m_databases == databases
625 && query.match(&(*set_it)->m_query))
629 bc->use_backend(found_backend);
630 found_set->timestamp();
631 // found matching set. No need to search again
637 override_set(found_backend, result_set_id);
639 bc->use_backend(found_backend);
643 // create a new backend set (and new set)
644 found_backend = bc->create_backend(package);
650 if (apdu_req->which == Z_APDU_searchRequest)
652 f_apdu = odr.create_searchResponse(
653 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
655 else if (apdu_req->which == Z_APDU_presentRequest)
657 f_apdu = odr.create_presentResponse(
658 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
662 f_apdu = odr.create_close(
663 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
665 package.response() = f_apdu;
668 if (bc->m_named_result_sets)
670 result_set_id = boost::io::str(
671 boost::format("%1%") % found_backend->m_result_set_sequence);
672 found_backend->m_result_set_sequence++;
675 result_set_id = "default";
677 // we must search ...
678 BackendSetPtr new_set(new BackendSet(result_set_id,
680 Z_Records *z_records = 0;
681 if (!new_set->search(package, apdu_req, found_backend, &z_records))
683 bc->remove_backend(found_backend);
684 return; // search error
690 if (z_records->which == Z_Records_NSD)
693 get_diagnostic(z_records->u.nonSurrogateDiagnostic);
695 else if (z_records->which == Z_Records_multipleNSD)
697 if (z_records->u.multipleNonSurDiagnostics->num_diagRecs >= 1
700 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->which ==
701 Z_DiagRec_defaultFormat)
703 condition = get_diagnostic(
704 z_records->u.multipleNonSurDiagnostics->diagRecs[0]->u.defaultFormat);
708 if (!session_restarted &&
709 condition == YAZ_BIB1_TEMPORARY_SYSTEM_ERROR)
711 bc->remove_backend(found_backend);
712 session_restarted = true;
713 found_backend.reset();
721 if (apdu_req->which == Z_APDU_searchRequest)
723 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req,
725 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
726 *f_resp->searchStatus = Z_SearchResponse_none;
727 f_resp->records = z_records;
728 package.response() = f_apdu;
730 if (apdu_req->which == Z_APDU_presentRequest)
732 Z_APDU *f_apdu = odr.create_presentResponse(apdu_req,
734 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
735 f_resp->records = z_records;
736 package.response() = f_apdu;
738 bc->release_backend(found_backend);
739 return; // search error
742 if (!session_restarted && new_set->m_result_set_size < 0)
744 bc->remove_backend(found_backend);
745 session_restarted = true;
746 found_backend.reset();
751 found_set->timestamp();
752 found_backend->m_sets.push_back(found_set);
755 void yf::SessionShared::Frontend::search(mp::Package &package,
758 Z_SearchRequest *req = apdu_req->u.searchRequest;
759 FrontendSets::iterator fset_it =
760 m_frontend_sets.find(req->resultSetName);
761 if (fset_it != m_frontend_sets.end())
763 // result set already exist
764 // if replace indicator is off: we return diagnostic if
765 // result set already exist.
766 if (*req->replaceIndicator == 0)
770 odr.create_searchResponse(
772 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
774 package.response() = apdu;
778 m_frontend_sets.erase(fset_it);
781 yazpp_1::Yaz_Z_Query query;
782 query.set_Z_Query(req->query);
785 for (i = 0; i<req->num_databaseNames; i++)
786 databases.push_back(req->databaseNames[i]);
788 BackendSetPtr found_set; // null
789 BackendInstancePtr found_backend; // null
791 get_set(package, apdu_req, databases, query, found_backend, found_set);
796 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
797 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
798 *f_resp->resultCount = found_set->m_result_set_size;
799 package.response() = f_apdu;
801 FrontendSetPtr fset(new FrontendSet(databases, query));
802 m_frontend_sets[req->resultSetName] = fset;
804 m_backend_class->release_backend(found_backend);
807 void yf::SessionShared::Frontend::present(mp::Package &package,
811 Z_PresentRequest *req = apdu_req->u.presentRequest;
813 FrontendSets::iterator fset_it =
814 m_frontend_sets.find(req->resultSetId);
816 if (fset_it == m_frontend_sets.end())
819 odr.create_presentResponse(
821 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
823 package.response() = apdu;
826 FrontendSetPtr fset = fset_it->second;
828 Databases databases = fset->get_databases();
829 yazpp_1::Yaz_Z_Query query = fset->get_query();
831 BackendClassPtr bc = m_backend_class;
832 BackendSetPtr found_set; // null
833 BackendInstancePtr found_backend;
835 get_set(package, apdu_req, databases, query, found_backend, found_set);
839 Z_NamePlusRecordList *npr_res = 0;
840 if (found_set->m_record_cache.lookup(odr, &npr_res,
841 *req->resultSetStartPoint,
842 *req->numberOfRecordsRequested,
843 req->preferredRecordSyntax,
844 req->recordComposition))
846 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
847 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
849 yaz_log(YLOG_LOG, "Found " ODR_INT_PRINTF "+" ODR_INT_PRINTF
850 " records in cache %p",
851 *req->resultSetStartPoint,
852 *req->numberOfRecordsRequested,
853 &found_set->m_record_cache);
855 *f_resp->numberOfRecordsReturned = *req->numberOfRecordsRequested;
856 *f_resp->nextResultSetPosition =
857 *req->resultSetStartPoint + *req->numberOfRecordsRequested;
858 // f_resp->presentStatus assumed OK.
859 f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
860 f_resp->records->which = Z_Records_DBOSD;
861 f_resp->records->u.databaseOrSurDiagnostics = npr_res;
862 package.response() = f_apdu_res;
863 bc->release_backend(found_backend);
867 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
868 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
869 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
870 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
871 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
872 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
873 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
874 p_req->recordComposition = req->recordComposition;
876 Package present_package(found_backend->m_session, package.origin());
877 present_package.copy_filter(package);
879 present_package.request() = p_apdu;
881 present_package.move();
883 Z_GDU *gdu = present_package.response().get();
884 if (!present_package.session().is_closed()
885 && gdu && gdu->which == Z_GDU_Z3950
886 && gdu->u.z3950->which == Z_APDU_presentResponse)
888 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
889 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
890 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
892 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
893 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
894 f_resp->presentStatus= b_resp->presentStatus;
895 f_resp->records = b_resp->records;
896 f_resp->otherInfo = b_resp->otherInfo;
897 package.response() = f_apdu_res;
899 if (b_resp->records && b_resp->records->which == Z_Records_DBOSD)
901 yaz_log(YLOG_LOG, "Adding " ODR_INT_PRINTF "+" ODR_INT_PRINTF
902 " records to cache %p",
903 *req->resultSetStartPoint,
904 *f_resp->numberOfRecordsReturned,
905 &found_set->m_record_cache);
906 found_set->m_record_cache.add(
908 b_resp->records->u.databaseOrSurDiagnostics,
909 *req->resultSetStartPoint,
910 *f_resp->numberOfRecordsReturned);
912 bc->release_backend(found_backend);
916 bc->remove_backend(found_backend);
918 odr.create_presentResponse(
919 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
920 package.response() = f_apdu_res;
924 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
927 BackendClassPtr bc = m_backend_class;
928 BackendInstancePtr backend = bc->get_backend(frontend_package);
932 Z_APDU *apdu = odr.create_scanResponse(
933 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
934 frontend_package.response() = apdu;
938 Package scan_package(backend->m_session, frontend_package.origin());
939 scan_package.copy_filter(frontend_package);
940 scan_package.request() = apdu_req;
942 frontend_package.response() = scan_package.response();
943 if (scan_package.session().is_closed())
945 frontend_package.session().close();
946 bc->remove_backend(backend);
949 bc->release_backend(backend);
953 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
957 void yf::SessionShared::Worker::operator() (void)
962 void yf::SessionShared::BackendClass::expire_class()
966 boost::mutex::scoped_lock lock(m_mutex_backend_class);
967 BackendInstanceList::iterator bit = m_backend_list.begin();
968 while (bit != m_backend_list.end())
970 time_t last_use = (*bit)->m_time_last_use;
972 if ((*bit)->m_in_use)
976 else if ((now >= last_use && now - last_use > m_backend_expiry_ttl)
980 (*bit)->m_close_package->response() = odr.create_close(
981 0, Z_Close_lackOfActivity, 0);
982 (*bit)->m_close_package->session().close();
983 (*bit)->m_close_package->move();
985 bit = m_backend_list.erase(bit);
994 void yf::SessionShared::Rep::expire()
999 boost::xtime_get(&xt, boost::TIME_UTC);
1001 boost::thread::sleep(xt);
1003 BackendClassMap::const_iterator b_it = m_backend_map.begin();
1004 for (; b_it != m_backend_map.end(); b_it++)
1005 b_it->second->expire_class();
1009 yf::SessionShared::Rep::Rep()
1011 m_resultset_ttl = 30;
1012 m_resultset_max = 10;
1014 yf::SessionShared::Worker w(this);
1015 m_thrds.add_thread(new boost::thread(w));
1018 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
1022 yf::SessionShared::~SessionShared() {
1026 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
1030 yf::SessionShared::Frontend::~Frontend()
1034 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
1036 boost::mutex::scoped_lock lock(m_mutex);
1038 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1042 it = m_clients.find(package.session());
1043 if (it == m_clients.end())
1046 if (!it->second->m_in_use)
1048 it->second->m_in_use = true;
1051 m_cond_session_ready.wait(lock);
1053 FrontendPtr f(new Frontend(this));
1054 m_clients[package.session()] = f;
1059 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
1061 boost::mutex::scoped_lock lock(m_mutex);
1062 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
1064 it = m_clients.find(package.session());
1065 if (it != m_clients.end())
1067 if (package.session().is_closed())
1069 m_clients.erase(it);
1073 it->second->m_in_use = false;
1075 m_cond_session_ready.notify_all();
1080 void yf::SessionShared::process(mp::Package &package) const
1082 FrontendPtr f = m_p->get_frontend(package);
1084 Z_GDU *gdu = package.request().get();
1086 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1087 Z_APDU_initRequest && !f->m_is_virtual)
1089 m_p->init(package, gdu, f);
1091 else if (!f->m_is_virtual)
1093 else if (gdu && gdu->which == Z_GDU_Z3950)
1095 Z_APDU *apdu = gdu->u.z3950;
1096 if (apdu->which == Z_APDU_initRequest)
1100 package.response() = odr.create_close(
1102 Z_Close_protocolError,
1105 package.session().close();
1107 else if (apdu->which == Z_APDU_close)
1111 package.response() = odr.create_close(
1113 Z_Close_peerAbort, "received close from client");
1114 package.session().close();
1116 else if (apdu->which == Z_APDU_searchRequest)
1118 f->search(package, apdu);
1120 else if (apdu->which == Z_APDU_presentRequest)
1122 f->present(package, apdu);
1124 else if (apdu->which == Z_APDU_scanRequest)
1126 f->scan(package, apdu);
1132 package.response() = odr.create_close(
1133 apdu, Z_Close_protocolError,
1134 "unsupported APDU in filter_session_shared");
1136 package.session().close();
1139 m_p->release_frontend(package);
1142 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only)
1144 for (ptr = ptr->children; ptr; ptr = ptr->next)
1146 if (ptr->type != XML_ELEMENT_NODE)
1148 if (!strcmp((const char *) ptr->name, "resultset"))
1150 const struct _xmlAttr *attr;
1151 for (attr = ptr->properties; attr; attr = attr->next)
1153 if (!strcmp((const char *) attr->name, "ttl"))
1154 m_p->m_resultset_ttl =
1155 mp::xml::get_int(attr->children, 30);
1156 else if (!strcmp((const char *) attr->name, "max"))
1158 m_p->m_resultset_max =
1159 mp::xml::get_int(attr->children, 10);
1162 throw mp::filter::FilterException(
1163 "Bad attribute " + std::string((const char *)
1167 else if (!strcmp((const char *) ptr->name, "session"))
1169 const struct _xmlAttr *attr;
1170 for (attr = ptr->properties; attr; attr = attr->next)
1172 if (!strcmp((const char *) attr->name, "ttl"))
1173 m_p->m_session_ttl =
1174 mp::xml::get_int(attr->children, 120);
1176 throw mp::filter::FilterException(
1177 "Bad attribute " + std::string((const char *)
1183 throw mp::filter::FilterException("Bad element "
1184 + std::string((const char *)
1190 static mp::filter::Base* filter_creator()
1192 return new mp::filter::SessionShared;
1196 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1206 * c-file-style: "Stroustrup"
1207 * indent-tabs-mode: nil
1209 * vim: shiftwidth=4 tabstop=8 expandtab