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