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/util.hpp>
23 #include <metaproxy/filter.hpp>
24 #include <metaproxy/package.hpp>
25 #include "thread_pool_observer.hpp"
26 #include "filter_frontend_net.hpp"
27 #include <yazpp/z-assoc.h>
28 #include <yazpp/pdu-assoc.h>
29 #include <yazpp/socket-manager.h>
30 #include <yazpp/limit-connect.h>
35 namespace mp = metaproxy_1;
37 namespace metaproxy_1 {
38 class My_Timer_Thread;
41 class FrontendNet::Port {
43 friend class FrontendNet;
47 class FrontendNet::Rep {
48 friend class FrontendNet;
50 std::vector<Port> m_ports;
51 int m_listen_duration;
52 int m_session_timeout;
54 yazpp_1::SocketManager mySocketManager;
58 class My_Timer_Thread : public yazpp_1::ISocketObserver {
60 yazpp_1::ISocketObservable *m_obs;
64 My_Timer_Thread(yazpp_1::ISocketObservable *obs, int duration);
65 void socketNotify(int event);
68 class ZAssocChild : public yazpp_1::Z_Assoc {
71 ZAssocChild(yazpp_1::IPDU_Observable *the_PDU_Observable,
72 mp::ThreadPoolSocketObserver *m_thread_pool_observer,
73 const mp::Package *package,
78 yazpp_1::IPDU_Observer* sessionNotify(
79 yazpp_1::IPDU_Observable *the_PDU_Observable,
81 void recv_GDU(Z_GDU *apdu, int len);
87 mp::ThreadPoolSocketObserver *m_thread_pool_observer;
88 mp::Session m_session;
91 const mp::Package *m_package;
93 class ThreadPoolPackage : public mp::IThreadPoolMsg {
95 ThreadPoolPackage(mp::Package *package, mp::ZAssocChild *ses) :
96 m_assoc_child(ses), m_package(package) { };
98 IThreadPoolMsg *handle();
102 mp::ZAssocChild *m_assoc_child;
103 mp::Package *m_package;
106 class ZAssocServer : public yazpp_1::Z_Assoc {
109 ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable, int timeout,
110 int connect_max, std::string route);
111 void set_package(const mp::Package *package);
112 void set_thread_pool(ThreadPoolSocketObserver *m_thread_pool_observer);
114 yazpp_1::IPDU_Observer* sessionNotify(
115 yazpp_1::IPDU_Observable *the_PDU_Observable,
117 void recv_GDU(Z_GDU *apdu, int len);
120 void timeoutNotify();
121 void connectNotify();
123 mp::ThreadPoolSocketObserver *m_thread_pool_observer;
124 const mp::Package *m_package;
125 int m_session_timeout;
127 yazpp_1::LimitConnect limit_connect;
132 mp::ThreadPoolPackage::~ThreadPoolPackage()
137 void mp::ThreadPoolPackage::result()
139 m_assoc_child->m_no_requests--;
141 yazpp_1::GDU *gdu = &m_package->response();
146 m_assoc_child->send_GDU(gdu->get(), &len);
148 else if (!m_package->session().is_closed())
150 // no response package and yet the session is still open..
151 // means that request is unhandled..
152 yazpp_1::GDU *gdu_req = &m_package->request();
153 Z_GDU *z_gdu = gdu_req->get();
154 if (z_gdu && z_gdu->which == Z_GDU_Z3950)
156 // For Z39.50, response with a Close and shutdown
159 Z_APDU *apdu_response = odr.create_close(
160 z_gdu->u.z3950, Z_Close_systemProblem,
161 "unhandled Z39.50 request");
163 m_assoc_child->send_Z_PDU(apdu_response, &len);
165 else if (z_gdu && z_gdu->which == Z_GDU_HTTP_Request)
167 // For HTTP, respond with Server Error
171 = odr.create_HTTP_Response(m_package->session(),
172 z_gdu->u.HTTP_Request, 500);
173 m_assoc_child->send_GDU(zgdu_res, &len);
175 m_package->session().close();
178 if (m_assoc_child->m_no_requests == 0 && m_package->session().is_closed())
180 m_assoc_child->close();
185 mp::IThreadPoolMsg *mp::ThreadPoolPackage::handle()
187 m_package->move(m_assoc_child->m_route);
192 mp::ZAssocChild::ZAssocChild(yazpp_1::IPDU_Observable *PDU_Observable,
193 mp::ThreadPoolSocketObserver *my_thread_pool,
194 const mp::Package *package,
196 : Z_Assoc(PDU_Observable)
198 m_thread_pool_observer = my_thread_pool;
200 m_delete_flag = false;
203 const char *peername = PDU_Observable->getpeername();
205 peername = "unknown";
206 m_origin.set_tcpip_address(std::string(peername), m_session.id());
210 yazpp_1::IPDU_Observer *mp::ZAssocChild::sessionNotify(yazpp_1::IPDU_Observable
211 *the_PDU_Observable, int fd)
216 mp::ZAssocChild::~ZAssocChild()
220 void mp::ZAssocChild::recv_GDU(Z_GDU *z_pdu, int len)
224 mp::Package *p = new mp::Package(m_session, m_origin);
226 mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
227 p->copy_filter(*m_package);
228 p->request() = yazpp_1::GDU(z_pdu);
229 m_thread_pool_observer->put(tp);
232 void mp::ZAssocChild::failNotify()
234 // TODO: send Package to signal "close"
235 if (m_session.is_closed())
237 if (m_no_requests == 0)
245 mp::Package *p = new mp::Package(m_session, m_origin);
247 mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
248 p->copy_filter(*m_package);
249 m_thread_pool_observer->put(tp);
252 void mp::ZAssocChild::timeoutNotify()
257 void mp::ZAssocChild::connectNotify()
262 mp::ZAssocServer::ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable,
263 int timeout, int connect_max,
265 : Z_Assoc(PDU_Observable), m_session_timeout(timeout),
266 m_connect_max(connect_max), m_route(route)
272 void mp::ZAssocServer::set_package(const mp::Package *package)
277 void mp::ZAssocServer::set_thread_pool(ThreadPoolSocketObserver *observer)
279 m_thread_pool_observer = observer;
282 yazpp_1::IPDU_Observer *mp::ZAssocServer::sessionNotify(yazpp_1::IPDU_Observable
283 *the_PDU_Observable, int fd)
286 const char *peername = the_PDU_Observable->getpeername();
289 limit_connect.add_connect(peername);
290 limit_connect.cleanup(false);
291 int con_sz = limit_connect.get_total(peername);
292 if (m_connect_max && con_sz > m_connect_max)
295 mp::ZAssocChild *my =
296 new mp::ZAssocChild(the_PDU_Observable, m_thread_pool_observer,
298 my->timeout(m_session_timeout);
302 mp::ZAssocServer::~ZAssocServer()
306 void mp::ZAssocServer::recv_GDU(Z_GDU *apdu, int len)
310 void mp::ZAssocServer::failNotify()
314 void mp::ZAssocServer::timeoutNotify()
318 void mp::ZAssocServer::connectNotify()
322 mp::filter::FrontendNet::FrontendNet() : m_p(new Rep)
324 m_p->m_no_threads = 5;
325 m_p->m_listen_duration = 0;
326 m_p->m_session_timeout = 300; // 5 minutes
327 m_p->m_connect_max = 0;
331 mp::filter::FrontendNet::~FrontendNet()
336 for (i = 0; i<m_p->m_ports.size(); i++)
342 bool mp::My_Timer_Thread::timeout()
347 mp::My_Timer_Thread::My_Timer_Thread(yazpp_1::ISocketObservable *obs,
349 m_obs(obs), m_pipe(9123), m_timeout(false)
351 obs->addObserver(m_pipe.read_fd(), this);
352 obs->maskObserver(this, yazpp_1::SOCKET_OBSERVE_READ);
353 obs->timeoutObserver(this, duration);
356 void mp::My_Timer_Thread::socketNotify(int event)
359 m_obs->deleteObserver(this);
362 void mp::filter::FrontendNet::process(Package &package) const
367 My_Timer_Thread *tt = 0;
369 if (m_p->m_listen_duration)
370 tt = new My_Timer_Thread(&m_p->mySocketManager,
371 m_p->m_listen_duration);
373 ThreadPoolSocketObserver tp(&m_p->mySocketManager, m_p->m_no_threads);
375 for (i = 0; i<m_p->m_ports.size(); i++)
377 m_p->az[i]->set_package(&package);
378 m_p->az[i]->set_thread_pool(&tp);
380 while (m_p->mySocketManager.processEvent() > 0)
382 if (tt && tt->timeout())
388 void mp::filter::FrontendNet::configure(const xmlNode * ptr, bool test_only)
390 if (!ptr || !ptr->children)
392 throw mp::filter::FilterException("No ports for Frontend");
394 std::vector<Port> ports;
395 for (ptr = ptr->children; ptr; ptr = ptr->next)
397 if (ptr->type != XML_ELEMENT_NODE)
399 if (!strcmp((const char *) ptr->name, "port"))
402 const struct _xmlAttr *attr;
403 for (attr = ptr->properties; attr; attr = attr->next)
405 if (!strcmp((const char *) attr->name, "route"))
406 port.route = mp::xml::get_text(attr);
408 port.port = mp::xml::get_text(ptr);
409 ports.push_back(port);
412 else if (!strcmp((const char *) ptr->name, "threads"))
414 std::string threads_str = mp::xml::get_text(ptr);
415 int threads = atoi(threads_str.c_str());
417 throw mp::filter::FilterException("Bad value for threads: "
419 m_p->m_no_threads = threads;
421 else if (!strcmp((const char *) ptr->name, "timeout"))
423 std::string timeout_str = mp::xml::get_text(ptr);
424 int timeout = atoi(timeout_str.c_str());
426 throw mp::filter::FilterException("Bad value for timeout: "
428 m_p->m_session_timeout = timeout;
430 else if (!strcmp((const char *) ptr->name, "connect-max"))
432 m_p->m_connect_max = mp::xml::get_int(ptr, 0);
436 throw mp::filter::FilterException("Bad element "
437 + std::string((const char *)
446 void mp::filter::FrontendNet::set_ports(std::vector<std::string> &ports)
448 std::vector<Port> nports;
451 for (i = 0; i < ports.size(); i++)
455 nport.port = ports[i];
457 nports.push_back(nport);
463 void mp::filter::FrontendNet::set_ports(std::vector<Port> &ports)
465 m_p->m_ports = ports;
467 m_p->az = new mp::ZAssocServer *[m_p->m_ports.size()];
469 // Create mp::ZAssocServer for each port
471 for (i = 0; i<m_p->m_ports.size(); i++)
473 // create a PDU assoc object (one per mp::ZAssocServer)
474 yazpp_1::PDU_Assoc *as = new yazpp_1::PDU_Assoc(&m_p->mySocketManager);
476 // create ZAssoc with PDU Assoc
477 m_p->az[i] = new mp::ZAssocServer(as,
478 m_p->m_session_timeout,
480 m_p->m_ports[i].route);
481 if (m_p->az[i]->server(m_p->m_ports[i].port.c_str()))
483 throw mp::filter::FilterException("Unable to bind to address "
484 + std::string(m_p->m_ports[i].port));
489 void mp::filter::FrontendNet::set_listen_duration(int d)
491 m_p->m_listen_duration = d;
494 static mp::filter::Base* filter_creator()
496 return new mp::filter::FrontendNet;
500 struct metaproxy_1_filter_struct metaproxy_1_filter_frontend_net = {
510 * c-file-style: "Stroustrup"
511 * indent-tabs-mode: nil
513 * vim: shiftwidth=4 tabstop=8 expandtab