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 P2_Session : public yazpp_1::Z_Assoc {
  20 public:
  21     ~P2_Session();
  22     P2_Session(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, P2_Session *ses) :
  47         m_session(ses), m_package(package) { };
  48     ~ThreadPoolPackage();
  49     IThreadPoolMsg *handle();
  50     void result();
  51
  52 private:
  53     P2_Session *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 P2_Session::P2_Session(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 *P2_Session::sessionNotify(yazpp_1::IPDU_Observable
  98                                                   *the_PDU_Observable, int fd)
  99 {
 100     return 0;
 101 }
 102
 103 P2_Session::~P2_Session()
 104 {
 105 }
 106
 107 void P2_Session::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 P2_Session::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 P2_Session::timeoutNotify()
 136 {
 137     failNotify();
 138 }
 139
 140 void P2_Session::connectNotify()
 141 {
 142
 143 }
 144
 145 class P2_Server : public yazpp_1::Z_Assoc {
 146 public:
 147     ~P2_Server();
 148     P2_Server(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 P2_Server::P2_Server(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 *P2_Server::sessionNotify(yazpp_1::IPDU_Observable
 177                                                  *the_PDU_Observable, int fd)
 178 {
 179     P2_Session *my = new P2_Session(the_PDU_Observable, m_thread_pool_observer,
 180                                     m_package);
 181     return my;
 182 }
 183
 184 P2_Server::~P2_Server()
 185 {
 186 }
 187
 188 void P2_Server::recv_GDU(Z_GDU *apdu, int len)
 189 {
 190 }
 191
 192 void P2_Server::failNotify()
 193 {
 194 }
 195
 196 void P2_Server::timeoutNotify()
 197 {
 198 }
 199
 200 void P2_Server::connectNotify()
 201 {
 202 }
 203
 204 FilterFrontendNet::FilterFrontendNet()
 205 {
 206     m_no_threads = 5;
 207     m_listen_address = "@:9001";
 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     yazpp_1::PDU_Assoc *my_PDU_Assoc =
 253         new yazpp_1::PDU_Assoc(&mySocketManager);
 254
 255     ThreadPoolSocketObserver threadPool(&mySocketManager, m_no_threads);
 256
 257     P2_Server z(my_PDU_Assoc, &threadPool, &package);
 258     z.server(m_listen_address.c_str());
 259
 260     while (mySocketManager.processEvent() > 0)
 261     {
 262         if (tt && tt->timeout())
 263             break;
 264     }
 265     delete tt;
 266 }
 267
 268 std::string &FilterFrontendNet::listen_address()
 269 {
 270     return m_listen_address;
 271 }
 272
 273 int &FilterFrontendNet::listen_duration()
 274 {
 275     return m_listen_duration;
 276 }
 277