src/yaz-pdu-assoc.cpp

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