src/filter_frontend_net.cpp

   1 /* $Id: filter_frontend_net.cpp,v 1.16 2006-03-16 10:40:59 adam Exp $
   2    Copyright (c) 2005-2006, Index Data.
   3
   4 %LICENSE%
   5  */
   6
   7 #include "config.hpp"
   8
   9 #include "pipe.hpp"
  10 #include "filter.hpp"
  11 #include "package.hpp"
  12 #include "thread_pool_observer.hpp"
  13 #include "filter_frontend_net.hpp"
  14 #include <yaz++/z-assoc.h>
  15 #include <yaz++/pdu-assoc.h>
  16 #include <yaz++/socket-manager.h>
  17 #include <yaz/log.h>
  18
  19 #include <iostream>
  20
  21 namespace mp = metaproxy_1;
  22
  23 namespace metaproxy_1 {
  24     namespace filter {
  25         class FrontendNet::Rep {
  26             friend class FrontendNet;
  27             int m_no_threads;
  28             std::vector<std::string> m_ports;
  29             int m_listen_duration;
  30         };
  31     }
  32     class My_Timer_Thread : public yazpp_1::ISocketObserver {
  33     private:
  34         yazpp_1::ISocketObservable *m_obs;
  35         Pipe m_pipe;
  36         bool m_timeout;
  37     public:
  38         My_Timer_Thread(yazpp_1::ISocketObservable *obs, int duration);
  39         void socketNotify(int event);
  40         bool timeout();
  41     };
  42     class ZAssocChild : public yazpp_1::Z_Assoc {
  43     public:
  44         ~ZAssocChild();
  45         ZAssocChild(yazpp_1::IPDU_Observable *the_PDU_Observable,
  46                           mp::ThreadPoolSocketObserver *m_thread_pool_observer,
  47                           const mp::Package *package);
  48         int m_no_requests;
  49     private:
  50         yazpp_1::IPDU_Observer* sessionNotify(
  51             yazpp_1::IPDU_Observable *the_PDU_Observable,
  52             int fd);
  53         void recv_GDU(Z_GDU *apdu, int len);
  54
  55         void failNotify();
  56         void timeoutNotify();
  57         void connectNotify();
  58     private:
  59         mp::ThreadPoolSocketObserver *m_thread_pool_observer;
  60         mp::Session m_session;
  61         mp::Origin m_origin;
  62         bool m_delete_flag;
  63         const mp::Package *m_package;
  64     };
  65     class ThreadPoolPackage : public mp::IThreadPoolMsg {
  66     public:
  67         ThreadPoolPackage(mp::Package *package, mp::ZAssocChild *ses) :
  68             m_session(ses), m_package(package) { };
  69         ~ThreadPoolPackage();
  70         IThreadPoolMsg *handle();
  71         void result();
  72
  73     private:
  74         mp::ZAssocChild *m_session;
  75         mp::Package *m_package;
  76
  77     };
  78     class ZAssocServer : public yazpp_1::Z_Assoc {
  79     public:
  80         ~ZAssocServer();
  81         ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable,
  82                      mp::ThreadPoolSocketObserver *m_thread_pool_observer,
  83                      const mp::Package *package);
  84     private:
  85         yazpp_1::IPDU_Observer* sessionNotify(
  86             yazpp_1::IPDU_Observable *the_PDU_Observable,
  87             int fd);
  88         void recv_GDU(Z_GDU *apdu, int len);
  89
  90         void failNotify();
  91         void timeoutNotify();
  92     void connectNotify();
  93     private:
  94         mp::ThreadPoolSocketObserver *m_thread_pool_observer;
  95         const mp::Package *m_package;
  96     };
  97 }
  98
  99 mp::ThreadPoolPackage::~ThreadPoolPackage()
 100 {
 101     delete m_package;
 102 }
 103
 104 void mp::ThreadPoolPackage::result()
 105 {
 106     m_session->m_no_requests--;
 107
 108     yazpp_1::GDU *gdu = &m_package->response();
 109     if (gdu->get())
 110     {
 111         int len;
 112         m_session->send_GDU(gdu->get(), &len);
 113     }
 114     if (m_session->m_no_requests == 0 && m_package->session().is_closed())
 115         delete m_session;
 116     delete this;
 117 }
 118
 119 mp::IThreadPoolMsg *mp::ThreadPoolPackage::handle()
 120 {
 121     m_package->move();
 122     return this;
 123 }
 124
 125
 126 mp::ZAssocChild::ZAssocChild(yazpp_1::IPDU_Observable *PDU_Observable,
 127                                      mp::ThreadPoolSocketObserver *my_thread_pool,
 128                                      const mp::Package *package)
 129     :  Z_Assoc(PDU_Observable)
 130 {
 131     m_thread_pool_observer = my_thread_pool;
 132     m_no_requests = 0;
 133     m_delete_flag = false;
 134     m_package = package;
 135 }
 136
 137
 138 yazpp_1::IPDU_Observer *mp::ZAssocChild::sessionNotify(yazpp_1::IPDU_Observable
 139                                                   *the_PDU_Observable, int fd)
 140 {
 141     return 0;
 142 }
 143
 144 mp::ZAssocChild::~ZAssocChild()
 145 {
 146 }
 147
 148 void mp::ZAssocChild::recv_GDU(Z_GDU *z_pdu, int len)
 149 {
 150     m_no_requests++;
 151
 152     mp::Package *p = new mp::Package(m_session, m_origin);
 153
 154     mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
 155     p->copy_filter(*m_package);
 156     p->request() = yazpp_1::GDU(z_pdu);
 157     m_thread_pool_observer->put(tp);
 158 }
 159
 160 void mp::ZAssocChild::failNotify()
 161 {
 162     // TODO: send Package to signal "close"
 163     if (m_session.is_closed())
 164         return;
 165     m_no_requests++;
 166
 167     m_session.close();
 168
 169     mp::Package *p = new mp::Package(m_session, m_origin);
 170
 171     mp::ThreadPoolPackage *tp = new mp::ThreadPoolPackage(p, this);
 172     p->copy_filter(*m_package);
 173     m_thread_pool_observer->put(tp);
 174 }
 175
 176 void mp::ZAssocChild::timeoutNotify()
 177 {
 178     failNotify();
 179 }
 180
 181 void mp::ZAssocChild::connectNotify()
 182 {
 183
 184 }
 185
 186 mp::ZAssocServer::ZAssocServer(yazpp_1::IPDU_Observable *PDU_Observable,
 187                            mp::ThreadPoolSocketObserver *thread_pool_observer,
 188                            const mp::Package *package)
 189     :  Z_Assoc(PDU_Observable)
 190 {
 191     m_thread_pool_observer = thread_pool_observer;
 192     m_package = package;
 193
 194 }
 195
 196 yazpp_1::IPDU_Observer *mp::ZAssocServer::sessionNotify(yazpp_1::IPDU_Observable
 197                                                  *the_PDU_Observable, int fd)
 198 {
 199     mp::ZAssocChild *my =
 200         new mp::ZAssocChild(the_PDU_Observable, m_thread_pool_observer,
 201                              m_package);
 202     return my;
 203 }
 204
 205 mp::ZAssocServer::~ZAssocServer()
 206 {
 207 }
 208
 209 void mp::ZAssocServer::recv_GDU(Z_GDU *apdu, int len)
 210 {
 211 }
 212
 213 void mp::ZAssocServer::failNotify()
 214 {
 215 }
 216
 217 void mp::ZAssocServer::timeoutNotify()
 218 {
 219 }
 220
 221 void mp::ZAssocServer::connectNotify()
 222 {
 223 }
 224
 225 mp::filter::FrontendNet::FrontendNet() : m_p(new Rep)
 226 {
 227     m_p->m_no_threads = 5;
 228     m_p->m_listen_duration = 0;
 229 }
 230
 231 mp::filter::FrontendNet::~FrontendNet()
 232 {
 233 }
 234
 235 bool mp::My_Timer_Thread::timeout()
 236 {
 237     return m_timeout;
 238 }
 239
 240 mp::My_Timer_Thread::My_Timer_Thread(yazpp_1::ISocketObservable *obs,
 241                                  int duration) :
 242     m_obs(obs), m_pipe(9123), m_timeout(false)
 243 {
 244     obs->addObserver(m_pipe.read_fd(), this);
 245     obs->maskObserver(this, yazpp_1::SOCKET_OBSERVE_READ);
 246     obs->timeoutObserver(this, duration);
 247 }
 248
 249 void mp::My_Timer_Thread::socketNotify(int event)
 250 {
 251     m_timeout = true;
 252     m_obs->deleteObserver(this);
 253 }
 254
 255 void mp::filter::FrontendNet::process(Package &package) const
 256 {
 257     if (m_p->m_ports.size() == 0)
 258         return;
 259
 260     yazpp_1::SocketManager mySocketManager;
 261
 262     My_Timer_Thread *tt = 0;
 263     if (m_p->m_listen_duration)
 264         tt = new My_Timer_Thread(&mySocketManager, m_p->m_listen_duration);
 265
 266     ThreadPoolSocketObserver threadPool(&mySocketManager, m_p->m_no_threads);
 267
 268     mp::ZAssocServer **az = new mp::ZAssocServer *[m_p->m_ports.size()];
 269
 270     // Create mp::ZAssocServer for each port
 271     size_t i;
 272     for (i = 0; i<m_p->m_ports.size(); i++)
 273     {
 274         // create a PDU assoc object (one per mp::ZAssocServer)
 275         yazpp_1::PDU_Assoc *as = new yazpp_1::PDU_Assoc(&mySocketManager);
 276
 277         // create ZAssoc with PDU Assoc
 278         az[i] = new mp::ZAssocServer(as, &threadPool, &package);
 279         az[i]->server(m_p->m_ports[i].c_str());
 280     }
 281     while (mySocketManager.processEvent() > 0)
 282     {
 283         if (tt && tt->timeout())
 284             break;
 285     }
 286     for (i = 0; i<m_p->m_ports.size(); i++)
 287         delete az[i];
 288
 289     delete [] az;
 290     delete tt;
 291 }
 292
 293 void mp::filter::FrontendNet::configure(const xmlNode * ptr)
 294 {
 295     if (!ptr || !ptr->children)
 296     {
 297         throw mp::filter::FilterException("No ports for Frontend");
 298     }
 299     std::vector<std::string> ports;
 300     for (ptr = ptr->children; ptr; ptr = ptr->next)
 301     {
 302         if (ptr->type != XML_ELEMENT_NODE)
 303             continue;
 304         if (!strcmp((const char *) ptr->name, "port"))
 305         {
 306             std::string port = mp::xml::get_text(ptr);
 307             ports.push_back(port);
 308
 309         }
 310         else if (!strcmp((const char *) ptr->name, "threads"))
 311         {
 312             std::string threads_str = mp::xml::get_text(ptr);
 313             int threads = atoi(threads_str.c_str());
 314             if (threads < 1)
 315                 throw mp::filter::FilterException("Bad value for threads: "
 316                                                    + threads_str);
 317             m_p->m_no_threads = threads;
 318         }
 319         else
 320         {
 321             throw mp::filter::FilterException("Bad element "
 322                                                + std::string((const char *)
 323                                                              ptr->name));
 324         }
 325     }
 326     m_p->m_ports = ports;
 327 }
 328
 329 std::vector<std::string> &mp::filter::FrontendNet::ports()
 330 {
 331     return m_p->m_ports;
 332 }
 333
 334 int &mp::filter::FrontendNet::listen_duration()
 335 {
 336     return m_p->m_listen_duration;
 337 }
 338
 339 static mp::filter::Base* filter_creator()
 340 {
 341     return new mp::filter::FrontendNet;
 342 }
 343
 344 extern "C" {
 345     struct metaproxy_1_filter_struct metaproxy_1_filter_frontend_net = {
 346         0,
 347         "frontend_net",
 348         filter_creator
 349     };
 350 }
 351
 352 /*
 353  * Local variables:
 354  * c-basic-offset: 4
 355  * indent-tabs-mode: nil
 356  * c-file-style: "stroustrup"
 357  * End:
 358  * vim: shiftwidth=4 tabstop=8 expandtab
 359  */