src/filter_z3950_client.cpp

   1 /* $Id: filter_z3950_client.cpp,v 1.11 2005-11-10 23:10:42 adam Exp $
   2    Copyright (c) 2005, Index Data.
   3
   4 %LICENSE%
   5  */
   6
   7 #include "config.hpp"
   8
   9 #include "filter.hpp"
  10 #include "router.hpp"
  11 #include "package.hpp"
  12 #include "util.hpp"
  13 #include "filter_z3950_client.hpp"
  14
  15 #include <map>
  16 #include <stdexcept>
  17 #include <list>
  18 #include <iostream>
  19
  20 #include <boost/thread/mutex.hpp>
  21 #include <boost/thread/condition.hpp>
  22
  23 #include <yaz/zgdu.h>
  24 #include <yaz/log.h>
  25 #include <yaz/otherinfo.h>
  26 #include <yaz/diagbib1.h>
  27
  28 #include <yaz++/socket-manager.h>
  29 #include <yaz++/pdu-assoc.h>
  30 #include <yaz++/z-assoc.h>
  31
  32 namespace yf = yp2::filter;
  33
  34 namespace yp2 {
  35     namespace filter {
  36         class Z3950Client::Assoc : public yazpp_1::Z_Assoc{
  37             friend class Rep;
  38             Assoc(yazpp_1::SocketManager *socket_manager,
  39                   yazpp_1::IPDU_Observable *PDU_Observable,
  40                   std::string host);
  41             ~Assoc();
  42             void connectNotify();
  43             void failNotify();
  44             void timeoutNotify();
  45             void recv_GDU(Z_GDU *gdu, int len);
  46             yazpp_1::IPDU_Observer* sessionNotify(
  47                 yazpp_1::IPDU_Observable *the_PDU_Observable,
  48                 int fd);
  49
  50             yazpp_1::SocketManager *m_socket_manager;
  51             yazpp_1::IPDU_Observable *m_PDU_Observable;
  52             Package *m_package;
  53             bool m_in_use;
  54             bool m_waiting;
  55             bool m_connected;
  56             std::string m_host;
  57         };
  58
  59         class Z3950Client::Rep {
  60         public:
  61             boost::mutex m_mutex;
  62             boost::condition m_cond_session_ready;
  63             std::map<yp2::Session,Z3950Client::Assoc *> m_clients;
  64             Z3950Client::Assoc *get_assoc(Package &package);
  65             void send_and_receive(Package &package,
  66                                   yf::Z3950Client::Assoc *c);
  67             void release_assoc(Package &package);
  68         };
  69     }
  70 }
  71
  72 using namespace yp2;
  73
  74 yf::Z3950Client::Assoc::Assoc(yazpp_1::SocketManager *socket_manager,
  75                               yazpp_1::IPDU_Observable *PDU_Observable,
  76                               std::string host)
  77     :  Z_Assoc(PDU_Observable),
  78        m_socket_manager(socket_manager), m_PDU_Observable(PDU_Observable),
  79        m_package(0), m_in_use(true), m_waiting(false), m_connected(false),
  80        m_host(host)
  81 {
  82     // std::cout << "create assoc " << this << "\n";
  83 }
  84
  85 yf::Z3950Client::Assoc::~Assoc()
  86 {
  87     // std::cout << "destroy assoc " << this << "\n";
  88 }
  89
  90 void yf::Z3950Client::Assoc::connectNotify()
  91 {
  92     m_waiting = false;
  93
  94     m_connected = true;
  95 }
  96
  97 void yf::Z3950Client::Assoc::failNotify()
  98 {
  99     m_waiting = false;
 100
 101     yp2::odr odr;
 102
 103     if (m_package)
 104     {
 105         m_package->response() = odr.create_close(Z_Close_peerAbort, 0);
 106         m_package->session().close();
 107     }
 108 }
 109
 110 void yf::Z3950Client::Assoc::timeoutNotify()
 111 {
 112     m_waiting = false;
 113
 114     yp2::odr odr;
 115
 116     if (m_package)
 117     {
 118         m_package->response() = odr.create_close(Z_Close_lackOfActivity, 0);
 119         m_package->session().close();
 120     }
 121 }
 122
 123 void yf::Z3950Client::Assoc::recv_GDU(Z_GDU *gdu, int len)
 124 {
 125     m_waiting = false;
 126
 127     if (m_package)
 128         m_package->response() = gdu;
 129 }
 130
 131 yazpp_1::IPDU_Observer *yf::Z3950Client::Assoc::sessionNotify(
 132     yazpp_1::IPDU_Observable *the_PDU_Observable,
 133     int fd)
 134 {
 135     return 0;
 136 }
 137
 138
 139 yf::Z3950Client::Z3950Client() :  m_p(new yf::Z3950Client::Rep)
 140 {
 141 }
 142
 143 yf::Z3950Client::~Z3950Client() {
 144 }
 145
 146 yf::Z3950Client::Assoc *yf::Z3950Client::Rep::get_assoc(Package &package)
 147 {
 148     // only one thread messes with the clients list at a time
 149     boost::mutex::scoped_lock lock(m_mutex);
 150
 151     std::map<yp2::Session,yf::Z3950Client::Assoc *>::iterator it;
 152
 153     while(true)
 154     {
 155         it = m_clients.find(package.session());
 156         if (it == m_clients.end())
 157             break;
 158
 159         if (!it->second->m_in_use)
 160         {
 161             it->second->m_in_use = true;
 162             return it->second;
 163         }
 164         m_cond_session_ready.wait(lock);
 165     }
 166
 167     // only deal with Z39.50
 168     Z_GDU *gdu = package.request().get();
 169
 170     if (!gdu || gdu->which != Z_GDU_Z3950)
 171     {
 172         package.move();
 173         return 0;
 174     }
 175     Z_APDU *apdu = gdu->u.z3950;
 176
 177     // new Z39.50 session ..
 178
 179     // check that it is init. If not, close
 180     if (apdu->which != Z_APDU_initRequest)
 181     {
 182         yp2::odr odr;
 183
 184         package.response() = odr.create_close(Z_Close_protocolError,
 185                                               "no init request for session");
 186         package.session().close();
 187         return 0;
 188     }
 189     // check virtual host
 190     const char *vhost =
 191         yaz_oi_get_string_oidval(&apdu->u.initRequest->otherInfo,
 192                                  VAL_PROXY, 1, 0);
 193     if (!vhost)
 194     {
 195         yp2::odr odr;
 196         package.response() = odr.create_initResponse(
 197             YAZ_BIB1_INIT_NEGOTIATION_OPTION_REQUIRED,
 198             "Virtual host not given");
 199
 200         package.session().close();
 201         return 0;
 202     }
 203
 204     yazpp_1::SocketManager *sm = new yazpp_1::SocketManager;
 205     yazpp_1::PDU_Assoc *pdu_as = new yazpp_1::PDU_Assoc(sm);
 206     yf::Z3950Client::Assoc *as = new yf::Z3950Client::Assoc(sm, pdu_as, vhost);
 207     m_clients[package.session()] = as;
 208     return as;
 209 }
 210
 211 void yf::Z3950Client::Rep::send_and_receive(Package &package,
 212                                             yf::Z3950Client::Assoc *c)
 213 {
 214     Z_GDU *gdu = package.request().get();
 215
 216     if (!gdu || gdu->which != Z_GDU_Z3950)
 217         return;
 218
 219     c->m_package = &package;
 220     c->m_waiting = true;
 221     if (!c->m_connected)
 222     {
 223         c->client(c->m_host.c_str());
 224
 225         while (c->m_waiting && c->m_socket_manager->processEvent() > 0)
 226             ;
 227     }
 228     if (!c->m_connected)
 229     {
 230         return;
 231     }
 232
 233     // prepare response
 234     c->m_waiting = true;
 235
 236     // relay the package  ..
 237     int len;
 238     c->send_GDU(gdu, &len);
 239
 240     while (c->m_waiting && c->m_socket_manager->processEvent() > 0)
 241         ;
 242 }
 243
 244 void yf::Z3950Client::Rep::release_assoc(Package &package)
 245 {
 246     boost::mutex::scoped_lock lock(m_mutex);
 247     std::map<yp2::Session,yf::Z3950Client::Assoc *>::iterator it;
 248
 249     it = m_clients.find(package.session());
 250     if (it != m_clients.end())
 251     {
 252         if (package.session().is_closed())
 253         {
 254             // the Z_Assoc and PDU_Assoc must be destroyed before
 255             // the socket manager.. so pull that out.. first..
 256             yazpp_1::SocketManager *s = it->second->m_socket_manager;
 257             delete it->second;  // destroy Z_Assoc
 258             delete s;    // then manager
 259             m_clients.erase(it);
 260         }
 261         else
 262         {
 263             it->second->m_in_use = false;
 264         }
 265         m_cond_session_ready.notify_all();
 266     }
 267 }
 268
 269 void yf::Z3950Client::process(Package &package) const
 270 {
 271     yf::Z3950Client::Assoc *c = m_p->get_assoc(package);
 272     if (c)
 273     {
 274         m_p->send_and_receive(package, c);
 275     }
 276     m_p->release_assoc(package);
 277 }
 278
 279
 280 /*
 281  * Local variables:
 282  * c-basic-offset: 4
 283  * indent-tabs-mode: nil
 284  * c-file-style: "stroustrup"
 285  * End:
 286  * vim: shiftwidth=4 tabstop=8 expandtab
 287  */