X-Git-Url: http://jsfdemo.indexdata.com/?a=blobdiff_plain;f=src%2Ffilter_frontend_net.cpp;h=d610ceeaa42c5173b745ef53f2ad2151886a0087;hb=941d13b6a4dc2c2dd0e4bda89d380bfd8991b03b;hp=ed9a4759025b9ed14b6482e13a5582bf8384ec63;hpb=57c1c4822f1de4d983c1aee3bc6059ad2600d991;p=metaproxy-moved-to-github.git diff --git a/src/filter_frontend_net.cpp b/src/filter_frontend_net.cpp index ed9a475..d610cee 100644 --- a/src/filter_frontend_net.cpp +++ b/src/filter_frontend_net.cpp @@ -16,11 +16,11 @@ using namespace yp2; -class P2_Session : public yazpp_1::Z_Assoc { +class ZAssocServerChild : public yazpp_1::Z_Assoc { public: - ~P2_Session(); - P2_Session(yazpp_1::IPDU_Observable *the_PDU_Observable, - ThreadPoolSocketObserver *m_my_thread, + ~ZAssocServerChild(); + ZAssocServerChild(yazpp_1::IPDU_Observable *the_PDU_Observable, + ThreadPoolSocketObserver *m_thread_pool_observer, const Package *package); int m_no_requests; private: @@ -33,7 +33,7 @@ private: void timeoutNotify(); void connectNotify(); private: - ThreadPoolSocketObserver *m_my_thread; + ThreadPoolSocketObserver *m_thread_pool_observer; Session m_session; Origin m_origin; bool m_delete_flag; @@ -43,14 +43,14 @@ private: class ThreadPoolPackage : public IThreadPoolMsg { public: - ThreadPoolPackage(Package *package, P2_Session *ses) : + ThreadPoolPackage(Package *package, ZAssocServerChild *ses) : m_session(ses), m_package(package) { }; ~ThreadPoolPackage(); IThreadPoolMsg *handle(); void result(); private: - P2_Session *m_session; + ZAssocServerChild *m_session; Package *m_package; }; @@ -70,6 +70,9 @@ void ThreadPoolPackage::result() int len; m_session->send_GDU(gdu->get(), &len); } + if (m_session->m_no_requests == 0 && m_package->session().is_closed()) + delete m_session; + delete this; } IThreadPoolMsg *ThreadPoolPackage::handle() @@ -79,67 +82,71 @@ IThreadPoolMsg *ThreadPoolPackage::handle() } -P2_Session::P2_Session(yazpp_1::IPDU_Observable *the_PDU_Observable, +ZAssocServerChild::ZAssocServerChild(yazpp_1::IPDU_Observable *the_PDU_Observable, ThreadPoolSocketObserver *my_thread_pool, const Package *package) : Z_Assoc(the_PDU_Observable) { - m_my_thread = my_thread_pool; + m_thread_pool_observer = my_thread_pool; m_no_requests = 0; m_delete_flag = false; m_package = package; } -yazpp_1::IPDU_Observer *P2_Session::sessionNotify(yazpp_1::IPDU_Observable +yazpp_1::IPDU_Observer *ZAssocServerChild::sessionNotify(yazpp_1::IPDU_Observable *the_PDU_Observable, int fd) { return 0; } -P2_Session::~P2_Session() +ZAssocServerChild::~ZAssocServerChild() { } -void P2_Session::recv_GDU(Z_GDU *z_pdu, int len) +void ZAssocServerChild::recv_GDU(Z_GDU *z_pdu, int len) { m_no_requests++; Package *p = new Package(m_session, m_origin); - ThreadPoolPackage *m = new ThreadPoolPackage(p, this); + ThreadPoolPackage *tp = new ThreadPoolPackage(p, this); p->copy_filter(*m_package); p->request() = yazpp_1::GDU(z_pdu); - m_my_thread->put(m); + m_thread_pool_observer->put(tp); } -void P2_Session::failNotify() +void ZAssocServerChild::failNotify() { // TODO: send Package to signal "close" - m_delete_flag = true; - if (m_no_requests == 0) - delete this; - + if (m_session.is_closed()) + return; + m_no_requests++; + + m_session.close(); + + Package *p = new Package(m_session, m_origin); + + ThreadPoolPackage *tp = new ThreadPoolPackage(p, this); + p->copy_filter(*m_package); + m_thread_pool_observer->put(tp); } -void P2_Session::timeoutNotify() +void ZAssocServerChild::timeoutNotify() { - // TODO: send Package to signal "close" - m_delete_flag = true; - if (m_no_requests == 0) - delete this; + failNotify(); } -void P2_Session::connectNotify() +void ZAssocServerChild::connectNotify() { } -class P2_Server : public yazpp_1::Z_Assoc { +class ZAssocServer : public yazpp_1::Z_Assoc { public: - ~P2_Server(); - P2_Server(yazpp_1::IPDU_Observable *the_PDU_Observable, - ThreadPoolSocketObserver *m_my_thread, + ~ZAssocServer(); + ZAssocServer(yazpp_1::IPDU_Observable *the_PDU_Observable, + ThreadPoolSocketObserver *m_thread_pool_observer, const Package *package); private: yazpp_1::IPDU_Observer* sessionNotify( @@ -151,53 +158,53 @@ private: void timeoutNotify(); void connectNotify(); private: - ThreadPoolSocketObserver *m_my_thread; + ThreadPoolSocketObserver *m_thread_pool_observer; const Package *m_package; }; -P2_Server::P2_Server(yazpp_1::IPDU_Observable *the_PDU_Observable, - ThreadPoolSocketObserver *my_thread, +ZAssocServer::ZAssocServer(yazpp_1::IPDU_Observable *the_PDU_Observable, + ThreadPoolSocketObserver *thread_pool_observer, const Package *package) : Z_Assoc(the_PDU_Observable) { - m_my_thread = my_thread; + m_thread_pool_observer = thread_pool_observer; m_package = package; } -yazpp_1::IPDU_Observer *P2_Server::sessionNotify(yazpp_1::IPDU_Observable +yazpp_1::IPDU_Observer *ZAssocServer::sessionNotify(yazpp_1::IPDU_Observable *the_PDU_Observable, int fd) { - P2_Session *my = new P2_Session(the_PDU_Observable, m_my_thread, - m_package); + ZAssocServerChild *my = + new ZAssocServerChild(the_PDU_Observable, m_thread_pool_observer, + m_package); return my; } -P2_Server::~P2_Server() +ZAssocServer::~ZAssocServer() { } -void P2_Server::recv_GDU(Z_GDU *apdu, int len) +void ZAssocServer::recv_GDU(Z_GDU *apdu, int len) { } -void P2_Server::failNotify() +void ZAssocServer::failNotify() { } -void P2_Server::timeoutNotify() +void ZAssocServer::timeoutNotify() { } -void P2_Server::connectNotify() +void ZAssocServer::connectNotify() { } FilterFrontendNet::FilterFrontendNet() { m_no_threads = 5; - m_listen_address = "@:9001"; m_listen_duration = 0; } @@ -235,32 +242,43 @@ void My_Timer_Thread::socketNotify(int event) close(m_fd[1]); } -Package &FilterFrontendNet::process(Package &package) const { +void FilterFrontendNet::process(Package &package) const { yazpp_1::SocketManager mySocketManager; My_Timer_Thread *tt = 0; if (m_listen_duration) tt = new My_Timer_Thread(&mySocketManager, m_listen_duration); - yazpp_1::PDU_Assoc *my_PDU_Assoc = - new yazpp_1::PDU_Assoc(&mySocketManager); - ThreadPoolSocketObserver threadPool(&mySocketManager, m_no_threads); - P2_Server z(my_PDU_Assoc, &threadPool, &package); - z.server(m_listen_address.c_str()); + ZAssocServer **az = new ZAssocServer *[m_ports.size()]; + + // Create ZAssocServer for each port + size_t i; + for (i = 0; iserver(m_ports[i].c_str()); + } while (mySocketManager.processEvent() > 0) { if (tt && tt->timeout()) break; } - return package; + for (i = 0; i &FilterFrontendNet::ports() { - return m_listen_address; + return m_ports; } int &FilterFrontendNet::listen_duration()