Change ISocketObservable interface

[yazpp-moved-to-github.git] / src / yaz-pdu-assoc.cpp

diff --git a/src/yaz-pdu-assoc.cpp b/src/yaz-pdu-assoc.cpp
index 9cf7d22..50d4a79 100644 (file)
--- a/src/yaz-pdu-assoc.cpp
+++ b/src/yaz-pdu-assoc.cpp
@@ -1,5 +1,5 @@
 /* This file is part of the yazpp toolkit.
- * Copyright (C) 1998-2012 Index Data and Mike Taylor
+ * Copyright (C) Index Data 
  * See the file LICENSE for details.
  */
 
@@ -54,6 +54,7 @@ namespace yazpp_1 {
         void init(yazpp_1::ISocketObservable *socketObservable);
         COMSTACK comstack(const char *type_and_host, void **vp);
         bool m_session_is_dead;
+        char *cert_fname;
     };
 }
 
@@ -73,10 +74,12 @@ void PDU_Assoc_priv::init(ISocketObservable *socketObservable)
     idleTime = 0;
     log = YLOG_DEBUG;
     m_session_is_dead = false;
+    cert_fname = 0;
 }
 
 PDU_Assoc::~PDU_Assoc()
 {
+    xfree(m_p->cert_fname);
     delete m_p;
 }
 
@@ -99,7 +102,7 @@ PDU_Assoc::PDU_Assoc(ISocketObservable *socketObservable,
         mask |= SOCKET_OBSERVE_WRITE;
     if (cs->io_pending & CS_WANT_READ)
         mask |= SOCKET_OBSERVE_READ;
-    m_p->m_socketObservable->addObserver(cs_fileno(cs), this);
+    m_p->m_socketObservable->addObserver(this);
     if (!mask)
     {
         yaz_log(m_p->log, "new PDU_Assoc. Ready");
@@ -111,10 +114,11 @@ PDU_Assoc::PDU_Assoc(ISocketObservable *socketObservable,
         yaz_log(m_p->log, "new PDU_Assoc. Accepting");
         // assume comstack is accepting...
         m_p->state = PDU_Assoc_priv::Accepting;
-        m_p->m_socketObservable->addObserver(cs_fileno(cs), this);
+        m_p->m_socketObservable->addObserver(this);
         yaz_log(m_p->log, "maskObserver 1");
         m_p->m_socketObservable->maskObserver(this,
-                                              mask|SOCKET_OBSERVE_EXCEPT);
+                                              mask|SOCKET_OBSERVE_EXCEPT,
+                                              cs_fileno(cs));
     }
 }
 
@@ -129,17 +133,21 @@ void PDU_Assoc::socketNotify(int event)
 {
     yaz_log(m_p->log, "PDU_Assoc::socketNotify p=%p state=%d event = %d",
             this, m_p->state, event);
-    if (event & SOCKET_OBSERVE_EXCEPT)
+    if ((event & SOCKET_OBSERVE_EXCEPT) &&
+        m_p->state != PDU_Assoc_priv::Connecting)
     {
+        yaz_log(m_p->log, "PDU_Assoc::socketNotify except");
         shutdown();
         m_PDU_Observer->failNotify();
         return;
     }
     else if (event & SOCKET_OBSERVE_TIMEOUT)
     {
+        yaz_log(m_p->log, "PDU_Assoc::socketNotify timeout");
         m_PDU_Observer->timeoutNotify();
         return;
     }
+    int res;
     switch (m_p->state)
     {
     case PDU_Assoc_priv::Accepting:
@@ -165,46 +173,37 @@ void PDU_Assoc::socketNotify(int event)
             else
             {   // accept still incomplete.
                 yaz_log(m_p->log, "maskObserver 2");
-                m_p->m_socketObservable->maskObserver(this,
-                                             mask|SOCKET_OBSERVE_EXCEPT);
+                m_p->m_socketObservable->maskObserver(
+                    this,
+                    mask|SOCKET_OBSERVE_EXCEPT, cs_fileno(m_p->cs));
             }
         }
         break;
     case PDU_Assoc_priv::Connecting:
-        if (event & SOCKET_OBSERVE_READ &&
-            event & SOCKET_OBSERVE_WRITE)
+        yaz_log(m_p->log, "PDU_Assoc::socketNotify Connecting");
+        res = cs_rcvconnect(m_p->cs);
+        if (res == 1)
         {
-            // For Unix: if both read and write is set, then connect failed.
-            shutdown();
-            m_PDU_Observer->failNotify();
+            unsigned mask = SOCKET_OBSERVE_EXCEPT;
+            if (m_p->cs->io_pending & CS_WANT_WRITE)
+                mask |= SOCKET_OBSERVE_WRITE;
+            if (m_p->cs->io_pending & CS_WANT_READ)
+                mask |= SOCKET_OBSERVE_READ;
+            yaz_log(m_p->log, "maskObserver 3");
+            m_p->m_socketObservable->maskObserver(this, mask,
+                                                  cs_fileno(m_p->cs));
         }
         else
         {
-            yaz_log(m_p->log, "cs_rcvconnect");
-            int res = cs_rcvconnect(m_p->cs);
-            if (res == 1)
-            {
-                unsigned mask = SOCKET_OBSERVE_EXCEPT;
-                if (m_p->cs->io_pending & CS_WANT_WRITE)
-                    mask |= SOCKET_OBSERVE_WRITE;
-                if (m_p->cs->io_pending & CS_WANT_READ)
-                    mask |= SOCKET_OBSERVE_READ;
-                yaz_log(m_p->log, "maskObserver 3");
-                m_p->m_socketObservable->maskObserver(this, mask);
-            }
-            else
-            {
-                m_p->state = PDU_Assoc_priv::Ready;
-                if (m_PDU_Observer)
-                    m_PDU_Observer->connectNotify();
-                flush_PDU();
-            }
+            m_p->state = PDU_Assoc_priv::Ready;
+            if (m_PDU_Observer)
+                m_PDU_Observer->connectNotify();
+            flush_PDU();
         }
         break;
     case PDU_Assoc_priv::Listen:
         if (event & SOCKET_OBSERVE_READ)
         {
-            int res;
             COMSTACK new_line;
 
             if ((res = cs_listen(m_p->cs, 0, 0)) == 1)
@@ -228,15 +227,17 @@ void PDU_Assoc::socketNotify(int event)
         }
         break;
     case PDU_Assoc_priv::Writing:
+        yaz_log(m_p->log, "PDU_Assoc::socketNotify writing");
         if (event & (SOCKET_OBSERVE_READ|SOCKET_OBSERVE_WRITE))
             flush_PDU();
         break;
     case PDU_Assoc_priv::Ready:
+        yaz_log(m_p->log, "PDU_Assoc::socketNotify ready");
         if (event & (SOCKET_OBSERVE_READ|SOCKET_OBSERVE_WRITE))
         {
             do
             {
-                int res = cs_get(m_p->cs, &m_p->input_buf, &m_p->input_len);
+                res = cs_get(m_p->cs, &m_p->input_buf, &m_p->input_len);
                 if (res == 1)
                 {
                     unsigned mask = SOCKET_OBSERVE_EXCEPT;
@@ -245,7 +246,8 @@ void PDU_Assoc::socketNotify(int event)
                     if (m_p->cs->io_pending & CS_WANT_READ)
                         mask |= SOCKET_OBSERVE_READ;
                     yaz_log(m_p->log, "maskObserver 4");
-                    m_p->m_socketObservable->maskObserver(this, mask);
+                    m_p->m_socketObservable->maskObserver(this, mask,
+                                                          cs_fileno(m_p->cs));
                     return;
                 }
                 else if (res <= 0)
@@ -280,7 +282,8 @@ void PDU_Assoc::socketNotify(int event)
                 yaz_log(m_p->log, "maskObserver 5");
                 m_p->m_socketObservable->maskObserver(this,
                                                       SOCKET_OBSERVE_EXCEPT|
-                                                      SOCKET_OBSERVE_READ);
+                                                      SOCKET_OBSERVE_READ,
+                                                      cs_fileno(m_p->cs));
             }
         }
         break;
@@ -380,20 +383,22 @@ int PDU_Assoc::flush_PDU()
 {
     int r;
 
+    yaz_log(m_p->log, "PDU_Assoc::flush_PDU");
     if (m_p->state != PDU_Assoc_priv::Ready && m_p->state != PDU_Assoc_priv::Writing)
     {
-        yaz_log(m_p->log, "YAZ_PDU_Assoc::flush_PDU, not ready");
+        yaz_log(m_p->log, "PDU_Assoc::flush_PDU, not ready");
         return 1;
     }
     PDU_Assoc_priv::PDU_Queue *q = m_p->queue_out;
     if (!q)
     {
         m_p->state = PDU_Assoc_priv::Ready;
-        yaz_log(m_p->log, "YAZ_PDU_Assoc::flush_PDU queue empty");
+        yaz_log(m_p->log, "PDU_Assoc::flush_PDU queue empty");
         yaz_log(m_p->log, "maskObserver 6");
         m_p->m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
                                               SOCKET_OBSERVE_WRITE|
-                                              SOCKET_OBSERVE_EXCEPT);
+                                              SOCKET_OBSERVE_EXCEPT,
+                                              cs_fileno(m_p->cs));
         if (m_p->m_session_is_dead)
         {
             shutdown();
@@ -420,7 +425,7 @@ int PDU_Assoc::flush_PDU()
 
         mask |= SOCKET_OBSERVE_WRITE;
         yaz_log(m_p->log, "maskObserver 7");
-        m_p->m_socketObservable->maskObserver(this, mask);
+        m_p->m_socketObservable->maskObserver(this, mask, cs_fileno(m_p->cs));
         yaz_log(m_p->log, "PDU_Assoc::flush_PDU cs_put %d bytes fd=%d (inc)",
                 q->m_len, cs_fileno(m_p->cs));
         return r;
@@ -435,7 +440,8 @@ int PDU_Assoc::flush_PDU()
         m_p->state = PDU_Assoc_priv::Ready;
         yaz_log(m_p->log, "maskObserver 8");
         m_p->m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
-                                              SOCKET_OBSERVE_EXCEPT);
+                                              SOCKET_OBSERVE_EXCEPT,
+                                              cs_fileno(m_p->cs));
         if (m_p->m_session_is_dead)
             shutdown();
     }
@@ -502,6 +508,10 @@ int PDU_Assoc::listen(IPDU_Observer *observer, const char *addr)
 
     if (!m_p->cs)
         return -1;
+
+    if (m_p->cert_fname)
+        cs_set_ssl_certificate_file(m_p->cs, m_p->cert_fname);
+
     if (cs_bind(m_p->cs, ap, CS_SERVER) < 0)
         return -2;
 
@@ -514,10 +524,11 @@ int PDU_Assoc::listen(IPDU_Observer *observer, const char *addr)
         fcntl(fd, F_SETFD, oldflags);
     }
 #endif
-    m_p->m_socketObservable->addObserver(fd, this);
+    m_p->m_socketObservable->addObserver(this);
     yaz_log(m_p->log, "maskObserver 9");
     m_p->m_socketObservable->maskObserver(this, SOCKET_OBSERVE_READ|
-                                          SOCKET_OBSERVE_EXCEPT);
+                                          SOCKET_OBSERVE_EXCEPT,
+                                          cs_fileno(m_p->cs));
     yaz_log(m_p->log, "PDU_Assoc::listen ok fd=%d", fd);
     m_p->state = PDU_Assoc_priv::Listen;
     return 0;
@@ -547,7 +558,7 @@ int PDU_Assoc::connect(IPDU_Observer *observer, const char *addr)
     int res = cs_connect(m_p->cs, ap);
     yaz_log(m_p->log, "PDU_Assoc::connect fd=%d res=%d", cs_fileno(m_p->cs),
             res);
-    m_p->m_socketObservable->addObserver(cs_fileno(m_p->cs), this);
+    m_p->m_socketObservable->addObserver(this);
 
     if (res == 0)
     {   // Connect complete
@@ -556,7 +567,7 @@ int PDU_Assoc::connect(IPDU_Observer *observer, const char *addr)
         mask |= SOCKET_OBSERVE_WRITE;
         mask |= SOCKET_OBSERVE_READ;
         yaz_log(m_p->log, "maskObserver 11");
-        m_p->m_socketObservable->maskObserver(this, mask);
+        m_p->m_socketObservable->maskObserver(this, mask, cs_fileno(m_p->cs));
     }
     else if (res > 0)
     {   // Connect pending
@@ -567,7 +578,7 @@ int PDU_Assoc::connect(IPDU_Observer *observer, const char *addr)
         if (m_p->cs->io_pending & CS_WANT_READ)
             mask |= SOCKET_OBSERVE_READ;
         yaz_log(m_p->log, "maskObserver 11");
-        m_p->m_socketObservable->maskObserver(this, mask);
+        m_p->m_socketObservable->maskObserver(this, mask, cs_fileno(m_p->cs));
     }
     else
     {   // Connect failed immediately
@@ -575,7 +586,8 @@ int PDU_Assoc::connect(IPDU_Observer *observer, const char *addr)
         // normal connect in socketNotify handler
         yaz_log(m_p->log, "maskObserver 12");
         m_p->m_socketObservable->maskObserver(this, SOCKET_OBSERVE_WRITE|
-                                              SOCKET_OBSERVE_EXCEPT);
+                                              SOCKET_OBSERVE_EXCEPT,
+                                              cs_fileno(m_p->cs));
     }
     return 0;
 }
@@ -607,6 +619,15 @@ const char*PDU_Assoc::getpeername()
         return 0;
     return cs_addrstr(m_p->cs);
 }
+
+void PDU_Assoc::set_cert_fname(const char *fname)
+{
+    xfree(m_p->cert_fname);
+    m_p->cert_fname = 0;
+    if (fname)
+        m_p->cert_fname = xstrdup(fname);
+}
+
 /*
  * Local variables:
  * c-basic-offset: 4