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