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