Fixed problem processEvent which in rare cases return 0.
[yazpp-moved-to-github.git] / src / yaz-proxy.cpp
index beff26c..a73c95a 100644 (file)
@@ -1,34 +1,15 @@
 /*
- * Copyright (c) 1998-1999, Index Data.
+ * Copyright (c) 1998-2001, Index Data.
  * See the file LICENSE for details.
- * Sebastian Hammer, Adam Dickmeiss
  * 
- * $Log: yaz-proxy.cpp,v $
- * Revision 1.5  1999-04-21 12:09:01  adam
- * Many improvements. Modified to proxy server to work with "sessions"
- * based on cookies.
- *
- * Revision 1.4  1999/04/20 10:30:05  adam
- * Implemented various stuff for client and proxy. Updated calls
- * to ODR to reflect new name parameter.
- *
- * Revision 1.3  1999/04/09 11:46:57  adam
- * Added object Yaz_Z_Assoc. Much more functional client.
- *
- * Revision 1.2  1999/01/28 13:08:46  adam
- * Yaz_PDU_Assoc better encapsulated. Memory leak fix in
- * yaz-socket-manager.cc.
- *
- * Revision 1.1.1.1  1999/01/28 09:41:07  adam
- * First implementation of YAZ++.
- *
+ * $Id: yaz-proxy.cpp,v 1.28 2001-11-04 22:36:21 adam Exp $
  */
 
 #include <assert.h>
+#include <time.h>
 
-#include <log.h>
-
-#include <yaz-proxy.h>
+#include <yaz/log.h>
+#include <yaz++/yaz-proxy.h>
 
 Yaz_Proxy::Yaz_Proxy(IYaz_PDU_Observable *the_PDU_Observable) :
     Yaz_Z_Assoc(the_PDU_Observable)
@@ -39,18 +20,32 @@ Yaz_Proxy::Yaz_Proxy(IYaz_PDU_Observable *the_PDU_Observable) :
     m_clientPool = 0;
     m_seqno = 1;
     m_keepalive = 1;
+    m_proxyTarget = 0;
+    m_max_clients = 50;
+    m_seed = time(0);
 }
 
 Yaz_Proxy::~Yaz_Proxy()
 {
+    xfree (m_proxyTarget);
+}
+
+void Yaz_Proxy::set_proxyTarget(const char *target)
+{
+    xfree (m_proxyTarget);
+    m_proxyTarget = 0;
+    if (target)
+       m_proxyTarget = (char *) xstrdup (target);
 }
 
-IYaz_PDU_Observer *Yaz_Proxy::clone(IYaz_PDU_Observable
-                                   *the_PDU_Observable)
+IYaz_PDU_Observer *Yaz_Proxy::sessionNotify(IYaz_PDU_Observable
+                                           *the_PDU_Observable, int fd)
 {
     Yaz_Proxy *new_proxy = new Yaz_Proxy(the_PDU_Observable);
     new_proxy->m_parent = this;
-    new_proxy->timeout(120);
+    new_proxy->timeout(500);
+    new_proxy->set_proxyTarget(m_proxyTarget);
+    new_proxy->set_APDU_log(get_APDU_log());
     return new_proxy;
 }
 
@@ -65,7 +60,7 @@ char *Yaz_Proxy::get_cookie(Z_OtherInformation **otherInfo)
     assert (oid_ent_to_oid (&ent, oid));
 
     if (oid_ent_to_oid (&ent, oid) && 
-       (oi = update_otherInformation(otherInfo, 0, oid, 1)) &&
+       (oi = update_otherInformation(otherInfo, 0, oid, 1, 1)) &&
        oi->which == Z_OtherInfo_characterInfo)
        return oi->information.characterInfo;
     return 0;
@@ -80,7 +75,7 @@ char *Yaz_Proxy::get_proxy(Z_OtherInformation **otherInfo)
     ent.oclass = CLASS_USERINFO;
     ent.value = (oid_value) VAL_PROXY;
     if (oid_ent_to_oid (&ent, oid) &&
-       (oi = update_otherInformation(otherInfo, 0, oid, 1)) &&
+       (oi = update_otherInformation(otherInfo, 0, oid, 1, 1)) &&
        oi->which == Z_OtherInfo_characterInfo)
        return oi->information.characterInfo;
     return 0;
@@ -95,52 +90,234 @@ Yaz_ProxyClient *Yaz_Proxy::get_client(Z_APDU *apdu)
     
     get_otherInfoAPDU(apdu, &oi);
     char *cookie = get_cookie(oi);
-    logf (LOG_LOG, "Yaz_Proxy::get_client cookie=%s", cookie ? cookie :
-         "<null>");
-    if (cookie)
+    yaz_log (LOG_LOG, "Yaz_Proxy::get_client cookie=%s", cookie ? cookie :
+         "null");
+
+    const char *proxy_host = get_proxy(oi);
+    if (proxy_host)
+       set_proxyTarget(proxy_host);
+    yaz_log (LOG_LOG, "proxy_host = %s", m_proxyTarget ? m_proxyTarget:"none");
+    
+    // no target specified at all?
+    if (!m_proxyTarget)
+       return 0;
+
+    if (cookie && *cookie)
     {
+       yaz_log (LOG_LOG, "lookup of clients cookie=%s target=%s",
+             cookie, m_proxyTarget);
+       Yaz_ProxyClient *cc = 0;
+       
        for (c = parent->m_clientPool; c; c = c->m_next)
        {
+           yaz_log (LOG_LOG, " found client cookie = %s target=%s seqno=%d",
+                 c->m_cookie, c->get_hostname(), c->m_seqno);
            assert (c->m_prev);
            assert (*c->m_prev == c);
-           if (!strcmp(cookie,c->m_cookie))
+           if (!strcmp(cookie,c->m_cookie) &&
+               !strcmp(m_proxyTarget, c->get_hostname()))
+           {
+               yaz_log (LOG_LOG, "found!");
+               cc = c;
+           }
+       }
+       if (cc)
+       {
+           // found it in cache
+           c = cc;
+           // The following handles "cancel"
+           // If connection is busy (waiting for PDU) and
+           // we have an initRequest we can safely do re-open
+           if (c->m_waiting && apdu->which == Z_APDU_initRequest)
            {
-               logf (LOG_LOG, "Yaz_Proxy::get_client cached");
-               break;
+               yaz_log (LOG_LOG, "reopen target=%s", c->get_hostname());
+               c->close();
+               c->client(m_proxyTarget);
+               c->m_init_flag = 0;
+               
+               delete c->m_last_query;
+               c->m_last_query = 0;
+               c->m_last_resultCount = 0;
+               c->m_sr_transform = 0;
+               c->m_waiting = 0;
+               c->timeout(600); 
            }
+           c->m_seqno = parent->m_seqno;
+           if (c->m_server && c->m_server != this)
+               c->m_server->m_client = 0;
+           c->m_server = this;
+           c->m_seqno = parent->m_seqno;
+           (parent->m_seqno)++;
+           yaz_log (LOG_LOG, "get_client 1 %p %p", this, c);
+           return c;
        }
     }
-    else if (!m_client)
+    if (!m_client)
     {
-       logf (LOG_LOG, "Yaz_Proxy::get_client creating new");
-       c = new Yaz_ProxyClient(m_PDU_Observable->clone());
-       c->m_next = parent->m_clientPool;
-       if (c->m_next)
-           c->m_next->m_prev = &c->m_next;
-       parent->m_clientPool = c;
-       c->m_prev = &parent->m_clientPool;
-
-       sprintf (c->m_cookie, "%d", parent->m_seqno);
-       (parent->m_seqno)++;
-
-       if (apdu->which == Z_APDU_initRequest)
+       if (apdu->which != Z_APDU_initRequest)
        {
-           logf (LOG_LOG, "got InitRequest");
+           yaz_log (LOG_LOG, "no first INIT!");
+           return 0;
+       }
+       yaz_log (LOG_LOG, "got InitRequest");
            
-           char *proxy_host = get_proxy(&apdu->u.initRequest->otherInfo);
-           if (proxy_host)
-               c->client(proxy_host);
+       // go through list of clients - and find the lowest/oldest one.
+       Yaz_ProxyClient *c_min = 0;
+       int min_seq = -1;
+       int no_of_clients = 0;
+       for (c = parent->m_clientPool; c; c = c->m_next)
+       {
+           no_of_clients++;
+           if (min_seq < 0 || c->m_seqno < min_seq)
+           {
+               min_seq = c->m_seqno;
+               c_min = c;
+           }
+       }
+       if (no_of_clients >= parent->m_max_clients)
+       {
+           c = c_min;
+           if (c->m_waiting || strcmp(m_proxyTarget, c->get_hostname()))
+           {
+               yaz_log (LOG_LOG, "Yaz_Proxy::get_client re-init session %d",
+                     c->m_seqno);
+               if (c->m_server && c->m_server != this)
+                   delete c->m_server;
+               c->m_server = 0;
+           }
            else
-               c->client("localhost:9999");
+           {
+               yaz_log (LOG_LOG,
+                        "Yaz_Proxy::get_client re-use session %d to %d",
+                     c->m_seqno, parent->m_seqno);
+               if (cookie)
+                   strcpy (c->m_cookie, cookie);
+               else
+                   c->m_cookie[0] = '\0';
+               c->m_seqno = parent->m_seqno;
+               if (c->m_server && c->m_server != this)
+               {
+                   c->m_server->m_client = 0;
+                   delete c->m_server;
+               }
+               (parent->m_seqno)++;
+               yaz_log (LOG_LOG, "get_client 2 %p %p", this, c);
+               return c;
+           }
        }
-       c->timeout(600);
+       else
+       {
+           yaz_log (LOG_LOG, "Yaz_Proxy::get_client making session %d",
+                 parent->m_seqno);
+           c = new Yaz_ProxyClient(m_PDU_Observable->clone());
+           c->m_next = parent->m_clientPool;
+           if (c->m_next)
+               c->m_next->m_prev = &c->m_next;
+           parent->m_clientPool = c;
+           c->m_prev = &parent->m_clientPool;
+       }
+       if (cookie)
+           strcpy (c->m_cookie, cookie);
+       else
+           c->m_cookie[0] = '\0';
+       yaz_log (LOG_LOG, "Yaz_Proxy::get_client connect to %s",
+                m_proxyTarget);
+       c->m_seqno = parent->m_seqno;
+       c->client(m_proxyTarget);
+       c->m_init_flag = 0;
+
+       delete c->m_last_query;
+       c->m_last_query = 0;
+       c->m_last_resultCount = 0;
+       c->m_sr_transform = 0;
+       c->m_waiting = 0;
+       c->timeout(10);
+
+       (parent->m_seqno)++;
     }
+    yaz_log (LOG_LOG, "get_client 3 %p %p", this, c);
     return c;
 }
 
+Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
+{
+    if (apdu->which != Z_APDU_searchRequest)
+       return apdu;
+    Z_SearchRequest *sr = apdu->u.searchRequest;
+    Yaz_Z_Query *this_query = new Yaz_Z_Query;
+    
+    this_query->set_Z_Query(sr->query);
+    
+    if (m_client->m_last_query &&
+       m_client->m_last_query->match(this_query))
+    {
+       delete this_query;
+       if (m_client->m_last_resultCount > *sr->smallSetUpperBound &&
+           m_client->m_last_resultCount < *sr->largeSetLowerBound)
+       {
+           // medium Set
+           yaz_log (LOG_LOG, "Yaz_Proxy::result_set_optimize medium set");
+           Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentRequest);
+           Z_PresentRequest *pr = new_apdu->u.presentRequest;
+           pr->referenceId = sr->referenceId;
+           pr->resultSetId = sr->resultSetName;
+           pr->preferredRecordSyntax = sr->preferredRecordSyntax;
+           *pr->numberOfRecordsRequested = *sr->mediumSetPresentNumber;
+           if (sr->mediumSetElementSetNames)
+           {
+               pr->recordComposition = (Z_RecordComposition *)
+                   odr_malloc(odr_encode(), sizeof(Z_RecordComposition));
+               pr->recordComposition->which = Z_RecordComp_simple;
+               pr->recordComposition->u.simple = sr->mediumSetElementSetNames;
+           }
+           m_client->m_sr_transform = 1;
+           return new_apdu;
+       }
+       else if (m_client->m_last_resultCount >= *sr->largeSetLowerBound ||
+           m_client->m_last_resultCount == 0)
+       {
+           // large set
+           yaz_log (LOG_LOG, "Yaz_Proxy::result_set_optimize large set");
+           Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
+           new_apdu->u.searchResponse->referenceId = sr->referenceId;
+           new_apdu->u.searchResponse->resultCount =
+               &m_client->m_last_resultCount;
+           send_Z_PDU(new_apdu);
+           return 0;
+       }
+       else
+       {
+           // small set
+           yaz_log (LOG_LOG, "Yaz_Proxy::result_set_optimize small set");
+           Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentRequest);
+           Z_PresentRequest *pr = new_apdu->u.presentRequest;
+           pr->referenceId = sr->referenceId;
+           pr->resultSetId = sr->resultSetName;
+           pr->preferredRecordSyntax = sr->preferredRecordSyntax;
+           *pr->numberOfRecordsRequested = m_client->m_last_resultCount;
+           if (sr->smallSetElementSetNames)
+           {
+               pr->recordComposition = (Z_RecordComposition *)
+                   odr_malloc(odr_encode(), sizeof(Z_RecordComposition));
+               pr->recordComposition->which = Z_RecordComp_simple;
+               pr->recordComposition->u.simple = sr->smallSetElementSetNames;
+           }
+           m_client->m_sr_transform = 1;
+           return new_apdu;
+       }
+    }
+    else
+    {
+       yaz_log (LOG_LOG, "Yaz_Proxy::result_set_optimize new set");
+       delete m_client->m_last_query;
+       m_client->m_last_query = this_query;
+    }
+    return apdu;
+}
+
 void Yaz_Proxy::recv_Z_PDU(Z_APDU *apdu)
 {
-    logf (LOG_LOG, "Yaz_Proxy::recv_Z_PDU");
+    yaz_log (LOG_LOG, "Yaz_Proxy::recv_Z_PDU");
     // Determine our client.
     m_client = get_client(apdu);
     if (!m_client)
@@ -150,42 +327,102 @@ void Yaz_Proxy::recv_Z_PDU(Z_APDU *apdu)
     }
     m_client->m_server = this;
 
+    if (apdu->which == Z_APDU_initRequest)
+    {
+       if (m_client->m_init_flag)
+       {
+           Z_APDU *apdu = create_Z_PDU(Z_APDU_initResponse);
+           if (m_client->m_cookie)
+               set_otherInformationString(apdu, VAL_COOKIE, 1,
+                                          m_client->m_cookie);
+           send_Z_PDU(apdu);
+           return;
+       }
+       m_client->m_init_flag = 1;
+    }
+    apdu = result_set_optimize(apdu);
+    if (!apdu)
+       return;
+
+    yaz_log (LOG_LOG, "Yaz_ProxyClient::send_Z_PDU %s",
+            m_client->get_hostname());
+
+    // delete other info part from PDU before sending to target
     Z_OtherInformation **oi;
     get_otherInfoAPDU(apdu, &oi);
-    *oi = 0;
-    logf (LOG_LOG, "Yaz_ProxyClient::send_Z_PDU");
+    if (oi)
+        *oi = 0;
+
     if (m_client->send_Z_PDU(apdu) < 0)
     {
        delete m_client;
+       m_client = 0;
        delete this;
     }
+    else
+       m_client->m_waiting = 1;
 }
 
-void Yaz_Proxy::failNotify()
+void Yaz_Proxy::connectNotify()
 {
-    logf (LOG_LOG, "failNotity server");
-    if (m_keepalive)
+}
+
+void Yaz_Proxy::shutdown()
+{
+    yaz_log (LOG_LOG, "shutdown (client to proxy)");
+    // only keep if keep_alive flag and cookie is set...
+    if (m_keepalive && m_client && m_client->m_cookie[0])
     {
-       // Tell client (if any) that not server connection is there..
-       if (m_client)
-           m_client->m_server = 0;
+       yaz_log (LOG_LOG, "shutdown - keepalive this=%p, m_server=%p",
+            this, m_client->m_server);
+       if (m_client->m_waiting == 2)
+           abort();
+       // Tell client (if any) that no server connection is there..
+       m_client->m_server = 0;
     }
-    else
+    else if (m_client)
     {
+       yaz_log (LOG_LOG, "deleting %p %p", this, m_client);
+       if (m_client->m_waiting == 2)
+           abort();
        delete m_client;
     }
+    else if (!m_parent)
+    {
+       yaz_log (LOG_LOG, "abort %p", this);
+        abort();
+    }
     delete this;
 }
 
-void Yaz_ProxyClient::failNotify()
+void Yaz_ProxyClient::shutdown()
 {
-    logf (LOG_LOG, "failNotity client");
+    yaz_log (LOG_LOG, "shutdown (proxy to server) %s", get_hostname());
     delete m_server;
     delete this;
 }
 
-IYaz_PDU_Observer *Yaz_ProxyClient::clone(IYaz_PDU_Observable
-                                         *the_PDU_Observable)
+void Yaz_Proxy::failNotify()
+{
+    yaz_log (LOG_LOG, "connection closed by client");
+    shutdown();
+}
+
+void Yaz_ProxyClient::failNotify()
+{
+    yaz_log (LOG_LOG, "Yaz_ProxyClient connection closed by %s", get_hostname());
+    shutdown();
+}
+
+void Yaz_ProxyClient::connectNotify()
+{
+    yaz_log (LOG_LOG, "Yaz_ProxyClient connection accepted by %s",
+         get_hostname());
+    timeout(600);
+}
+
+IYaz_PDU_Observer *Yaz_ProxyClient::sessionNotify(IYaz_PDU_Observable
+                                                 *the_PDU_Observable, int fd)
 {
     return new Yaz_ProxyClient(the_PDU_Observable);
 }
@@ -193,21 +430,23 @@ IYaz_PDU_Observer *Yaz_ProxyClient::clone(IYaz_PDU_Observable
 Yaz_ProxyClient::~Yaz_ProxyClient()
 {
     if (m_prev)
-    {
        *m_prev = m_next;
-       if (m_next)
-           m_next->m_prev = m_prev;
-    }
+    if (m_next)
+       m_next->m_prev = m_prev;
+    m_waiting = 2;     // for debugging purposes only.
+    delete m_last_query;
 }
 
 void Yaz_Proxy::timeoutNotify()
 {
-    failNotify();
+    yaz_log (LOG_LOG, "timeout (client to proxy)");
+    shutdown();
 }
 
 void Yaz_ProxyClient::timeoutNotify()
 {
-    failNotify();
+    yaz_log (LOG_LOG, "timeout (proxy to target) %s", get_hostname());
+    shutdown();
 }
 
 Yaz_ProxyClient::Yaz_ProxyClient(IYaz_PDU_Observable *the_PDU_Observable) :
@@ -216,16 +455,47 @@ Yaz_ProxyClient::Yaz_ProxyClient(IYaz_PDU_Observable *the_PDU_Observable) :
     m_cookie[0] = 0;
     m_next = 0;
     m_prev = 0;
+    m_init_flag = 0;
+    m_last_query = 0;
+    m_last_resultCount = 0;
+    m_sr_transform = 0;
+    m_waiting = 0;
 }
 
 void Yaz_ProxyClient::recv_Z_PDU(Z_APDU *apdu)
 {
-    logf (LOG_LOG, "Yaz_ProxyClient::recv_Z_PDU");
+    m_waiting = 0;
+    yaz_log (LOG_LOG, "Yaz_ProxyClient::recv_Z_PDU %s", get_hostname());
+    if (apdu->which == Z_APDU_searchResponse)
+    {
+       m_last_resultCount = *apdu->u.searchResponse->resultCount;
+       int status = *apdu->u.searchResponse->searchStatus;
+       if (! status || (
+               apdu->u.searchResponse->records &&
+               apdu->u.searchResponse->records->which != Z_Records_DBOSD))
+       {
+           delete m_last_query;
+           m_last_query = 0;
+       }
+    }
+    if (apdu->which == Z_APDU_presentResponse && m_sr_transform)
+    {
+       m_sr_transform = 0;
+       Z_PresentResponse *pr = apdu->u.presentResponse;
+       Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
+       Z_SearchResponse *sr = new_apdu->u.searchResponse;
+       sr->referenceId = pr->referenceId;
+       *sr->resultCount = m_last_resultCount;
+       sr->records = pr->records;
+       sr->nextResultSetPosition = pr->nextResultSetPosition;
+       sr->numberOfRecordsReturned = pr->numberOfRecordsReturned;
+       apdu = new_apdu;
+    }
     if (m_cookie)
        set_otherInformationString (apdu, VAL_COOKIE, 1, m_cookie);
     if (m_server)
     {
-       logf (LOG_LOG, "Yaz_Proxy::send_Z_PDU");
+       yaz_log (LOG_LOG, "Yaz_Proxy::send_Z_PDU");
        m_server->send_Z_PDU(apdu);
     }
 }