src/yaz-pdu-assoc.cpp

   1 /*
   2  * Copyright (c) 1998-1999, Index Data.
   3  * See the file LICENSE for details.
   4  * Sebastian Hammer, Adam Dickmeiss
   5  *
   6  * $Log: yaz-pdu-assoc.cpp,v $
   7  * Revision 1.10  2000-08-10 08:42:42  adam
   8  * Fixes for {set,get}_APDU_log.
   9  *
  10  * Revision 1.9  1999/12/06 13:52:45  adam
  11  * Modified for new location of YAZ header files. Experimental threaded
  12  * operation.
  13  *
  14  * Revision 1.8  1999/04/28 13:04:03  adam
  15  * Fixed setting of proxy otherInfo so that database(s) are removed.
  16  *
  17  * Revision 1.7  1999/04/21 12:09:01  adam
  18  * Many improvements. Modified to proxy server to work with "sessions"
  19  * based on cookies.
  20  *
  21  * Revision 1.6  1999/04/20 10:30:05  adam
  22  * Implemented various stuff for client and proxy. Updated calls
  23  * to ODR to reflect new name parameter.
  24  *
  25  * Revision 1.5  1999/04/09 11:46:57  adam
  26  * Added object Yaz_Z_Assoc. Much more functional client.
  27  *
  28  * Revision 1.4  1999/03/23 14:17:57  adam
  29  * More work on timeout handling. Work on yaz-client.
  30  *
  31  * Revision 1.3  1999/02/02 14:01:20  adam
  32  * First WIN32 port of YAZ++.
  33  *
  34  * Revision 1.2  1999/01/28 13:08:44  adam
  35  * Yaz_PDU_Assoc better encapsulated. Memory leak fix in
  36  * yaz-socket-manager.cc.
  37  *
  38  * Revision 1.1.1.1  1999/01/28 09:41:07  adam
  39  * First implementation of YAZ++.
  40  *
  41  */
  42
  43 #include <assert.h>
  44
  45 #include <yaz-pdu-assoc.h>
  46
  47 #include <yaz/log.h>
  48 #include <yaz/tcpip.h>
  49
  50 Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable)
  51 {
  52     m_state = Closed;
  53     m_cs = 0;
  54     m_socketObservable = socketObservable;
  55     m_PDU_Observer = 0;
  56     m_queue_out = 0;
  57     m_input_buf = 0;
  58     m_input_len = 0;
  59     m_children = 0;
  60     m_parent = 0;
  61     m_next = 0;
  62     m_destroyed = 0;
  63     m_idleTime = 0;
  64     m_log = LOG_DEBUG;
  65 }
  66
  67 IYaz_PDU_Observable *Yaz_PDU_Assoc::clone()
  68 {
  69     Yaz_PDU_Assoc *copy = new Yaz_PDU_Assoc(m_socketObservable);
  70     return copy;
  71 }
  72
  73 Yaz_PDU_Assoc::~Yaz_PDU_Assoc()
  74 {
  75     destroy();
  76 }
  77
  78 void Yaz_PDU_Assoc::socketNotify(int event)
  79 {
  80     logf (m_log, "Yaz_PDU_Assoc::socketNotify p=%p event = %d", this, event);
  81     if (0 /* m_state == Connected */)
  82     {
  83         m_state = Ready;
  84         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
  85                                          YAZ_SOCKET_OBSERVE_EXCEPT);
  86         m_PDU_Observer->connectNotify();
  87         flush_PDU();
  88     }
  89     else if (m_state == Connecting)
  90     {
  91         if (event & YAZ_SOCKET_OBSERVE_READ)
  92         {
  93             close();
  94             m_PDU_Observer->failNotify();
  95         }
  96         else if (event & YAZ_SOCKET_OBSERVE_TIMEOUT)
  97         {
  98             m_PDU_Observer->timeoutNotify();
  99         }
 100         else
 101         {
 102             m_state = Ready;
 103             m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 104                                              YAZ_SOCKET_OBSERVE_EXCEPT);
 105             m_PDU_Observer->connectNotify();
 106             flush_PDU();
 107         }
 108     }
 109     else if (m_state == Listen)
 110     {
 111         if (event & YAZ_SOCKET_OBSERVE_READ)
 112         {
 113             int res;
 114             COMSTACK new_line;
 115
 116             if ((res = cs_listen(m_cs, 0, 0)) == 1)
 117                 return;
 118             if (res < 0)
 119             {
 120                 logf(LOG_FATAL, "cs_listen failed");
 121                 return;
 122             }
 123             if (!(new_line = cs_accept(m_cs)))
 124                 return;
 125
 126             /* 1. create socket-manager
 127                2. create pdu-assoc
 128                3. create top-level object
 129                     setup observer for child fileid in pdu-assoc
 130                4. start thread
 131             */
 132             int fd = cs_fileno(new_line);
 133             cs_fileno(new_line) = -1;
 134             cs_close (new_line);        /* potential problem ... */
 135 #if 1
 136             childNotify(fd);
 137 #else
 138             Yaz_PDU_Assoc *assoc = new Yaz_PDU_Assoc (m_socketObservable);
 139             assoc->m_parent = this;
 140             assoc->m_next = m_children;
 141             m_children = assoc;
 142
 143             assoc->m_PDU_Observer = m_PDU_Observer->clone(assoc);
 144             socket(fd);
 145 #endif
 146         }
 147     }
 148     else if (m_state == Ready)
 149     {
 150         if (event & YAZ_SOCKET_OBSERVE_WRITE)
 151         {
 152             flush_PDU();
 153         }
 154         if (event & YAZ_SOCKET_OBSERVE_READ)
 155         {
 156             do
 157             {
 158                 int res = cs_get (m_cs, &m_input_buf, &m_input_len);
 159                 if (res == 1)
 160                     return;
 161                 else if (res <= 0)
 162                 {
 163                     logf (m_log, "Connection closed by peer");
 164                     close();
 165                     m_PDU_Observer->failNotify();
 166                     return;
 167                 }
 168                 // lock it, so we know if recv_PDU deletes it.
 169                 int destroyed = 0;
 170                 m_destroyed = &destroyed;
 171
 172                 m_PDU_Observer->recv_PDU(m_input_buf, res);
 173                 if (destroyed)   // it really was destroyed, return now.
 174                     return;
 175             } while (m_cs && cs_more (m_cs));
 176         }
 177         if (event & YAZ_SOCKET_OBSERVE_TIMEOUT)
 178         {
 179             m_PDU_Observer->timeoutNotify();
 180         }
 181     }
 182 }
 183
 184 void Yaz_PDU_Assoc::close()
 185 {
 186     m_socketObservable->deleteObserver(this);
 187     m_state = Closed;
 188     if (m_cs)
 189     {
 190         logf (m_log, "Yaz_PDU_Assoc::close fd=%d", cs_fileno(m_cs));
 191         cs_close (m_cs);
 192     }
 193     m_cs = 0;
 194     while (m_queue_out)
 195     {
 196         PDU_Queue *q_this = m_queue_out;
 197         m_queue_out = m_queue_out->m_next;
 198         delete q_this;
 199     }
 200 //   free (m_input_buf);
 201     m_input_buf = 0;
 202     m_input_len = 0;
 203 }
 204
 205 void Yaz_PDU_Assoc::destroy()
 206 {
 207     close();
 208     if (m_destroyed)
 209         *m_destroyed = 1;
 210     Yaz_PDU_Assoc **c;
 211
 212     // delete from parent's child list (if any)
 213     if (m_parent)
 214     {
 215         c = &m_parent->m_children;
 216         while (*c != this)
 217         {
 218             assert (*c);
 219             c = &(*c)->m_next;
 220         }
 221         *c = (*c)->m_next;
 222     }
 223     // delete all children ...
 224     c = &m_children;
 225     while (*c)
 226     {
 227         Yaz_PDU_Assoc *here = *c;
 228         *c = (*c)->m_next;
 229         here->m_parent = 0;
 230         delete here;
 231     }
 232 }
 233
 234 Yaz_PDU_Assoc::PDU_Queue::PDU_Queue(const char *buf, int len)
 235 {
 236     m_buf = (char *) malloc (len);
 237     memcpy (m_buf, buf, len);
 238     m_len = len;
 239     m_next = 0;
 240 }
 241
 242 Yaz_PDU_Assoc::PDU_Queue::~PDU_Queue()
 243 {
 244     free (m_buf);
 245 }
 246
 247 int Yaz_PDU_Assoc::flush_PDU()
 248 {
 249     int r;
 250
 251     logf (m_log, "Yaz_PDU_Assoc::flush_PDU");
 252     if (m_state != Ready)
 253     {
 254         logf (m_log, "YAZ_PDU_Assoc::flush_PDU, not ready");
 255         return 1;
 256     }
 257     PDU_Queue *q = m_queue_out;
 258     if (!q)
 259     {
 260         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 261                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 262         return 0;
 263     }
 264     r = cs_put (m_cs, q->m_buf, q->m_len);
 265     if (r < 0)
 266     {
 267         close();
 268         m_PDU_Observer->failNotify();
 269         return r;
 270     }
 271     if (r == 1)
 272     {
 273         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 274                                          YAZ_SOCKET_OBSERVE_EXCEPT|
 275                                          YAZ_SOCKET_OBSERVE_WRITE);
 276         logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes (incomplete)",
 277               q->m_len);
 278         return r;
 279     }
 280     logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes", q->m_len);
 281     // whole packet sent... delete this and proceed to next ...
 282     m_queue_out = q->m_next;
 283     delete q;
 284     // don't select on write if queue is empty ...
 285     if (!m_queue_out)
 286         m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 287                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 288     return r;
 289 }
 290
 291 int Yaz_PDU_Assoc::send_PDU(const char *buf, int len)
 292 {
 293     logf (m_log, "Yaz_PDU_Assoc::send_PDU");
 294     PDU_Queue **pq = &m_queue_out;
 295     int is_idle = (*pq ? 0 : 1);
 296
 297     if (!m_cs)
 298     {
 299         logf (m_log, "Yaz_PDU_Assoc::send_PDU failed, m_cs == 0");
 300         return -1;
 301     }
 302     while (*pq)
 303         pq = &(*pq)->m_next;
 304     *pq = new PDU_Queue(buf, len);
 305     if (is_idle)
 306         return flush_PDU ();
 307     else
 308         logf (m_log, "Yaz_PDU_Assoc::cannot send_PDU fd=%d",
 309               cs_fileno(m_cs));
 310     return 0;
 311 }
 312
 313 COMSTACK Yaz_PDU_Assoc::comstack()
 314 {
 315     if (!m_cs)
 316     {
 317         CS_TYPE cs_type = tcpip_type;
 318         m_cs = cs_create (cs_type, 0, PROTO_Z3950);
 319     }
 320     return m_cs;
 321 }
 322
 323 void Yaz_PDU_Assoc::listen(IYaz_PDU_Observer *observer,
 324                            const char *addr)
 325 {
 326     close();
 327     void *ap;
 328     COMSTACK cs = comstack();
 329
 330     logf (m_log, "Yaz_PDU_Assoc::listen %s", addr);
 331     m_PDU_Observer = observer;
 332     if (!cs)
 333         return;
 334     ap = cs_straddr (cs, addr);
 335     if (!ap)
 336         return;
 337     if (cs_bind(cs, ap, CS_SERVER) < 0)
 338         return;
 339     m_socketObservable->addObserver(cs_fileno(cs), this);
 340     m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 341                                      YAZ_SOCKET_OBSERVE_EXCEPT);
 342     m_state = Listen;
 343 }
 344
 345 void Yaz_PDU_Assoc::idleTime(int idleTime)
 346 {
 347     m_idleTime = idleTime;
 348     logf (m_log, "Yaz_PDU_Assoc::idleTime(%d)", idleTime);
 349     m_socketObservable->timeoutObserver(this, m_idleTime);
 350 }
 351
 352 void Yaz_PDU_Assoc::connect(IYaz_PDU_Observer *observer,
 353                             const char *addr)
 354 {
 355     logf (m_log, "Yaz_PDU_Assoc::connect %s", addr);
 356     close();
 357     m_PDU_Observer = observer;
 358     COMSTACK cs = comstack();
 359     void *ap = cs_straddr (cs, addr);
 360     if (!ap)
 361     {
 362         logf (m_log, "cs_straddr failed");
 363         return;
 364     }
 365     int res = cs_connect (cs, ap);
 366     logf (m_log, "Yaz_PDU_Assoc::connect fd=%d res=%d", cs_fileno(cs), res);
 367     m_socketObservable->addObserver(cs_fileno(cs), this);
 368     m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ|
 369                                            YAZ_SOCKET_OBSERVE_EXCEPT|
 370                                            YAZ_SOCKET_OBSERVE_WRITE);
 371     m_state = Connecting;
 372 }
 373
 374 void Yaz_PDU_Assoc::socket(IYaz_PDU_Observer *observer, int fd)
 375 {
 376     close();
 377     m_PDU_Observer = observer;
 378     if (fd >= 0)
 379     {
 380         CS_TYPE cs_type = tcpip_type;
 381         m_cs = cs_createbysocket(fd, cs_type, 0, PROTO_Z3950);
 382         m_state = Ready;
 383         m_socketObservable->addObserver(fd, this);
 384         m_socketObservable->maskObserver(this,
 385                                          YAZ_SOCKET_OBSERVE_READ|
 386                                          YAZ_SOCKET_OBSERVE_EXCEPT);
 387         m_socketObservable->timeoutObserver(this, m_idleTime);
 388     }
 389 }
 390
 391 #if 1
 392 void Yaz_PDU_Assoc::childNotify(int fd)
 393 {
 394     /// Clone PDU Observable (keep socket manager)
 395     IYaz_PDU_Observable *new_observable = clone();
 396
 397     /// Clone PDU Observer
 398     IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable);
 399
 400     /// Attach new socket to it
 401     new_observable->socket(observer, fd);
 402 }
 403 #else
 404
 405 #include <yaz-socket-manager.h>
 406 #include <pthread.h>
 407
 408 class thread_info {
 409     Yaz_SocketManager *socketManager;
 410     IYaz_PDU_Observable *
 411
 412 };
 413
 414 static void *events(void *p)
 415 {
 416     Yaz_SocketManager *s = (Yaz_SocketManager *) p;
 417
 418     while (s->processEvent() > 0)
 419         ;
 420     return 0;
 421 }
 422
 423 void Yaz_PDU_Assoc::childNotify(int fd)
 424 {
 425     Yaz_SocketManager *socket_observable = new Yaz_SocketManager;
 426     IYaz_PDU_Observable *new_observable = clone();
 427
 428     m_socketObservable = socket_observable;
 429
 430     /// Clone PDU Observer
 431     IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable);
 432
 433     /// Attach new socket to it
 434     new_observable->socket(observer, fd);
 435
 436     pthread_t type;
 437
 438     int id = pthread_create (&type, 0, events, socket_observable);
 439     logf (LOG_LOG, "pthread_create returned id=%d", id);
 440 }
 441 #endif
 442