src/filter_frontend_net.cpp

   1
   2
   3 #include "config.hpp"
   4
   5 #include "filter.hpp"
   6 #include "router.hpp"
   7 #include "package.hpp"
   8 #include "thread_pool_observer.hpp"
   9 #include "filter_frontend_net.hpp"
  10 #include <yaz++/z-assoc.h>
  11 #include <yaz++/pdu-assoc.h>
  12 #include <yaz++/socket-manager.h>
  13 #include <yaz/log.h>
  14
  15 #include <iostream>
  16
  17 using namespace yp2;
  18
  19 class ZAssocServerChild : public yazpp_1::Z_Assoc {
  20 public:
  21     ~ZAssocServerChild();
  22     ZAssocServerChild(yazpp_1::IPDU_Observable *the_PDU_Observable,
  23                ThreadPoolSocketObserver *m_thread_pool_observer,
  24                const Package *package);
  25     int m_no_requests;
  26 private:
  27     yazpp_1::IPDU_Observer* sessionNotify(
  28         yazpp_1::IPDU_Observable *the_PDU_Observable,
  29         int fd);
  30     void recv_GDU(Z_GDU *apdu, int len);
  31
  32     void failNotify();
  33     void timeoutNotify();
  34     void connectNotify();
  35 private:
  36     ThreadPoolSocketObserver *m_thread_pool_observer;
  37     Session m_session;
  38     Origin m_origin;
  39     bool m_delete_flag;
  40     const Package *m_package;
  41 };
  42
  43
  44 class ThreadPoolPackage : public IThreadPoolMsg {
  45 public:
  46     ThreadPoolPackage(Package *package, ZAssocServerChild *ses) :
  47         m_session(ses), m_package(package) { };
  48     ~ThreadPoolPackage();
  49     IThreadPoolMsg *handle();
  50     void result();
  51
  52 private:
  53     ZAssocServerChild *m_session;
  54     Package *m_package;
  55
  56 };
  57
  58 ThreadPoolPackage::~ThreadPoolPackage()
  59 {
  60     delete m_package;
  61 }
  62
  63 void ThreadPoolPackage::result()
  64 {
  65     m_session->m_no_requests--;
  66
  67     yazpp_1::GDU *gdu = &m_package->response();
  68     if (gdu->get())
  69     {
  70         int len;
  71         m_session->send_GDU(gdu->get(), &len);
  72     }
  73     if (m_session->m_no_requests == 0 && m_package->session().is_closed())
  74         delete m_session;
  75     delete this;
  76 }
  77
  78 IThreadPoolMsg *ThreadPoolPackage::handle()
  79 {
  80     m_package->move();
  81     return this;
  82 }
  83
  84
  85 ZAssocServerChild::ZAssocServerChild(yazpp_1::IPDU_Observable *the_PDU_Observable,
  86                        ThreadPoolSocketObserver *my_thread_pool,
  87                        const Package *package)
  88     :  Z_Assoc(the_PDU_Observable)
  89 {
  90     m_thread_pool_observer = my_thread_pool;
  91     m_no_requests = 0;
  92     m_delete_flag = false;
  93     m_package = package;
  94 }
  95
  96
  97 yazpp_1::IPDU_Observer *ZAssocServerChild::sessionNotify(yazpp_1::IPDU_Observable
  98                                                   *the_PDU_Observable, int fd)
  99 {
 100     return 0;
 101 }
 102
 103 ZAssocServerChild::~ZAssocServerChild()
 104 {
 105 }
 106
 107 void ZAssocServerChild::recv_GDU(Z_GDU *z_pdu, int len)
 108 {
 109     m_no_requests++;
 110
 111     Package *p = new Package(m_session, m_origin);
 112
 113     ThreadPoolPackage *tp = new ThreadPoolPackage(p, this);
 114     p->copy_filter(*m_package);
 115     p->request() = yazpp_1::GDU(z_pdu);
 116     m_thread_pool_observer->put(tp);
 117 }
 118
 119 void ZAssocServerChild::failNotify()
 120 {
 121     // TODO: send Package to signal "close"
 122     if (m_session.is_closed())
 123         return;
 124     m_no_requests++;
 125
 126     m_session.close();
 127
 128     Package *p = new Package(m_session, m_origin);
 129
 130     ThreadPoolPackage *tp = new ThreadPoolPackage(p, this);
 131     p->copy_filter(*m_package);
 132     m_thread_pool_observer->put(tp);
 133 }
 134
 135 void ZAssocServerChild::timeoutNotify()
 136 {
 137     failNotify();
 138 }
 139
 140 void ZAssocServerChild::connectNotify()
 141 {
 142
 143 }
 144
 145 class ZAssocServer : public yazpp_1::Z_Assoc {
 146 public:
 147     ~ZAssocServer();
 148     ZAssocServer(yazpp_1::IPDU_Observable *the_PDU_Observable,
 149               ThreadPoolSocketObserver *m_thread_pool_observer,
 150               const Package *package);
 151 private:
 152     yazpp_1::IPDU_Observer* sessionNotify(
 153         yazpp_1::IPDU_Observable *the_PDU_Observable,
 154         int fd);
 155     void recv_GDU(Z_GDU *apdu, int len);
 156
 157     void failNotify();
 158     void timeoutNotify();
 159     void connectNotify();
 160 private:
 161     ThreadPoolSocketObserver *m_thread_pool_observer;
 162     const Package *m_package;
 163 };
 164
 165
 166 ZAssocServer::ZAssocServer(yazpp_1::IPDU_Observable *the_PDU_Observable,
 167                      ThreadPoolSocketObserver *thread_pool_observer,
 168                      const Package *package)
 169     :  Z_Assoc(the_PDU_Observable)
 170 {
 171     m_thread_pool_observer = thread_pool_observer;
 172     m_package = package;
 173
 174 }
 175
 176 yazpp_1::IPDU_Observer *ZAssocServer::sessionNotify(yazpp_1::IPDU_Observable
 177                                                  *the_PDU_Observable, int fd)
 178 {
 179     ZAssocServerChild *my =
 180         new ZAssocServerChild(the_PDU_Observable, m_thread_pool_observer,
 181                               m_package);
 182     return my;
 183 }
 184
 185 ZAssocServer::~ZAssocServer()
 186 {
 187 }
 188
 189 void ZAssocServer::recv_GDU(Z_GDU *apdu, int len)
 190 {
 191 }
 192
 193 void ZAssocServer::failNotify()
 194 {
 195 }
 196
 197 void ZAssocServer::timeoutNotify()
 198 {
 199 }
 200
 201 void ZAssocServer::connectNotify()
 202 {
 203 }
 204
 205 FilterFrontendNet::FilterFrontendNet()
 206 {
 207     m_no_threads = 5;
 208     m_listen_duration = 0;
 209 }
 210
 211 class My_Timer_Thread : public yazpp_1::ISocketObserver {
 212 private:
 213     yazpp_1::ISocketObservable *m_obs;
 214     int m_fd[2];
 215     bool m_timeout;
 216 public:
 217     My_Timer_Thread(yazpp_1::ISocketObservable *obs, int duration);
 218     void socketNotify(int event);
 219     bool timeout();
 220 };
 221
 222 bool My_Timer_Thread::timeout()
 223 {
 224     return m_timeout;
 225 }
 226
 227 My_Timer_Thread::My_Timer_Thread(yazpp_1::ISocketObservable *obs,
 228                                  int duration) :
 229     m_obs(obs), m_timeout(false)
 230 {
 231     pipe(m_fd);
 232     obs->addObserver(m_fd[0], this);
 233     obs->maskObserver(this, yazpp_1::SOCKET_OBSERVE_READ);
 234     obs->timeoutObserver(this, duration);
 235 }
 236
 237 void My_Timer_Thread::socketNotify(int event)
 238 {
 239     m_timeout = true;
 240     m_obs->deleteObserver(this);
 241     close(m_fd[0]);
 242     close(m_fd[1]);
 243 }
 244
 245 void FilterFrontendNet::process(Package &package) const {
 246     yazpp_1::SocketManager mySocketManager;
 247
 248     My_Timer_Thread *tt = 0;
 249     if (m_listen_duration)
 250         tt = new My_Timer_Thread(&mySocketManager, m_listen_duration);
 251
 252     ThreadPoolSocketObserver threadPool(&mySocketManager, m_no_threads);
 253
 254     ZAssocServer **az = new ZAssocServer *[m_ports.size()];
 255
 256     // Create ZAssocServer for each port
 257     size_t i;
 258     for (i = 0; i<m_ports.size(); i++)
 259     {
 260         // create a PDU assoc object (one per ZAssocServer)
 261         yazpp_1::PDU_Assoc *as = new yazpp_1::PDU_Assoc(&mySocketManager);
 262
 263         // create ZAssoc with PDU Assoc
 264         az[i] = new ZAssocServer(as, &threadPool, &package);
 265         az[i]->server(m_ports[i].c_str());
 266     }
 267     while (mySocketManager.processEvent() > 0)
 268     {
 269         if (tt && tt->timeout())
 270             break;
 271     }
 272     for (i = 0; i<m_ports.size(); i++)
 273         delete az[i];
 274
 275     delete [] az;
 276     delete tt;
 277 }
 278
 279 std::vector<std::string> &FilterFrontendNet::ports()
 280 {
 281     return m_ports;
 282 }
 283
 284 int &FilterFrontendNet::listen_duration()
 285 {
 286     return m_listen_duration;
 287 }
 288