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