1 /* This file is part of Metaproxy.
2 Copyright (C) 2005-2008 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
22 #include "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>
32 #include "filter_session_shared.hpp"
36 #include <yaz/otherinfo.h>
37 #include <yaz/diagbib1.h>
38 #include <yazpp/z-query.h>
43 namespace mp = metaproxy_1;
44 namespace yf = metaproxy_1::filter;
46 namespace metaproxy_1 {
49 // key for session.. We'll only share sessions with same InitKey
50 class SessionShared::InitKey {
52 bool operator < (const SessionShared::InitKey &k) const;
53 InitKey(Z_InitRequest *req);
54 InitKey(const InitKey &);
57 char *m_idAuthentication_buf;
58 int m_idAuthentication_size;
59 char *m_otherInfo_buf;
63 // worker thread .. for expiry of sessions
64 class SessionShared::Worker {
66 Worker(SessionShared::Rep *rep);
67 void operator() (void);
69 SessionShared::Rep *m_p;
72 class SessionShared::BackendSet {
74 std::string m_result_set_id;
75 Databases m_databases;
76 int m_result_set_size;
77 yazpp_1::Yaz_Z_Query m_query;
78 time_t m_time_last_use;
81 const std::string &result_set_id,
82 const Databases &databases,
83 const yazpp_1::Yaz_Z_Query &query);
85 Package &frontend_package,
86 const Z_APDU *apdu_req,
87 const BackendInstancePtr bp,
90 // backend connection instance
91 class SessionShared::BackendInstance {
93 friend class BackendClass;
94 friend class BackendSet;
96 mp::Session m_session;
97 BackendSetList m_sets;
100 int m_result_set_sequence;
101 time_t m_time_last_use;
102 mp::Package * m_close_package;
105 // backends of some class (all with same InitKey)
106 class SessionShared::BackendClass : boost::noncopyable {
108 friend struct Frontend;
109 bool m_named_result_sets;
110 BackendInstanceList m_backend_list;
111 BackendInstancePtr create_backend(const Package &package);
112 void remove_backend(BackendInstancePtr b);
113 BackendInstancePtr get_backend(const Package &package);
114 void use_backend(BackendInstancePtr b);
115 void release_backend(BackendInstancePtr b);
117 yazpp_1::GDU m_init_request;
118 yazpp_1::GDU m_init_response;
119 boost::mutex m_mutex_backend_class;
121 time_t m_backend_set_ttl;
122 time_t m_backend_expiry_ttl;
123 size_t m_backend_set_max;
125 BackendClass(const yazpp_1::GDU &init_request,
131 // frontend result set
132 class SessionShared::FrontendSet {
133 Databases m_databases;
134 yazpp_1::Yaz_Z_Query m_query;
136 const Databases &get_databases();
137 const yazpp_1::Yaz_Z_Query &get_query();
139 const Databases &databases,
140 const yazpp_1::Yaz_Z_Query &query);
144 struct SessionShared::Frontend {
149 Z_Options m_init_options;
150 void search(Package &package, Z_APDU *apdu);
151 void present(Package &package, Z_APDU *apdu);
152 void scan(Package &package, Z_APDU *apdu);
154 void get_set(mp::Package &package,
155 const Z_APDU *apdu_req,
156 const Databases &databases,
157 yazpp_1::Yaz_Z_Query &query,
158 BackendInstancePtr &found_backend,
159 BackendSetPtr &found_set);
160 void override_set(BackendInstancePtr &found_backend,
161 std::string &result_set_id);
164 BackendClassPtr m_backend_class;
165 FrontendSets m_frontend_sets;
168 class SessionShared::Rep {
169 friend class SessionShared;
170 friend struct Frontend;
172 FrontendPtr get_frontend(Package &package);
173 void release_frontend(Package &package);
178 void init(Package &package, const Z_GDU *gdu,
179 FrontendPtr frontend);
180 boost::mutex m_mutex;
181 boost::condition m_cond_session_ready;
182 std::map<mp::Session, FrontendPtr> m_clients;
184 BackendClassMap m_backend_map;
185 boost::mutex m_mutex_backend_map;
186 boost::thread_group m_thrds;
194 yf::SessionShared::FrontendSet::FrontendSet(
195 const Databases &databases,
196 const yazpp_1::Yaz_Z_Query &query)
197 : m_databases(databases), m_query(query)
201 const yf::SessionShared::Databases &
202 yf::SessionShared::FrontendSet::get_databases()
207 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
212 yf::SessionShared::InitKey::InitKey(const InitKey &k)
214 m_odr = odr_createmem(ODR_ENCODE);
216 m_idAuthentication_size = k.m_idAuthentication_size;
217 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
218 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
219 m_idAuthentication_size);
221 m_otherInfo_size = k.m_otherInfo_size;
222 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
223 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
227 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
229 m_odr = odr_createmem(ODR_ENCODE);
231 Z_IdAuthentication *t = req->idAuthentication;
232 z_IdAuthentication(m_odr, &t, 1, 0);
233 m_idAuthentication_buf =
234 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
236 Z_OtherInformation *o = req->otherInfo;
237 z_OtherInformation(m_odr, &o, 1, 0);
238 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
241 yf::SessionShared::InitKey::~InitKey()
246 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
250 c = mp::util::memcmp2(
251 (void*) m_idAuthentication_buf, m_idAuthentication_size,
252 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
258 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
259 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
267 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
269 boost::mutex::scoped_lock lock(m_mutex_backend_class);
274 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
276 BackendInstanceList::iterator it = m_backend_list.begin();
278 while (it != m_backend_list.end())
283 (*it)->m_close_package->response() = odr.create_close(
284 0, Z_Close_lackOfActivity, 0);
285 (*it)->m_close_package->session().close();
286 (*it)->m_close_package->move();
288 it = m_backend_list.erase(it);
297 yf::SessionShared::BackendInstancePtr
298 yf::SessionShared::BackendClass::get_backend(
299 const mp::Package &frontend_package)
302 boost::mutex::scoped_lock lock(m_mutex_backend_class);
304 BackendInstanceList::const_iterator it = m_backend_list.begin();
306 BackendInstancePtr backend1; // null
308 for (; it != m_backend_list.end(); it++)
310 if (!(*it)->m_in_use)
313 || (*it)->m_sequence_this < backend1->m_sequence_this)
319 use_backend(backend1);
323 return create_backend(frontend_package);
326 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
328 backend->m_in_use = true;
329 time(&backend->m_time_last_use);
330 backend->m_sequence_this = m_sequence_top++;
333 yf::SessionShared::BackendInstance::~BackendInstance()
335 delete m_close_package;
338 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
339 const mp::Package &frontend_package)
341 BackendInstancePtr bp(new BackendInstance);
342 BackendInstancePtr null;
344 bp->m_close_package =
345 new mp::Package(bp->m_session, frontend_package.origin());
346 bp->m_close_package->copy_filter(frontend_package);
348 Package init_package(bp->m_session, frontend_package.origin());
350 init_package.copy_filter(frontend_package);
352 yazpp_1::GDU actual_init_request = m_init_request;
353 Z_GDU *init_pdu = actual_init_request.get();
355 assert(init_pdu->which == Z_GDU_Z3950);
356 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
358 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
359 ODR_MASK_ZERO(req->options);
361 ODR_MASK_SET(req->options, Z_Options_search);
362 ODR_MASK_SET(req->options, Z_Options_present);
363 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
364 ODR_MASK_SET(req->options, Z_Options_scan);
366 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
367 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
368 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
370 init_package.request() = init_pdu;
374 boost::mutex::scoped_lock lock(m_mutex_backend_class);
376 m_named_result_sets = false;
377 Z_GDU *gdu = init_package.response().get();
378 if (!init_package.session().is_closed()
379 && gdu && gdu->which == Z_GDU_Z3950
380 && gdu->u.z3950->which == Z_APDU_initResponse)
382 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
385 m_init_response = gdu->u.z3950;
386 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
388 m_named_result_sets = true;
393 // did not receive an init response or closed
397 time(&bp->m_time_last_use);
398 bp->m_sequence_this = 0;
399 bp->m_result_set_sequence = 0;
400 m_backend_list.push_back(bp);
406 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
410 : m_named_result_sets(false), m_init_request(init_request),
411 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
412 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
415 yf::SessionShared::BackendClass::~BackendClass()
418 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
419 FrontendPtr frontend)
421 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
423 frontend->m_is_virtual = true;
424 frontend->m_init_options = *req->options;
427 boost::mutex::scoped_lock lock(m_mutex_backend_map);
428 BackendClassMap::const_iterator it;
429 it = m_backend_map.find(k);
430 if (it == m_backend_map.end())
432 BackendClassPtr b(new BackendClass(gdu->u.z3950,
436 m_backend_map[k] = b;
437 frontend->m_backend_class = b;
441 frontend->m_backend_class = it->second;
444 BackendClassPtr bc = frontend->m_backend_class;
445 BackendInstancePtr backend = bc->get_backend(package);
450 Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
451 *apdu->u.initResponse->result = 0;
452 package.response() = apdu;
453 package.session().close();
457 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
458 yazpp_1::GDU init_response = bc->m_init_response;
459 Z_GDU *response_gdu = init_response.get();
460 mp::util::transfer_referenceId(odr, gdu->u.z3950,
461 response_gdu->u.z3950);
463 Z_Options *server_options =
464 response_gdu->u.z3950->u.initResponse->options;
465 Z_Options *client_options = &frontend->m_init_options;
468 for (i = 0; i<30; i++)
469 if (!ODR_MASK_GET(client_options, i))
470 ODR_MASK_CLEAR(server_options, i);
471 package.response() = init_response;
474 bc->release_backend(backend);
477 void yf::SessionShared::BackendSet::timestamp()
479 time(&m_time_last_use);
482 yf::SessionShared::BackendSet::BackendSet(
483 const std::string &result_set_id,
484 const Databases &databases,
485 const yazpp_1::Yaz_Z_Query &query) :
486 m_result_set_id(result_set_id),
487 m_databases(databases), m_result_set_size(0), m_query(query)
492 bool yf::SessionShared::BackendSet::search(
493 mp::Package &frontend_package,
494 const Z_APDU *frontend_apdu,
495 const BackendInstancePtr bp,
498 Package search_package(bp->m_session, frontend_package.origin());
500 search_package.copy_filter(frontend_package);
503 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
504 Z_SearchRequest *req = apdu_req->u.searchRequest;
506 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
507 req->query = m_query.get_Z_Query();
509 req->num_databaseNames = m_databases.size();
510 req->databaseNames = (char**)
511 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
512 Databases::const_iterator it = m_databases.begin();
514 for (; it != m_databases.end(); it++)
515 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
517 search_package.request() = apdu_req;
519 search_package.move();
520 fatal_error = false; // assume backend session is good
522 Z_Records *z_records_diag = 0;
523 Z_GDU *gdu = search_package.response().get();
524 if (!search_package.session().is_closed()
525 && gdu && gdu->which == Z_GDU_Z3950
526 && gdu->u.z3950->which == Z_APDU_searchResponse)
528 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
531 if (b_resp->records->which == Z_Records_NSD
532 || b_resp->records->which == Z_Records_multipleNSD)
533 z_records_diag = b_resp->records;
537 // there could be diagnostics that are so bad.. that
538 // we simply mark the error as fatal.. For now we assume
540 if (frontend_apdu->which == Z_APDU_searchRequest)
542 Z_APDU *f_apdu = odr.create_searchResponse(frontend_apdu,
544 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
545 *f_resp->searchStatus = *b_resp->searchStatus;
546 f_resp->records = z_records_diag;
547 frontend_package.response() = f_apdu;
550 if (frontend_apdu->which == Z_APDU_presentRequest)
552 Z_APDU *f_apdu = odr.create_presentResponse(frontend_apdu,
554 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
555 f_resp->records = z_records_diag;
556 frontend_package.response() = f_apdu;
560 m_result_set_size = *b_resp->resultCount;
564 if (frontend_apdu->which == Z_APDU_searchRequest)
565 f_apdu = odr.create_searchResponse(
566 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
567 else if (frontend_apdu->which == Z_APDU_presentRequest)
568 f_apdu = odr.create_presentResponse(
569 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
571 f_apdu = odr.create_close(
572 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
573 frontend_package.response() = f_apdu;
574 fatal_error = true; // weired response.. bad backend
578 void yf::SessionShared::Frontend::override_set(
579 BackendInstancePtr &found_backend,
580 std::string &result_set_id)
582 BackendClassPtr bc = m_backend_class;
583 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
587 for (; it != bc->m_backend_list.end(); it++)
589 if (!(*it)->m_in_use)
591 BackendSetList::iterator set_it = (*it)->m_sets.begin();
592 for (; set_it != (*it)->m_sets.end(); set_it++)
594 if (now >= (*set_it)->m_time_last_use &&
595 now - (*set_it)->m_time_last_use > bc->m_backend_set_ttl)
598 result_set_id = (*set_it)->m_result_set_id;
599 found_backend->m_sets.erase(set_it);
605 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
606 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
608 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
611 if (bc->m_named_result_sets)
613 result_set_id = boost::io::str(
614 boost::format("%1%") %
615 found_backend->m_result_set_sequence);
616 found_backend->m_result_set_sequence++;
619 result_set_id = "default";
625 void yf::SessionShared::Frontend::get_set(mp::Package &package,
626 const Z_APDU *apdu_req,
627 const Databases &databases,
628 yazpp_1::Yaz_Z_Query &query,
629 BackendInstancePtr &found_backend,
630 BackendSetPtr &found_set)
632 bool session_restarted = false;
635 std::string result_set_id;
636 BackendClassPtr bc = m_backend_class;
638 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
640 // look at each backend and see if we have a similar search
641 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
643 for (; it != bc->m_backend_list.end(); it++)
645 if (!(*it)->m_in_use)
647 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
648 for (; set_it != (*it)->m_sets.end(); set_it++)
650 if ((*set_it)->m_databases == databases
651 && query.match(&(*set_it)->m_query))
655 bc->use_backend(found_backend);
656 found_set->timestamp();
657 // found matching set. No need to search again
663 override_set(found_backend, result_set_id);
665 bc->use_backend(found_backend);
669 // create a new backend set (and new set)
670 found_backend = bc->create_backend(package);
676 if (apdu_req->which == Z_APDU_searchRequest)
678 f_apdu = odr.create_searchResponse(
679 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
681 else if (apdu_req->which == Z_APDU_presentRequest)
683 f_apdu = odr.create_presentResponse(
684 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
688 f_apdu = odr.create_close(
689 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
691 package.response() = f_apdu;
694 if (bc->m_named_result_sets)
696 result_set_id = boost::io::str(
697 boost::format("%1%") % found_backend->m_result_set_sequence);
698 found_backend->m_result_set_sequence++;
701 result_set_id = "default";
703 // we must search ...
704 BackendSetPtr new_set(new BackendSet(result_set_id,
706 bool fatal_error = false;
707 if (!new_set->search(package, apdu_req, found_backend, fatal_error))
710 bc->remove_backend(found_backend);
712 bc->release_backend(found_backend);
713 return; // search error
715 if (!session_restarted && new_set->m_result_set_size < 0)
717 bc->remove_backend(found_backend);
718 session_restarted = true;
719 found_backend.reset();
724 found_set->timestamp();
725 found_backend->m_sets.push_back(found_set);
728 void yf::SessionShared::Frontend::search(mp::Package &package,
731 Z_SearchRequest *req = apdu_req->u.searchRequest;
732 FrontendSets::iterator fset_it =
733 m_frontend_sets.find(req->resultSetName);
734 if (fset_it != m_frontend_sets.end())
736 // result set already exist
737 // if replace indicator is off: we return diagnostic if
738 // result set already exist.
739 if (*req->replaceIndicator == 0)
743 odr.create_searchResponse(
745 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
747 package.response() = apdu;
751 m_frontend_sets.erase(fset_it);
754 yazpp_1::Yaz_Z_Query query;
755 query.set_Z_Query(req->query);
758 for (i = 0; i<req->num_databaseNames; i++)
759 databases.push_back(req->databaseNames[i]);
761 BackendSetPtr found_set; // null
762 BackendInstancePtr found_backend; // null
764 get_set(package, apdu_req, databases, query, found_backend, found_set);
769 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
770 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
771 *f_resp->resultCount = found_set->m_result_set_size;
772 package.response() = f_apdu;
774 FrontendSetPtr fset(new FrontendSet(databases, query));
775 m_frontend_sets[req->resultSetName] = fset;
777 m_backend_class->release_backend(found_backend);
780 void yf::SessionShared::Frontend::present(mp::Package &package,
784 Z_PresentRequest *req = apdu_req->u.presentRequest;
786 FrontendSets::iterator fset_it =
787 m_frontend_sets.find(req->resultSetId);
789 if (fset_it == m_frontend_sets.end())
792 odr.create_presentResponse(
794 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
796 package.response() = apdu;
799 FrontendSetPtr fset = fset_it->second;
801 Databases databases = fset->get_databases();
802 yazpp_1::Yaz_Z_Query query = fset->get_query();
804 BackendClassPtr bc = m_backend_class;
805 BackendSetPtr found_set; // null
806 BackendInstancePtr found_backend;
808 get_set(package, apdu_req, databases, query, found_backend, found_set);
812 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
813 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
814 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
815 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
816 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
817 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
818 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
819 p_req->recordComposition = req->recordComposition;
821 Package present_package(found_backend->m_session, package.origin());
822 present_package.copy_filter(package);
824 present_package.request() = p_apdu;
826 present_package.move();
828 Z_GDU *gdu = present_package.response().get();
829 if (!present_package.session().is_closed()
830 && gdu && gdu->which == Z_GDU_Z3950
831 && gdu->u.z3950->which == Z_APDU_presentResponse)
833 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
834 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
835 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
837 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
838 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
839 f_resp->presentStatus= b_resp->presentStatus;
840 f_resp->records = b_resp->records;
841 f_resp->otherInfo = b_resp->otherInfo;
842 package.response() = f_apdu_res;
843 bc->release_backend(found_backend);
847 bc->remove_backend(found_backend);
849 odr.create_presentResponse(
850 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
851 package.response() = f_apdu_res;
855 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
858 BackendClassPtr bc = m_backend_class;
859 BackendInstancePtr backend = bc->get_backend(frontend_package);
863 Z_APDU *apdu = odr.create_scanResponse(
864 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
865 frontend_package.response() = apdu;
869 Package scan_package(backend->m_session, frontend_package.origin());
870 scan_package.copy_filter(frontend_package);
871 scan_package.request() = apdu_req;
873 frontend_package.response() = scan_package.response();
874 if (scan_package.session().is_closed())
876 frontend_package.session().close();
877 bc->remove_backend(backend);
880 bc->release_backend(backend);
884 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
888 void yf::SessionShared::Worker::operator() (void)
893 void yf::SessionShared::BackendClass::expire_class()
897 boost::mutex::scoped_lock lock(m_mutex_backend_class);
898 BackendInstanceList::iterator bit = m_backend_list.begin();
899 while (bit != m_backend_list.end())
901 time_t last_use = (*bit)->m_time_last_use;
903 if ((*bit)->m_in_use)
907 else if ((now >= last_use && now - last_use > m_backend_expiry_ttl)
911 (*bit)->m_close_package->response() = odr.create_close(
912 0, Z_Close_lackOfActivity, 0);
913 (*bit)->m_close_package->session().close();
914 (*bit)->m_close_package->move();
916 bit = m_backend_list.erase(bit);
925 void yf::SessionShared::Rep::expire()
930 boost::xtime_get(&xt, boost::TIME_UTC);
932 boost::thread::sleep(xt);
934 BackendClassMap::const_iterator b_it = m_backend_map.begin();
935 for (; b_it != m_backend_map.end(); b_it++)
936 b_it->second->expire_class();
940 yf::SessionShared::Rep::Rep()
942 m_resultset_ttl = 30;
943 m_resultset_max = 10;
945 yf::SessionShared::Worker w(this);
946 m_thrds.add_thread(new boost::thread(w));
949 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
953 yf::SessionShared::~SessionShared() {
957 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
961 yf::SessionShared::Frontend::~Frontend()
965 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
967 boost::mutex::scoped_lock lock(m_mutex);
969 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
973 it = m_clients.find(package.session());
974 if (it == m_clients.end())
977 if (!it->second->m_in_use)
979 it->second->m_in_use = true;
982 m_cond_session_ready.wait(lock);
984 FrontendPtr f(new Frontend(this));
985 m_clients[package.session()] = f;
990 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
992 boost::mutex::scoped_lock lock(m_mutex);
993 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
995 it = m_clients.find(package.session());
996 if (it != m_clients.end())
998 if (package.session().is_closed())
1000 m_clients.erase(it);
1004 it->second->m_in_use = false;
1006 m_cond_session_ready.notify_all();
1011 void yf::SessionShared::process(mp::Package &package) const
1013 FrontendPtr f = m_p->get_frontend(package);
1015 Z_GDU *gdu = package.request().get();
1017 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1018 Z_APDU_initRequest && !f->m_is_virtual)
1020 m_p->init(package, gdu, f);
1022 else if (!f->m_is_virtual)
1024 else if (gdu && gdu->which == Z_GDU_Z3950)
1026 Z_APDU *apdu = gdu->u.z3950;
1027 if (apdu->which == Z_APDU_initRequest)
1031 package.response() = odr.create_close(
1033 Z_Close_protocolError,
1036 package.session().close();
1038 else if (apdu->which == Z_APDU_close)
1042 package.response() = odr.create_close(
1044 Z_Close_peerAbort, "received close from client");
1045 package.session().close();
1047 else if (apdu->which == Z_APDU_searchRequest)
1049 f->search(package, apdu);
1051 else if (apdu->which == Z_APDU_presentRequest)
1053 f->present(package, apdu);
1055 else if (apdu->which == Z_APDU_scanRequest)
1057 f->scan(package, apdu);
1063 package.response() = odr.create_close(
1064 apdu, Z_Close_protocolError,
1065 "unsupported APDU in filter_session_shared");
1067 package.session().close();
1070 m_p->release_frontend(package);
1073 void yf::SessionShared::configure(const xmlNode *ptr, bool test_only)
1075 for (ptr = ptr->children; ptr; ptr = ptr->next)
1077 if (ptr->type != XML_ELEMENT_NODE)
1079 if (!strcmp((const char *) ptr->name, "resultset"))
1081 const struct _xmlAttr *attr;
1082 for (attr = ptr->properties; attr; attr = attr->next)
1084 if (!strcmp((const char *) attr->name, "ttl"))
1085 m_p->m_resultset_ttl =
1086 mp::xml::get_int(attr->children, 30);
1087 else if (!strcmp((const char *) attr->name, "max"))
1089 m_p->m_resultset_max =
1090 mp::xml::get_int(attr->children, 10);
1093 throw mp::filter::FilterException(
1094 "Bad attribute " + std::string((const char *)
1098 else if (!strcmp((const char *) ptr->name, "session"))
1100 const struct _xmlAttr *attr;
1101 for (attr = ptr->properties; attr; attr = attr->next)
1103 if (!strcmp((const char *) attr->name, "ttl"))
1104 m_p->m_session_ttl =
1105 mp::xml::get_int(attr->children, 120);
1107 throw mp::filter::FilterException(
1108 "Bad attribute " + std::string((const char *)
1114 throw mp::filter::FilterException("Bad element "
1115 + std::string((const char *)
1121 static mp::filter::Base* filter_creator()
1123 return new mp::filter::SessionShared;
1127 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1137 * c-file-style: "Stroustrup"
1138 * indent-tabs-mode: nil
1140 * vim: shiftwidth=4 tabstop=8 expandtab