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