1 /* $Id: filter_session_shared.cpp,v 1.16 2007-01-25 14:05:54 adam Exp $
2 Copyright (c) 2005-2007, Index Data.
4 See the LICENSE file for details
10 #include "package.hpp"
12 #include <boost/thread/mutex.hpp>
13 #include <boost/thread/condition.hpp>
14 #include <boost/thread/thread.hpp>
15 #include <boost/thread/xtime.hpp>
16 #include <boost/shared_ptr.hpp>
17 #include <boost/format.hpp>
20 #include "filter_session_shared.hpp"
24 #include <yaz/otherinfo.h>
25 #include <yaz/diagbib1.h>
26 #include <yazpp/z-query.h>
31 namespace mp = metaproxy_1;
32 namespace yf = metaproxy_1::filter;
34 namespace metaproxy_1 {
37 class SessionShared::InitKey {
39 bool operator < (const SessionShared::InitKey &k) const;
40 InitKey(Z_InitRequest *req);
41 InitKey(const InitKey &);
44 InitKey &operator = (const InitKey &k);
45 char *m_idAuthentication_buf;
46 int m_idAuthentication_size;
47 char *m_otherInfo_buf;
51 class SessionShared::Worker {
53 Worker(SessionShared::Rep *rep);
54 void operator() (void);
56 SessionShared::Rep *m_p;
58 class SessionShared::BackendSet {
60 std::string m_result_set_id;
61 Databases m_databases;
62 int m_result_set_size;
63 yazpp_1::Yaz_Z_Query m_query;
64 time_t m_time_last_use;
67 const std::string &result_set_id,
68 const Databases &databases,
69 const yazpp_1::Yaz_Z_Query &query);
71 Package &frontend_package,
72 const Z_APDU *apdu_req,
73 const BackendInstancePtr bp);
75 class SessionShared::BackendInstance {
77 friend class BackendClass;
78 friend class BackendSet;
80 mp::Session m_session;
81 BackendSetList m_sets;
84 int m_result_set_sequence;
85 time_t m_time_last_use;
86 mp::Package * m_close_package;
89 class SessionShared::BackendClass : boost::noncopyable {
91 friend struct Frontend;
92 bool m_named_result_sets;
93 BackendInstanceList m_backend_list;
94 BackendInstancePtr create_backend(const Package &package);
95 void remove_backend(BackendInstancePtr b);
96 BackendInstancePtr get_backend(const Package &package);
97 void use_backend(BackendInstancePtr b);
98 void release_backend(BackendInstancePtr b);
100 yazpp_1::GDU m_init_request;
101 yazpp_1::GDU m_init_response;
102 boost::mutex m_mutex_backend_class;
104 time_t m_backend_set_ttl;
105 time_t m_backend_expiry_ttl;
106 size_t m_backend_set_max;
108 BackendClass(const yazpp_1::GDU &init_request,
114 class SessionShared::FrontendSet {
115 Databases m_databases;
116 yazpp_1::Yaz_Z_Query m_query;
118 const Databases &get_databases();
119 const yazpp_1::Yaz_Z_Query &get_query();
121 const Databases &databases,
122 const yazpp_1::Yaz_Z_Query &query);
125 struct SessionShared::Frontend {
130 Z_Options m_init_options;
131 void search(Package &package, Z_APDU *apdu);
132 void present(Package &package, Z_APDU *apdu);
133 void scan(Package &package, Z_APDU *apdu);
135 void get_set(mp::Package &package,
136 const Z_APDU *apdu_req,
137 const Databases &databases,
138 yazpp_1::Yaz_Z_Query &query,
139 BackendInstancePtr &found_backend,
140 BackendSetPtr &found_set);
141 void override_set(BackendInstancePtr &found_backend,
142 std::string &result_set_id);
145 BackendClassPtr m_backend_class;
146 FrontendSets m_frontend_sets;
148 class SessionShared::Rep {
149 friend class SessionShared;
150 friend struct Frontend;
152 FrontendPtr get_frontend(Package &package);
153 void release_frontend(Package &package);
158 void init(Package &package, const Z_GDU *gdu,
159 FrontendPtr frontend);
160 boost::mutex m_mutex;
161 boost::condition m_cond_session_ready;
162 std::map<mp::Session, FrontendPtr> m_clients;
164 BackendClassMap m_backend_map;
165 boost::mutex m_mutex_backend_map;
166 boost::thread_group m_thrds;
174 yf::SessionShared::FrontendSet::FrontendSet(
175 const Databases &databases,
176 const yazpp_1::Yaz_Z_Query &query)
177 : m_databases(databases), m_query(query)
181 const yf::SessionShared::Databases &
182 yf::SessionShared::FrontendSet::get_databases()
187 const yazpp_1::Yaz_Z_Query& yf::SessionShared::FrontendSet::get_query()
192 yf::SessionShared::InitKey::InitKey(const InitKey &k)
194 m_odr = odr_createmem(ODR_ENCODE);
196 m_idAuthentication_size = k.m_idAuthentication_size;
197 m_idAuthentication_buf = (char*)odr_malloc(m_odr, m_idAuthentication_size);
198 memcpy(m_idAuthentication_buf, k.m_idAuthentication_buf,
199 m_idAuthentication_size);
201 m_otherInfo_size = k.m_otherInfo_size;
202 m_otherInfo_buf = (char*)odr_malloc(m_odr, m_otherInfo_size);
203 memcpy(m_otherInfo_buf, k.m_otherInfo_buf,
207 yf::SessionShared::InitKey::InitKey(Z_InitRequest *req)
209 m_odr = odr_createmem(ODR_ENCODE);
211 Z_IdAuthentication *t = req->idAuthentication;
212 z_IdAuthentication(m_odr, &t, 1, 0);
213 m_idAuthentication_buf =
214 odr_getbuf(m_odr, &m_idAuthentication_size, 0);
216 Z_OtherInformation *o = req->otherInfo;
217 z_OtherInformation(m_odr, &o, 1, 0);
218 m_otherInfo_buf = odr_getbuf(m_odr, &m_otherInfo_size, 0);
221 yf::SessionShared::InitKey::~InitKey()
226 bool yf::SessionShared::InitKey::operator < (const SessionShared::InitKey &k)
230 c = mp::util::memcmp2(
231 (void*) m_idAuthentication_buf, m_idAuthentication_size,
232 (void*) k.m_idAuthentication_buf, k.m_idAuthentication_size);
238 c = mp::util::memcmp2((void*) m_otherInfo_buf, m_otherInfo_size,
239 (void*) k.m_otherInfo_buf, k.m_otherInfo_size);
247 void yf::SessionShared::BackendClass::release_backend(BackendInstancePtr b)
249 boost::mutex::scoped_lock lock(m_mutex_backend_class);
254 void yf::SessionShared::BackendClass::remove_backend(BackendInstancePtr b)
256 BackendInstanceList::iterator it = m_backend_list.begin();
258 while (it != m_backend_list.end())
261 it = m_backend_list.erase(it);
269 yf::SessionShared::BackendInstancePtr
270 yf::SessionShared::BackendClass::get_backend(
271 const mp::Package &frontend_package)
274 boost::mutex::scoped_lock lock(m_mutex_backend_class);
276 BackendInstanceList::const_iterator it = m_backend_list.begin();
278 BackendInstancePtr backend1; // null
280 for (; it != m_backend_list.end(); it++)
282 if (!(*it)->m_in_use)
285 || (*it)->m_sequence_this < backend1->m_sequence_this)
291 use_backend(backend1);
295 return create_backend(frontend_package);
298 void yf::SessionShared::BackendClass::use_backend(BackendInstancePtr backend)
300 backend->m_in_use = true;
301 time(&backend->m_time_last_use);
302 backend->m_sequence_this = m_sequence_top++;
305 yf::SessionShared::BackendInstance::~BackendInstance()
307 delete m_close_package;
310 yf::SessionShared::BackendInstancePtr yf::SessionShared::BackendClass::create_backend(
311 const mp::Package &frontend_package)
313 BackendInstancePtr bp(new BackendInstance);
314 BackendInstancePtr null;
316 bp->m_close_package =
317 new mp::Package(bp->m_session, frontend_package.origin());
318 bp->m_close_package->copy_filter(frontend_package);
320 Package init_package(bp->m_session, frontend_package.origin());
322 init_package.copy_filter(frontend_package);
324 yazpp_1::GDU actual_init_request = m_init_request;
325 Z_GDU *init_pdu = actual_init_request.get();
327 assert(init_pdu->which == Z_GDU_Z3950);
328 assert(init_pdu->u.z3950->which == Z_APDU_initRequest);
330 Z_InitRequest *req = init_pdu->u.z3950->u.initRequest;
331 ODR_MASK_ZERO(req->options);
333 ODR_MASK_SET(req->options, Z_Options_search);
334 ODR_MASK_SET(req->options, Z_Options_present);
335 ODR_MASK_SET(req->options, Z_Options_namedResultSets);
336 ODR_MASK_SET(req->options, Z_Options_scan);
338 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_1);
339 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_2);
340 ODR_MASK_SET(req->protocolVersion, Z_ProtocolVersion_3);
342 init_package.request() = init_pdu;
346 boost::mutex::scoped_lock lock(m_mutex_backend_class);
348 m_named_result_sets = false;
349 Z_GDU *gdu = init_package.response().get();
350 if (!init_package.session().is_closed()
351 && gdu && gdu->which == Z_GDU_Z3950
352 && gdu->u.z3950->which == Z_APDU_initResponse)
354 Z_InitResponse *res = gdu->u.z3950->u.initResponse;
357 m_init_response = gdu->u.z3950;
358 if (ODR_MASK_GET(res->options, Z_Options_namedResultSets))
360 m_named_result_sets = true;
365 // did not receive an init response or closed
369 time(&bp->m_time_last_use);
370 bp->m_sequence_this = 0;
371 bp->m_result_set_sequence = 0;
372 m_backend_list.push_back(bp);
378 yf::SessionShared::BackendClass::BackendClass(const yazpp_1::GDU &init_request,
382 : m_named_result_sets(false), m_init_request(init_request),
383 m_sequence_top(0), m_backend_set_ttl(resultset_ttl),
384 m_backend_expiry_ttl(session_ttl), m_backend_set_max(resultset_max)
387 yf::SessionShared::BackendClass::~BackendClass()
390 void yf::SessionShared::Rep::init(mp::Package &package, const Z_GDU *gdu,
391 FrontendPtr frontend)
393 Z_InitRequest *req = gdu->u.z3950->u.initRequest;
395 frontend->m_is_virtual = true;
396 frontend->m_init_options = *req->options;
399 boost::mutex::scoped_lock lock(m_mutex_backend_map);
400 BackendClassMap::const_iterator it;
401 it = m_backend_map.find(k);
402 if (it == m_backend_map.end())
404 BackendClassPtr b(new BackendClass(gdu->u.z3950,
408 m_backend_map[k] = b;
409 frontend->m_backend_class = b;
410 std::cout << "SessionShared::Rep::init new session "
411 << frontend->m_backend_class << "\n";
415 frontend->m_backend_class = it->second;
416 std::cout << "SessionShared::Rep::init existing session "
417 << frontend->m_backend_class << "\n";
420 BackendClassPtr bc = frontend->m_backend_class;
421 BackendInstancePtr backend = bc->get_backend(package);
426 Z_APDU *apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
427 *apdu->u.initResponse->result = 0;
428 package.response() = apdu;
429 package.session().close();
433 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
434 yazpp_1::GDU init_response = bc->m_init_response;
435 Z_GDU *response_gdu = init_response.get();
436 mp::util::transfer_referenceId(odr, gdu->u.z3950,
437 response_gdu->u.z3950);
439 Z_Options *server_options =
440 response_gdu->u.z3950->u.initResponse->options;
441 Z_Options *client_options = &frontend->m_init_options;
444 for (i = 0; i<30; i++)
445 if (!ODR_MASK_GET(client_options, i))
446 ODR_MASK_CLEAR(server_options, i);
447 package.response() = init_response;
450 bc->release_backend(backend);
453 void yf::SessionShared::BackendSet::timestamp()
455 time(&m_time_last_use);
458 yf::SessionShared::BackendSet::BackendSet(
459 const std::string &result_set_id,
460 const Databases &databases,
461 const yazpp_1::Yaz_Z_Query &query) :
462 m_result_set_id(result_set_id),
463 m_databases(databases), m_result_set_size(0), m_query(query)
468 bool yf::SessionShared::BackendSet::search(
469 mp::Package &frontend_package,
470 const Z_APDU *frontend_apdu,
471 const BackendInstancePtr bp)
473 Package search_package(bp->m_session, frontend_package.origin());
475 search_package.copy_filter(frontend_package);
478 Z_APDU *apdu_req = zget_APDU(odr, Z_APDU_searchRequest);
479 Z_SearchRequest *req = apdu_req->u.searchRequest;
481 req->resultSetName = odr_strdup(odr, m_result_set_id.c_str());
482 req->query = m_query.get_Z_Query();
484 req->num_databaseNames = m_databases.size();
485 req->databaseNames = (char**)
486 odr_malloc(odr, req->num_databaseNames * sizeof(char *));
487 Databases::const_iterator it = m_databases.begin();
489 for (; it != m_databases.end(); it++)
490 req->databaseNames[i++] = odr_strdup(odr, it->c_str());
492 search_package.request() = apdu_req;
494 search_package.move();
496 Z_Records *z_records_diag = 0;
497 Z_GDU *gdu = search_package.response().get();
498 if (!search_package.session().is_closed()
499 && gdu && gdu->which == Z_GDU_Z3950
500 && gdu->u.z3950->which == Z_APDU_searchResponse)
502 Z_SearchResponse *b_resp = gdu->u.z3950->u.searchResponse;
505 if (b_resp->records->which == Z_Records_NSD
506 || b_resp->records->which == Z_Records_multipleNSD)
507 z_records_diag = b_resp->records;
511 if (frontend_apdu->which == Z_APDU_searchRequest)
513 Z_APDU *f_apdu = odr.create_searchResponse(frontend_apdu,
515 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
516 f_resp->records = z_records_diag;
517 frontend_package.response() = f_apdu;
520 if (frontend_apdu->which == Z_APDU_presentRequest)
522 Z_APDU *f_apdu = odr.create_presentResponse(frontend_apdu,
524 Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
525 f_resp->records = z_records_diag;
526 frontend_package.response() = f_apdu;
530 m_result_set_size = *b_resp->resultCount;
534 if (frontend_apdu->which == Z_APDU_searchRequest)
535 f_apdu = odr.create_searchResponse(
536 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
537 else if (frontend_apdu->which == Z_APDU_presentRequest)
538 f_apdu = odr.create_presentResponse(
539 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
541 f_apdu = odr.create_close(
542 frontend_apdu, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
543 frontend_package.response() = f_apdu;
547 void yf::SessionShared::Frontend::override_set(
548 BackendInstancePtr &found_backend,
549 std::string &result_set_id)
551 BackendClassPtr bc = m_backend_class;
552 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
556 for (; it != bc->m_backend_list.end(); it++)
558 if (!(*it)->m_in_use)
560 BackendSetList::iterator set_it = (*it)->m_sets.begin();
561 for (; set_it != (*it)->m_sets.end(); set_it++)
563 if (now >= (*set_it)->m_time_last_use &&
564 now - (*set_it)->m_time_last_use > bc->m_backend_set_ttl)
567 result_set_id = (*set_it)->m_result_set_id;
568 found_backend->m_sets.erase(set_it);
569 std::cout << "REUSE TTL SET: " << result_set_id << "\n";
575 size_t max_sets = bc->m_named_result_sets ? bc->m_backend_set_max : 1;
576 for (it = bc->m_backend_list.begin(); it != bc->m_backend_list.end(); it++)
578 if (!(*it)->m_in_use && (*it)->m_sets.size() < max_sets)
581 if (bc->m_named_result_sets)
583 result_set_id = boost::io::str(
584 boost::format("%1%") %
585 found_backend->m_result_set_sequence);
586 found_backend->m_result_set_sequence++;
589 result_set_id = "default";
590 std::cout << "AVAILABLE SET: " << result_set_id << "\n";
596 void yf::SessionShared::Frontend::get_set(mp::Package &package,
597 const Z_APDU *apdu_req,
598 const Databases &databases,
599 yazpp_1::Yaz_Z_Query &query,
600 BackendInstancePtr &found_backend,
601 BackendSetPtr &found_set)
603 std::string result_set_id;
604 BackendClassPtr bc = m_backend_class;
606 boost::mutex::scoped_lock lock(bc->m_mutex_backend_class);
608 // look at each backend and see if we have a similar search
609 BackendInstanceList::const_iterator it = bc->m_backend_list.begin();
611 for (; it != bc->m_backend_list.end(); it++)
613 if (!(*it)->m_in_use)
615 BackendSetList::const_iterator set_it = (*it)->m_sets.begin();
616 for (; set_it != (*it)->m_sets.end(); set_it++)
618 if ((*set_it)->m_databases == databases
619 && query.match(&(*set_it)->m_query))
623 bc->use_backend(found_backend);
624 found_set->timestamp();
625 std::cout << "MATCH SET: " <<
626 found_set->m_result_set_id << "\n";
627 // found matching set. No need to search again
633 override_set(found_backend, result_set_id);
635 bc->use_backend(found_backend);
639 // create a new backend set (and new set)
640 found_backend = bc->create_backend(package);
646 if (apdu_req->which == Z_APDU_searchRequest)
648 f_apdu = odr.create_searchResponse(
649 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
651 else if (apdu_req->which == Z_APDU_presentRequest)
653 f_apdu = odr.create_presentResponse(
654 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
658 f_apdu = odr.create_close(
659 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
661 package.response() = f_apdu;
664 std::cout << "NEW " << found_backend << "\n";
666 if (bc->m_named_result_sets)
668 result_set_id = boost::io::str(
669 boost::format("%1%") % found_backend->m_result_set_sequence);
670 found_backend->m_result_set_sequence++;
673 result_set_id = "default";
674 std::cout << "NEW SET: " << result_set_id << "\n";
676 // we must search ...
677 BackendSetPtr new_set(new BackendSet(result_set_id,
679 if (!new_set->search(package, apdu_req, found_backend))
681 std::cout << "search error\n";
682 bc->remove_backend(found_backend);
683 return; // search error
686 found_set->timestamp();
687 found_backend->m_sets.push_back(found_set);
690 void yf::SessionShared::Frontend::search(mp::Package &package,
693 Z_SearchRequest *req = apdu_req->u.searchRequest;
694 FrontendSets::iterator fset_it =
695 m_frontend_sets.find(req->resultSetName);
696 if (fset_it != m_frontend_sets.end())
698 // result set already exist
699 // if replace indicator is off: we return diagnostic if
700 // result set already exist.
701 if (*req->replaceIndicator == 0)
705 odr.create_searchResponse(
707 YAZ_BIB1_RESULT_SET_EXISTS_AND_REPLACE_INDICATOR_OFF,
709 package.response() = apdu;
713 m_frontend_sets.erase(fset_it);
716 yazpp_1::Yaz_Z_Query query;
717 query.set_Z_Query(req->query);
720 for (i = 0; i<req->num_databaseNames; i++)
721 databases.push_back(req->databaseNames[i]);
723 BackendSetPtr found_set; // null
724 BackendInstancePtr found_backend; // null
726 get_set(package, apdu_req, databases, query, found_backend, found_set);
731 Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
732 Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
733 *f_resp->resultCount = found_set->m_result_set_size;
734 package.response() = f_apdu;
736 FrontendSetPtr fset(new FrontendSet(databases, query));
737 m_frontend_sets[req->resultSetName] = fset;
739 m_backend_class->release_backend(found_backend);
742 void yf::SessionShared::Frontend::present(mp::Package &package,
746 Z_PresentRequest *req = apdu_req->u.presentRequest;
748 FrontendSets::iterator fset_it =
749 m_frontend_sets.find(req->resultSetId);
751 if (fset_it == m_frontend_sets.end())
754 odr.create_presentResponse(
756 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
758 package.response() = apdu;
761 FrontendSetPtr fset = fset_it->second;
763 Databases databases = fset->get_databases();
764 yazpp_1::Yaz_Z_Query query = fset->get_query();
766 BackendClassPtr bc = m_backend_class;
767 BackendSetPtr found_set; // null
768 BackendInstancePtr found_backend;
770 get_set(package, apdu_req, databases, query, found_backend, found_set);
774 Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
775 Z_PresentRequest *p_req = p_apdu->u.presentRequest;
776 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
777 p_req->resultSetId = odr_strdup(odr, found_set->m_result_set_id.c_str());
778 *p_req->resultSetStartPoint = *req->resultSetStartPoint;
779 *p_req->numberOfRecordsRequested = *req->numberOfRecordsRequested;
780 p_req->preferredRecordSyntax = req->preferredRecordSyntax;
781 p_req->recordComposition = req->recordComposition;
783 Package present_package(found_backend->m_session, package.origin());
784 present_package.copy_filter(package);
786 present_package.request() = p_apdu;
788 present_package.move();
790 Z_GDU *gdu = present_package.response().get();
791 if (!present_package.session().is_closed()
792 && gdu && gdu->which == Z_GDU_Z3950
793 && gdu->u.z3950->which == Z_APDU_presentResponse)
795 Z_PresentResponse *b_resp = gdu->u.z3950->u.presentResponse;
796 Z_APDU *f_apdu_res = odr.create_presentResponse(apdu_req, 0, 0);
797 Z_PresentResponse *f_resp = f_apdu_res->u.presentResponse;
799 f_resp->numberOfRecordsReturned = b_resp->numberOfRecordsReturned;
800 f_resp->nextResultSetPosition = b_resp->nextResultSetPosition;
801 f_resp->presentStatus= b_resp->presentStatus;
802 f_resp->records = b_resp->records;
803 f_resp->otherInfo = b_resp->otherInfo;
804 package.response() = f_apdu_res;
805 bc->release_backend(found_backend);
809 bc->remove_backend(found_backend);
811 odr.create_presentResponse(
812 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
813 package.response() = f_apdu_res;
817 void yf::SessionShared::Frontend::scan(mp::Package &frontend_package,
820 BackendClassPtr bc = m_backend_class;
821 BackendInstancePtr backend = bc->get_backend(frontend_package);
825 Z_APDU *apdu = odr.create_scanResponse(
826 apdu_req, YAZ_BIB1_TEMPORARY_SYSTEM_ERROR, 0);
827 frontend_package.response() = apdu;
831 Package scan_package(backend->m_session, frontend_package.origin());
832 scan_package.copy_filter(frontend_package);
833 scan_package.request() = apdu_req;
835 frontend_package.response() = scan_package.response();
836 if (scan_package.session().is_closed())
838 frontend_package.session().close();
839 bc->remove_backend(backend);
842 bc->release_backend(backend);
846 yf::SessionShared::Worker::Worker(SessionShared::Rep *rep) : m_p(rep)
850 void yf::SessionShared::Worker::operator() (void)
855 void yf::SessionShared::BackendClass::expire()
859 boost::mutex::scoped_lock lock(m_mutex_backend_class);
860 BackendInstanceList::iterator bit = m_backend_list.begin();
861 while (bit != m_backend_list.end())
863 std::cout << "expiry ";
864 time_t last_use = (*bit)->m_time_last_use;
866 if ((*bit)->m_in_use)
868 std::cout << "inuse";
871 else if ((now >= last_use && now - last_use > m_backend_expiry_ttl)
875 (*bit)->m_close_package->response() = odr.create_close(
876 0, Z_Close_lackOfActivity, 0);
877 (*bit)->m_close_package->session().close();
878 (*bit)->m_close_package->move();
880 bit = m_backend_list.erase(bit);
881 std::cout << "erase";
888 std::cout << std::endl;
892 void yf::SessionShared::Rep::expire()
897 boost::xtime_get(&xt, boost::TIME_UTC);
899 boost::thread::sleep(xt);
900 std::cout << "." << std::endl;
902 BackendClassMap::const_iterator b_it = m_backend_map.begin();
903 for (; b_it != m_backend_map.end(); b_it++)
904 b_it->second->expire();
908 yf::SessionShared::Rep::Rep()
910 m_resultset_ttl = 30;
911 m_resultset_max = 10;
913 yf::SessionShared::Worker w(this);
914 m_thrds.add_thread(new boost::thread(w));
917 yf::SessionShared::SessionShared() : m_p(new SessionShared::Rep)
921 yf::SessionShared::~SessionShared() {
925 yf::SessionShared::Frontend::Frontend(Rep *rep) : m_is_virtual(false), m_p(rep)
929 yf::SessionShared::Frontend::~Frontend()
933 yf::SessionShared::FrontendPtr yf::SessionShared::Rep::get_frontend(mp::Package &package)
935 boost::mutex::scoped_lock lock(m_mutex);
937 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
941 it = m_clients.find(package.session());
942 if (it == m_clients.end())
945 if (!it->second->m_in_use)
947 it->second->m_in_use = true;
950 m_cond_session_ready.wait(lock);
952 FrontendPtr f(new Frontend(this));
953 m_clients[package.session()] = f;
958 void yf::SessionShared::Rep::release_frontend(mp::Package &package)
960 boost::mutex::scoped_lock lock(m_mutex);
961 std::map<mp::Session,yf::SessionShared::FrontendPtr>::iterator it;
963 it = m_clients.find(package.session());
964 if (it != m_clients.end())
966 if (package.session().is_closed())
972 it->second->m_in_use = false;
974 m_cond_session_ready.notify_all();
979 void yf::SessionShared::process(mp::Package &package) const
981 FrontendPtr f = m_p->get_frontend(package);
983 Z_GDU *gdu = package.request().get();
985 if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
986 Z_APDU_initRequest && !f->m_is_virtual)
988 m_p->init(package, gdu, f);
990 else if (!f->m_is_virtual)
992 else if (gdu && gdu->which == Z_GDU_Z3950)
994 Z_APDU *apdu = gdu->u.z3950;
995 if (apdu->which == Z_APDU_initRequest)
999 package.response() = odr.create_close(
1001 Z_Close_protocolError,
1004 package.session().close();
1006 else if (apdu->which == Z_APDU_close)
1010 package.response() = odr.create_close(
1012 Z_Close_peerAbort, "received close from client");
1013 package.session().close();
1015 else if (apdu->which == Z_APDU_searchRequest)
1017 f->search(package, apdu);
1019 else if (apdu->which == Z_APDU_presentRequest)
1021 f->present(package, apdu);
1023 else if (apdu->which == Z_APDU_scanRequest)
1025 f->scan(package, apdu);
1031 package.response() = odr.create_close(
1032 apdu, Z_Close_protocolError,
1033 "unsupported APDU in filter_session_shared");
1035 package.session().close();
1038 m_p->release_frontend(package);
1041 void yf::SessionShared::configure(const xmlNode *ptr)
1043 for (ptr = ptr->children; ptr; ptr = ptr->next)
1045 if (ptr->type != XML_ELEMENT_NODE)
1047 if (!strcmp((const char *) ptr->name, "resultset"))
1049 const struct _xmlAttr *attr;
1050 for (attr = ptr->properties; attr; attr = attr->next)
1052 if (!strcmp((const char *) attr->name, "ttl"))
1053 m_p->m_resultset_ttl =
1054 mp::xml::get_int(attr->children, 30);
1055 else if (!strcmp((const char *) attr->name, "max"))
1057 m_p->m_resultset_max =
1058 mp::xml::get_int(attr->children, 10);
1061 throw mp::filter::FilterException(
1062 "Bad attribute " + std::string((const char *)
1066 else if (!strcmp((const char *) ptr->name, "session"))
1068 const struct _xmlAttr *attr;
1069 for (attr = ptr->properties; attr; attr = attr->next)
1071 if (!strcmp((const char *) attr->name, "ttl"))
1072 m_p->m_session_ttl =
1073 mp::xml::get_int(attr->children, 120);
1075 throw mp::filter::FilterException(
1076 "Bad attribute " + std::string((const char *)
1082 throw mp::filter::FilterException("Bad element "
1083 + std::string((const char *)
1089 static mp::filter::Base* filter_creator()
1091 return new mp::filter::SessionShared;
1095 struct metaproxy_1_filter_struct metaproxy_1_filter_session_shared = {
1105 * indent-tabs-mode: nil
1106 * c-file-style: "stroustrup"
1108 * vim: shiftwidth=4 tabstop=8 expandtab