src/filter_frontend_net.cpp

   1 /* $Id: filter_frontend_net.cpp,v 1.26 2008-02-20 15:07:51 adam Exp $
   2    Copyright (c) 2005-2008, Index Data.
   3
   4 This file is part of Metaproxy.
   5
   6 Metaproxy is free software; you can redistribute it and/or modify it under
   7 the terms of the GNU General Public License as published by the Free
   8 Software Foundation; either version 2, or (at your option) any later
   9 version.
  10
  11 Metaproxy is distributed in the hope that it will be useful, but WITHOUT ANY
  12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14 for more details.
  15
  16 You should have received a copy of the GNU General Public License
  17 along with Metaproxy; see the file LICENSE.  If not, write to the
  18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
  19 02111-1307, USA.
  20  */
  21
  22 #include "config.hpp"
  23
  24 #include "util.hpp"
  25 #include "pipe.hpp"
  26 #include "filter.hpp"
  27 #include "package.hpp"
  28 #include "thread_pool_observer.hpp"
  29 #include "filter_frontend_net.hpp"
  30 #include <yazpp/z-assoc.h>
  31 #include <yazpp/pdu-assoc.h>
  32 #include <yazpp/socket-manager.h>
  33 #include <yaz/log.h>
  34
  35 #include <iostream>
  36
  37 namespace mp = metaproxy_1;
  38
  39 namespace metaproxy_1 {
  40     class My_Timer_Thread;
  41     class ZAssocServer;
  42     namespace filter {
  43         class FrontendNet::Rep {
  44             friend class FrontendNet;
  45             int m_no_threads;
  46             std::vector<std::string> m_ports;
  47             int m_listen_duration;
  48             int m_session_timeout;
  49             yazpp_1::SocketManager mySocketManager;
  50
  51             ZAssocServer **az;
  52         };
  53     }
  54     class My_Timer_Thread : public yazpp_1::ISocketObserver {
  55     private:
  56         yazpp_1::ISocketObservable *m_obs;
  57         Pipe m_pipe;
  58         bool m_timeout;
  59     public:
  60         My_Timer_Thread(yazpp_1::ISocketObservable *obs, int duration);
  61         void socketNotify(int event);
  62         bool timeout();
  63     };
  64     class ZAssocChild : public yazpp_1::Z_Assoc {
  65     public:
  66         ~ZAssocChild();
  67         ZAssocChild(yazpp_1::IPDU_Observable *the_PDU_Observable,
  68                           mp::ThreadPoolSocketObserver *m_thread_pool_observer,
  69                           const mp::Package *package);
  70         int m_no_requests;
  71     private:
  72         yazpp_1::IPDU_Observer* sessionNotify(
  73             yazpp_1::IPDU_Observable *the_PDU_Observable,
  74             int fd);
  75         void recv_GDU(Z_GDU *apdu, int len);
  76
  77         void failNotify();
  78         void timeoutNotify();
  79         void connectNotify();
  80     private:
  81         mp::ThreadPoolSocketObserver *m_thread_pool_observer;
  82         mp::Session m_session;
  83         mp::Origin m_origin;
  84         bool m_delete_flag;
  85         const mp::Package *m_package;
  86     };
  87     class ThreadPoolPackage : public mp::IThreadPoolMsg {
  88     public:
  89         ThreadPoolPackage(mp::Package *package, mp::ZAssocChild *ses) :
  90             m_session(ses), m_package(package) { };
  91         ~ThreadPoolPackage();
  92         IThreadPoolMsg *handle();
  93         void result();
  94
  95     private:
  96         mp::ZAssocChild *m_session;
  97         mp::Package *m_package;
  98
  99     };
 100     class ZAssocServer : public yazpp_1::Z_Assoc {
 101     public:
 102         ~ZAssocServer();
 103         ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable, int timeout);
 104         void set_package(const mp::Package *package);
 105         void set_thread_pool(ThreadPoolSocketObserver *m_thread_pool_observer);
 106     private:
 107         yazpp_1::IPDU_Observer* sessionNotify(
 108             yazpp_1::IPDU_Observable *the_PDU_Observable,
 109             int fd);
 110         void recv_GDU(Z_GDU *apdu, int len);
 111
 112         void failNotify();
 113         void timeoutNotify();
 114     void connectNotify();
 115     private:
 116         mp::ThreadPoolSocketObserver *m_thread_pool_observer;
 117         const mp::Package *m_package;
 118         int m_session_timeout;
 119     };
 120 }
 121
 122 mp::ThreadPoolPackage::~ThreadPoolPackage()
 123 {
 124     delete m_package;
 125 }
 126
 127 void mp::ThreadPoolPackage::result()
 128 {
 129     m_session->m_no_requests--;
 130
 131     yazpp_1::GDU *gdu = &m_package->response();
 132
 133     if (gdu->get())
 134     {
 135         int len;
 136         m_session->send_GDU(gdu->get(), &len);
 137     }
 138     else if (!m_package->session().is_closed())
 139     {
 140         // no response package and yet the session is still open..
 141         // means that request is unhandled..
 142         yazpp_1::GDU *gdu_req = &m_package->request();
 143         Z_GDU *z_gdu = gdu_req->get();
 144         if (z_gdu && z_gdu->which == Z_GDU_Z3950)
 145         {
 146             // For Z39.50, response with a Close and shutdown
 147             mp::odr odr;
 148             int len;
 149             Z_APDU *apdu_response = odr.create_close(
 150                 z_gdu->u.z3950, Z_Close_systemProblem,
 151                 "unhandled Z39.50 request");
 152
 153             m_session->send_Z_PDU(apdu_response, &len);
 154             m_package->session().close();
 155         }
 156     }
 157
 158     if (m_session->m_no_requests == 0 && m_package->session().is_closed())
 159     {
 160         m_session->close();
 161     }
 162     delete this;
 163 }
 164
 165 mp::IThreadPoolMsg *mp::ThreadPoolPackage::handle()
 166 {
 167     m_package->move();
 168     return this;
 169 }
 170
 171
 172 mp::ZAssocChild::ZAssocChild(yazpp_1::IPDU_Observable *PDU_Observable,
 173                                      mp::ThreadPoolSocketObserver *my_thread_pool,
 174                                      const mp::Package *package)
 175     :  Z_Assoc(PDU_Observable)
 176 {
 177     m_thread_pool_observer = my_thread_pool;
 178     m_no_requests = 0;
 179     m_delete_flag = false;
 180     m_package = package;
 181     const char *peername = PDU_Observable->getpeername();
 182     if (!peername)
 183         peername = "unknown";
 184     m_origin.set_tcpip_address(std::string(peername), m_session.id());
 185 }
 186
 187
 188 yazpp_1::IPDU_Observer *mp::ZAssocChild::sessionNotify(yazpp_1::IPDU_Observable
 189                                                   *the_PDU_Observable, int fd)
 190 {
 191     return 0;
 192 }
 193
 194 mp::ZAssocChild::~ZAssocChild()
 195 {
 196 }
 197
 198 void mp::ZAssocChild::recv_GDU(Z_GDU *z_pdu, int len)
 199 {
 200     m_no_requests++;
 201
 202     mp::Package *p = new mp::Package(m_session, m_origin);
 203
 204     mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
 205     p->copy_filter(*m_package);
 206     p->request() = yazpp_1::GDU(z_pdu);
 207     m_thread_pool_observer->put(tp);
 208 }
 209
 210 void mp::ZAssocChild::failNotify()
 211 {
 212     // TODO: send Package to signal "close"
 213     if (m_session.is_closed())
 214     {
 215         delete this;
 216         return;
 217     }
 218     m_no_requests++;
 219
 220     m_session.close();
 221
 222     mp::Package *p = new mp::Package(m_session, m_origin);
 223
 224     mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
 225     p->copy_filter(*m_package);
 226     m_thread_pool_observer->put(tp);
 227 }
 228
 229 void mp::ZAssocChild::timeoutNotify()
 230 {
 231     failNotify();
 232 }
 233
 234 void mp::ZAssocChild::connectNotify()
 235 {
 236
 237 }
 238
 239 mp::ZAssocServer::ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable,
 240                                int timeout)
 241     :  Z_Assoc(PDU_Observable), m_session_timeout(timeout)
 242 {
 243     m_package = 0;
 244 }
 245
 246
 247 void mp::ZAssocServer::set_package(const mp::Package *package)
 248 {
 249     m_package = package;
 250 }
 251
 252 void mp::ZAssocServer::set_thread_pool(ThreadPoolSocketObserver *observer)
 253 {
 254     m_thread_pool_observer = observer;
 255 }
 256
 257 yazpp_1::IPDU_Observer *mp::ZAssocServer::sessionNotify(yazpp_1::IPDU_Observable
 258                                                  *the_PDU_Observable, int fd)
 259 {
 260     mp::ZAssocChild *my =
 261         new mp::ZAssocChild(the_PDU_Observable, m_thread_pool_observer,
 262                              m_package);
 263     my->timeout(m_session_timeout);
 264     return my;
 265 }
 266
 267 mp::ZAssocServer::~ZAssocServer()
 268 {
 269 }
 270
 271 void mp::ZAssocServer::recv_GDU(Z_GDU *apdu, int len)
 272 {
 273 }
 274
 275 void mp::ZAssocServer::failNotify()
 276 {
 277 }
 278
 279 void mp::ZAssocServer::timeoutNotify()
 280 {
 281 }
 282
 283 void mp::ZAssocServer::connectNotify()
 284 {
 285 }
 286
 287 mp::filter::FrontendNet::FrontendNet() : m_p(new Rep)
 288 {
 289     m_p->m_no_threads = 5;
 290     m_p->m_listen_duration = 0;
 291     m_p->m_session_timeout = 300; // 5 minutes
 292     m_p->az = 0;
 293 }
 294
 295 mp::filter::FrontendNet::~FrontendNet()
 296 {
 297     if (m_p->az)
 298     {
 299         size_t i;
 300         for (i = 0; i<m_p->m_ports.size(); i++)
 301             delete m_p->az[i];
 302         delete [] m_p->az;
 303     }
 304 }
 305
 306 bool mp::My_Timer_Thread::timeout()
 307 {
 308     return m_timeout;
 309 }
 310
 311 mp::My_Timer_Thread::My_Timer_Thread(yazpp_1::ISocketObservable *obs,
 312                                  int duration) :
 313     m_obs(obs), m_pipe(9123), m_timeout(false)
 314 {
 315     obs->addObserver(m_pipe.read_fd(), this);
 316     obs->maskObserver(this, yazpp_1::SOCKET_OBSERVE_READ);
 317     obs->timeoutObserver(this, duration);
 318 }
 319
 320 void mp::My_Timer_Thread::socketNotify(int event)
 321 {
 322     m_timeout = true;
 323     m_obs->deleteObserver(this);
 324 }
 325
 326 void mp::filter::FrontendNet::process(Package &package) const
 327 {
 328     if (m_p->az == 0)
 329         return;
 330     size_t i;
 331     My_Timer_Thread *tt = 0;
 332
 333     if (m_p->m_listen_duration)
 334         tt = new My_Timer_Thread(&m_p->mySocketManager,
 335                                  m_p->m_listen_duration);
 336
 337     ThreadPoolSocketObserver tp(&m_p->mySocketManager, m_p->m_no_threads);
 338
 339     for (i = 0; i<m_p->m_ports.size(); i++)
 340     {
 341         m_p->az[i]->set_package(&package);
 342         m_p->az[i]->set_thread_pool(&tp);
 343     }
 344     while (m_p->mySocketManager.processEvent() > 0)
 345     {
 346         if (tt && tt->timeout())
 347             break;
 348     }
 349     delete tt;
 350 }
 351
 352 void mp::filter::FrontendNet::configure(const xmlNode * ptr, bool test_only)
 353 {
 354     if (!ptr || !ptr->children)
 355     {
 356         throw mp::filter::FilterException("No ports for Frontend");
 357     }
 358     std::vector<std::string> ports;
 359     for (ptr = ptr->children; ptr; ptr = ptr->next)
 360     {
 361         if (ptr->type != XML_ELEMENT_NODE)
 362             continue;
 363         if (!strcmp((const char *) ptr->name, "port"))
 364         {
 365             std::string port = mp::xml::get_text(ptr);
 366             ports.push_back(port);
 367
 368         }
 369         else if (!strcmp((const char *) ptr->name, "threads"))
 370         {
 371             std::string threads_str = mp::xml::get_text(ptr);
 372             int threads = atoi(threads_str.c_str());
 373             if (threads < 1)
 374                 throw mp::filter::FilterException("Bad value for threads: "
 375                                                    + threads_str);
 376             m_p->m_no_threads = threads;
 377         }
 378         else if (!strcmp((const char *) ptr->name, "timeout"))
 379         {
 380             std::string timeout_str = mp::xml::get_text(ptr);
 381             int timeout = atoi(timeout_str.c_str());
 382             if (timeout < 1)
 383                 throw mp::filter::FilterException("Bad value for timeout: "
 384                                                    + timeout_str);
 385             m_p->m_session_timeout = timeout;
 386         }
 387         else
 388         {
 389             throw mp::filter::FilterException("Bad element "
 390                                                + std::string((const char *)
 391                                                              ptr->name));
 392         }
 393     }
 394     if (test_only)
 395         return;
 396     set_ports(ports);
 397 }
 398
 399 void mp::filter::FrontendNet::set_ports(std::vector<std::string> &ports)
 400 {
 401     m_p->m_ports = ports;
 402
 403     m_p->az = new mp::ZAssocServer *[m_p->m_ports.size()];
 404
 405     // Create mp::ZAssocServer for each port
 406     size_t i;
 407     for (i = 0; i<m_p->m_ports.size(); i++)
 408     {
 409         // create a PDU assoc object (one per mp::ZAssocServer)
 410         yazpp_1::PDU_Assoc *as = new yazpp_1::PDU_Assoc(&m_p->mySocketManager);
 411
 412         // create ZAssoc with PDU Assoc
 413         m_p->az[i] = new mp::ZAssocServer(as,
 414                                           m_p->m_session_timeout);
 415         if (m_p->az[i]->server(m_p->m_ports[i].c_str()))
 416         {
 417             throw mp::filter::FilterException("Unable to bind to address "
 418                                               + std::string(m_p->m_ports[i]));
 419         }
 420     }
 421 }
 422
 423 void mp::filter::FrontendNet::set_listen_duration(int d)
 424 {
 425     m_p->m_listen_duration = d;
 426 }
 427
 428 static mp::filter::Base* filter_creator()
 429 {
 430     return new mp::filter::FrontendNet;
 431 }
 432
 433 extern "C" {
 434     struct metaproxy_1_filter_struct metaproxy_1_filter_frontend_net = {
 435         0,
 436         "frontend_net",
 437         filter_creator
 438     };
 439 }
 440
 441 /*
 442  * Local variables:
 443  * c-basic-offset: 4
 444  * indent-tabs-mode: nil
 445  * c-file-style: "stroustrup"
 446  * End:
 447  * vim: shiftwidth=4 tabstop=8 expandtab
 448  */