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