X-Git-Url: http://lists.indexdata.com/cgi-bin?a=blobdiff_plain;f=src%2Ffilter_multi.cpp;h=778a527a1e2dc5509341a645c87492dc39d924bf;hb=f3de0c53e1a8f13a2c0b7c1a7a1a45cfa4a7f4e7;hp=09f5f2f60a3bfc893c3df332ac8fff9b6dc28b0b;hpb=b0e3c437bffcf99c6f5a3ceb77d661cc366be05f;p=metaproxy-moved-to-github.git

diff --git a/src/filter_multi.cpp b/src/filter_multi.cpp
index 09f5f2f..778a527 100644
--- a/src/filter_multi.cpp
+++ b/src/filter_multi.cpp
@@ -1,4 +1,4 @@
-/* $Id: filter_multi.cpp,v 1.1 2006-01-15 20:03:14 adam Exp $
+/* $Id: filter_multi.cpp,v 1.4 2006-01-17 13:34:51 adam Exp $
    Copyright (c) 2005, Index Data.
 
 %LICENSE%
@@ -31,12 +31,20 @@ namespace yp2 {
 
         struct Multi::BackendSet {
             BackendPtr m_backend;
-            long size;
+            int m_count;
+            bool operator < (const BackendSet &k) const;
         };
-        struct Multi::Set {
-            Set(std::string setname);
-            Set();
-            ~Set();
+        struct Multi::FrontendSet {
+            struct PresentJob {
+                BackendPtr m_backend;
+                int m_pos;
+                int m_inside_pos;
+            };
+            FrontendSet(std::string setname);
+            FrontendSet();
+            ~FrontendSet();
+
+            void round_robin(int pos, int number, std::list<PresentJob> &job);
 
             std::list<BackendSet> m_backend_sets;
             std::string m_setname;
@@ -55,15 +63,13 @@ namespace yp2 {
             bool m_is_multi;
             bool m_in_use;
             std::list<BackendPtr> m_backend_list;
-            std::map<std::string,Multi::Set> m_sets;
-            void multi_move();
+            std::map<std::string,Multi::FrontendSet> m_sets;
+
+            void multi_move(std::list<BackendPtr> &blist);
             void init(Package &package, Z_GDU *gdu);
             void close(Package &package);
             void search(Package &package, Z_APDU *apdu);
-#if 0
             void present(Package &package, Z_APDU *apdu);
-            void scan(Package &package, Z_APDU *apdu);
-#endif
             Rep *m_p;
         };            
         struct Multi::Map {
@@ -91,6 +97,11 @@ namespace yp2 {
 
 using namespace yp2;
 
+bool yf::Multi::BackendSet::operator < (const BackendSet &k) const
+{
+    return m_count < k.m_count;
+}
+
 yf::Multi::Frontend::Frontend(Rep *rep)
 {
     m_p = rep;
@@ -147,18 +158,18 @@ void yf::Multi::Rep::release_frontend(Package &package)
     }
 }
 
-yf::Multi::Set::Set(std::string setname)
+yf::Multi::FrontendSet::FrontendSet(std::string setname)
     :  m_setname(setname)
 {
 }
 
 
-yf::Multi::Set::Set()
+yf::Multi::FrontendSet::FrontendSet()
 {
 }
 
 
-yf::Multi::Set::~Set()
+yf::Multi::FrontendSet::~FrontendSet()
 {
 }
 
@@ -205,49 +216,96 @@ void yf::Multi::Frontend::close(Package &package)
     }
 }
 
-void yf::Multi::Frontend::multi_move()
+void yf::Multi::Frontend::multi_move(std::list<BackendPtr> &blist)
 {
     std::list<BackendPtr>::const_iterator bit;
     boost::thread_group g;
-    for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
+    for (bit = blist.begin(); bit != blist.end(); bit++)
     {
         g.add_thread(new boost::thread(**bit));
     }
     g.join_all();
 }
 
+
+void yf::Multi::FrontendSet::round_robin(int start, int number,
+                                         std::list<PresentJob> &jobs)
+{
+    int fetched = 0;
+    int p = 1;
+    bool eof = true;
+
+    std::list<int> pos;
+    std::list<int> inside_pos;
+    std::list<BackendSet>::const_iterator bsit;
+    for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end(); bsit++)
+    {
+        pos.push_back(1);
+        inside_pos.push_back(0);
+    }
+
+    std::list<int>::iterator psit = pos.begin();
+    std::list<int>::iterator esit = inside_pos.begin();
+    bsit = m_backend_sets.begin();
+    while (fetched < number)
+    {
+        if (bsit == m_backend_sets.end())
+        {
+            psit = pos.begin();
+            esit = inside_pos.begin();
+            bsit = m_backend_sets.begin();
+            if (eof)
+                break;
+            eof = true;
+        }
+        if (*psit <= bsit->m_count)
+        {
+            if (p >= start)
+            {
+                PresentJob job;
+                job.m_backend = bsit->m_backend;
+                job.m_pos = *psit;
+                job.m_inside_pos = *esit;
+                jobs.push_back(job);
+                (*esit)++;
+                fetched++;
+            }
+            (*psit)++;
+            p++;
+            eof = false;
+        }
+        psit++;
+        esit++;
+        bsit++;
+    }
+}
+
 void yf::Multi::Frontend::init(Package &package, Z_GDU *gdu)
 {
     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
 
-    // empty or non-existang vhost is the same..
-    const char *vhost_cstr =
-        yaz_oi_get_string_oidval(&req->otherInfo, VAL_PROXY, 1, 0);
-    std::string vhost;
-    if (vhost_cstr)
-        vhost = std::string(vhost_cstr);
+    std::list<std::string> targets;
+
+    yp2::util::get_vhost_otherinfo(&req->otherInfo, false, targets);
 
-    std::map<std::string, Map>::const_iterator it;
-    it = m_p->m_maps.find(std::string(vhost));
-    if (it == m_p->m_maps.end())
+    if (targets.size() < 1)
     {
-        // might return diagnostics if no match
         package.move();
         return;
     }
-    std::list<std::string>::const_iterator hit = it->second.m_hosts.begin();
-    for (; hit != it->second.m_hosts.end(); hit++)
+
+    std::list<std::string>::const_iterator t_it = targets.begin();
+    for (; t_it != targets.end(); t_it++)
     {
         Session s;
         Backend *b = new Backend;
-        b->m_vhost = *hit;
-        b->m_route = it->second.m_route;
+        b->m_vhost = *t_it;
+
+        // b->m_route unset
         b->m_package = PackagePtr(new Package(s, package.origin()));
 
         m_backend_list.push_back(BackendPtr(b));
     }
-    // we're going to deal with this for sure..
-
     m_is_multi = true;
 
     // create init request 
@@ -275,13 +333,11 @@ void yf::Multi::Frontend::init(Package &package, Z_GDU *gdu)
 
         b->m_package->copy_filter(package);
     }
-    multi_move();
+    multi_move(m_backend_list);
 
     // create the frontend init response based on each backend init response
     yp2::odr odr;
 
-    int i;
-
     Z_APDU *f_apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
     Z_InitResponse *f_resp = f_apdu->u.initResponse;
 
@@ -303,6 +359,7 @@ void yf::Multi::Frontend::init(Package &package, Z_GDU *gdu)
         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
             Z_APDU_initResponse)
         {
+            int i;
             Z_APDU *b_apdu = gdu->u.z3950;
             Z_InitResponse *b_resp = b_apdu->u.initResponse;
 
@@ -335,25 +392,38 @@ void yf::Multi::Frontend::search(Package &package, Z_APDU *apdu_req)
 {
     // create search request 
     Z_SearchRequest *req = apdu_req->u.searchRequest;
-
+        
     // deal with piggy back (for now disable)
     *req->smallSetUpperBound = 0;
     *req->largeSetLowerBound = 1;
     *req->mediumSetPresentNumber = 1;
 
+    int default_num_db = req->num_databaseNames;
+    char **default_db = req->databaseNames;
+
     std::list<BackendPtr>::const_iterator bit;
     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
     {
         PackagePtr p = (*bit)->m_package;
-        // we don't modify database name yet!
-
+        yp2::odr odr;
+    
+        if (!yp2::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
+                                                &req->num_databaseNames,
+                                                &req->databaseNames))
+        {
+            req->num_databaseNames = default_num_db;
+            req->databaseNames = default_db;
+        }
         p->request() = apdu_req;
         p->copy_filter(package);
     }
-    multi_move();
+    multi_move(m_backend_list);
 
     // look at each response
-    int total_hits = 0;
+    FrontendSet resultSet(std::string(req->resultSetName));
+
+    int total_count = 0;
+    Z_Records *z_records_diag = 0;  // no diagnostics (yet)
     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
     {
         PackagePtr p = (*bit)->m_package;
@@ -367,8 +437,20 @@ void yf::Multi::Frontend::search(Package &package, Z_APDU *apdu_req)
         {
             Z_APDU *b_apdu = gdu->u.z3950;
             Z_SearchResponse *b_resp = b_apdu->u.searchResponse;
-            
-            total_hits += *b_resp->resultCount;
+         
+            // see we get any errors (AKA diagnstics)
+            if (b_resp->records)
+            {
+                if (b_resp->records->which == Z_Records_NSD
+                    || b_resp->records->which == Z_Records_multipleNSD)
+                    z_records_diag = b_resp->records;
+                // we may set this multiple times (TOO BAD!)
+            }
+            BackendSet backendSet;
+            backendSet.m_backend = *bit;
+            backendSet.m_count = *b_resp->resultCount;
+            total_count += *b_resp->resultCount;
+            resultSet.m_backend_sets.push_back(backendSet);
         }
         else
         {
@@ -382,8 +464,150 @@ void yf::Multi::Frontend::search(Package &package, Z_APDU *apdu_req)
     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
 
-    *f_resp->resultCount = total_hits;
+    if (z_records_diag)
+    {
+        // search error
+        f_resp->records = z_records_diag;
+    }
+    else
+    {   // assume OK
+        m_sets[resultSet.m_setname] = resultSet;
+    }
+    *f_resp->resultCount = total_count;
+    
+    package.response() = f_apdu;
+}
+
+void yf::Multi::Frontend::present(Package &package, Z_APDU *apdu_req)
+{
+    // create present request 
+    Z_PresentRequest *req = apdu_req->u.presentRequest;
+
+    Sets_it it;
+    it = m_sets.find(std::string(req->resultSetId));
+    if (it == m_sets.end())
+    {
+        yp2::odr odr;
+        Z_APDU *apdu = 
+            odr.create_presentResponse(
+                apdu_req,
+                YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
+                req->resultSetId);
+        package.response() = apdu;
+        return;
+    }
+    std::list<Multi::FrontendSet::PresentJob> jobs;
+    int start = *req->resultSetStartPoint;
+    int number = *req->numberOfRecordsRequested;
+    it->second.round_robin(start, number, jobs);
+
+    std::list<BackendPtr> present_backend_list;
 
+    std::list<BackendSet>::const_iterator bsit;
+    bsit = it->second.m_backend_sets.begin();
+    for (; bsit != it->second.m_backend_sets.end(); bsit++)
+    {
+        std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
+        int start = -1;
+        int end = -1;
+        
+        for (jit = jobs.begin(); jit != jobs.end(); jit++)
+        {
+            if (jit->m_backend == bsit->m_backend)
+            {
+                if (start == -1 || jit->m_pos < start)
+                    start = jit->m_pos;
+                if (end == -1 || jit->m_pos > end)
+                    end = jit->m_pos;
+            }
+        }
+        if (start != -1)
+        {
+            PackagePtr p = bsit->m_backend->m_package;
+
+            *req->resultSetStartPoint = start;
+            *req->numberOfRecordsRequested = end - start + 1;
+            
+            p->request() = apdu_req;
+            p->copy_filter(package);
+
+            present_backend_list.push_back(bsit->m_backend);
+        }
+    }
+    multi_move(present_backend_list);
+
+    // look at each response
+    Z_Records *z_records_diag = 0;
+
+    std::list<BackendPtr>::const_iterator pbit = present_backend_list.begin();
+    for (; pbit != present_backend_list.end(); pbit++)
+    {
+        PackagePtr p = (*pbit)->m_package;
+        
+        if (p->session().is_closed()) // if any backend closes, close frontend
+            package.session().close();
+        
+        Z_GDU *gdu = p->response().get();
+        if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
+            Z_APDU_presentResponse)
+        {
+            Z_APDU *b_apdu = gdu->u.z3950;
+            Z_PresentResponse *b_resp = b_apdu->u.presentResponse;
+         
+            // see we get any errors (AKA diagnstics)
+            if (b_resp->records)
+            {
+                if (b_resp->records->which != Z_Records_DBOSD)
+                    z_records_diag = b_resp->records;
+                // we may set this multiple times (TOO BAD!)
+            }
+        }
+        else
+        {
+            // if any target does not return present response - return that 
+            package.response() = p->response();
+            return;
+        }
+    }
+
+    yp2::odr odr;
+    Z_APDU *f_apdu = odr.create_presentResponse(apdu_req, 0, 0);
+    Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
+
+    if (z_records_diag)
+    {
+        f_resp->records = z_records_diag;
+        *f_resp->presentStatus = Z_PresentStatus_failure;
+    }
+    else
+    {
+        f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
+        Z_Records * records = f_resp->records;
+        records->which = Z_Records_DBOSD;
+        records->u.databaseOrSurDiagnostics =
+            (Z_NamePlusRecordList *)
+            odr_malloc(odr, sizeof(Z_NamePlusRecordList));
+        Z_NamePlusRecordList *nprl = records->u.databaseOrSurDiagnostics;
+        nprl->num_records = jobs.size();
+        nprl->records = (Z_NamePlusRecord**)
+            odr_malloc(odr, sizeof(Z_NamePlusRecord *) * nprl->num_records);
+        int i = 0;
+        std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
+        for (jit = jobs.begin(); jit != jobs.end(); jit++)
+        {
+            PackagePtr p = jit->m_backend->m_package;
+            
+            Z_GDU *gdu = p->response().get();
+            Z_APDU *b_apdu = gdu->u.z3950;
+            Z_PresentResponse *b_resp = b_apdu->u.presentResponse;
+
+            nprl->records[i++] =
+                b_resp->records->u.databaseOrSurDiagnostics->
+                records[jit->m_inside_pos];
+        }
+        *f_resp->nextResultSetPosition = start + i;
+        *f_resp->numberOfRecordsReturned = i;
+    }
     package.response() = f_apdu;
 }
 
@@ -418,6 +642,10 @@ void yf::Multi::process(Package &package) const
         {
             f->search(package, apdu);
         }
+        else if (apdu->which == Z_APDU_presentRequest)
+        {
+            f->present(package, apdu);
+        }
         else
         {
             yp2::odr odr;