multi: do not mess with request package
[metaproxy-moved-to-github.git] / src / filter_multi.cpp
1 /* This file is part of Metaproxy.
2    Copyright (C) 2005-2013 Index Data
3
4 Metaproxy is free software; you can redistribute it and/or modify it under
5 the terms of the GNU General Public License as published by the Free
6 Software Foundation; either version 2, or (at your option) any later
7 version.
8
9 Metaproxy is distributed in the hope that it will be useful, but WITHOUT ANY
10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
12 for more details.
13
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
17 */
18
19 #include <yaz/log.h>
20
21 #include "config.hpp"
22
23 #include <metaproxy/filter.hpp>
24 #include <metaproxy/package.hpp>
25
26 #include <boost/thread/thread.hpp>
27 #include <boost/thread/mutex.hpp>
28 #include <boost/thread/condition.hpp>
29 #include <boost/shared_ptr.hpp>
30
31 #include <metaproxy/util.hpp>
32 #include "filter_multi.hpp"
33
34 #include <yaz/zgdu.h>
35 #include <yaz/otherinfo.h>
36 #include <yaz/diagbib1.h>
37 #include <yaz/match_glob.h>
38
39 #include <vector>
40 #include <algorithm>
41 #include <map>
42 #include <iostream>
43
44 namespace mp = metaproxy_1;
45 namespace yf = mp::filter;
46
47 namespace metaproxy_1 {
48     namespace filter {
49         enum multi_merge_type {
50             round_robin,
51             serve_order
52         };
53         struct Multi::BackendSet {
54             BackendPtr m_backend;
55             int m_count;
56             bool operator < (const BackendSet &k) const;
57             bool operator == (const BackendSet &k) const;
58         };
59         struct Multi::ScanTermInfo {
60             std::string m_norm_term;
61             std::string m_display_term;
62             int m_count;
63             bool operator < (const ScanTermInfo &) const;
64             bool operator == (const ScanTermInfo &) const;
65             Z_Entry *get_entry(ODR odr);
66         };
67         struct Multi::FrontendSet {
68             class PresentJob {
69             public:
70                 BackendPtr m_backend;
71                 int m_pos; // position for backend (1=first, 2=second,..
72                 int m_start; // present request start
73                 PresentJob(BackendPtr ptr, int pos) :
74                     m_backend(ptr), m_pos(pos), m_start(0) {};
75             };
76             FrontendSet(std::string setname);
77             FrontendSet();
78             ~FrontendSet();
79
80             void round_robin(int pos, int number, std::list<PresentJob> &job);
81             void serve_order(int pos, int number, std::list<PresentJob> &job);
82
83             std::list<BackendSet> m_backend_sets;
84             std::string m_setname;
85         };
86         struct Multi::Backend {
87             PackagePtr m_package;
88             std::string m_backend_database;
89             std::string m_vhost;
90             std::string m_route;
91             std::string m_auth;
92             void operator() (void);  // thread operation
93         };
94         struct Multi::Frontend {
95             Frontend(Rep *rep);
96             ~Frontend();
97             bool m_is_multi;
98             bool m_in_use;
99             std::list<BackendPtr> m_backend_list;
100             std::map<std::string,Multi::FrontendSet> m_sets;
101
102             void multi_move(std::list<BackendPtr> &blist);
103             void init(Package &package, Z_GDU *gdu);
104             void close(Package &package);
105             void search(Package &package, Z_APDU *apdu);
106             void present(Package &package, Z_APDU *apdu);
107             void scan1(Package &package, Z_APDU *apdu);
108             void scan2(Package &package, Z_APDU *apdu);
109             void relay_apdu(Package &package, Z_APDU *apdu);
110             void record_diagnostics(Z_Records *records, 
111                                     Z_DiagRecs * &z_diag,
112                                     ODR odr,
113                                     int &no_successful);
114             Rep *m_p;
115         };
116         class Multi::Map {
117             std::string m_target_pattern;
118             std::string m_route;
119             std::string m_auth;
120         public:
121             Map(std::string pattern, std::string route, std::string auth) :
122                 m_target_pattern(pattern), m_route(route), m_auth(auth) {};
123             bool match(const std::string target, std::string *ret,
124                        std::string *auth) const {
125                 if (yaz_match_glob(m_target_pattern.c_str(), target.c_str()))
126                 {
127                     *ret = m_route;
128                     *auth = m_auth;
129                     return true;
130                 }
131                 return false;
132             };
133         };
134         class Multi::Rep {
135             friend class Multi;
136             friend struct Frontend;
137
138             Rep();
139             FrontendPtr get_frontend(Package &package);
140             void release_frontend(Package &package);
141         private:
142             std::list<Multi::Map> m_route_patterns;
143             boost::mutex m_mutex;
144             boost::condition m_cond_session_ready;
145             std::map<mp::Session, FrontendPtr> m_clients;
146             bool m_hide_unavailable;
147             bool m_hide_errors;
148             multi_merge_type m_merge_type;
149         };
150     }
151 }
152
153 yf::Multi::Rep::Rep()
154 {
155     m_hide_unavailable = false;
156     m_hide_errors = false;
157     m_merge_type = round_robin;
158 }
159
160 bool yf::Multi::BackendSet::operator < (const BackendSet &k) const
161 {
162     return m_count < k.m_count;
163 }
164
165 yf::Multi::Frontend::Frontend(Rep *rep)
166 {
167     m_p = rep;
168     m_is_multi = false;
169 }
170
171 yf::Multi::Frontend::~Frontend()
172 {
173 }
174
175 yf::Multi::FrontendPtr yf::Multi::Rep::get_frontend(mp::Package &package)
176 {
177     boost::mutex::scoped_lock lock(m_mutex);
178
179     std::map<mp::Session,yf::Multi::FrontendPtr>::iterator it;
180
181     while(true)
182     {
183         it = m_clients.find(package.session());
184         if (it == m_clients.end())
185             break;
186
187         if (!it->second->m_in_use)
188         {
189             it->second->m_in_use = true;
190             return it->second;
191         }
192         m_cond_session_ready.wait(lock);
193     }
194     FrontendPtr f(new Frontend(this));
195     m_clients[package.session()] = f;
196     f->m_in_use = true;
197     return f;
198 }
199
200 void yf::Multi::Rep::release_frontend(mp::Package &package)
201 {
202     boost::mutex::scoped_lock lock(m_mutex);
203     std::map<mp::Session,yf::Multi::FrontendPtr>::iterator it;
204
205     it = m_clients.find(package.session());
206     if (it != m_clients.end())
207     {
208         if (package.session().is_closed())
209         {
210             it->second->close(package);
211             m_clients.erase(it);
212         }
213         else
214         {
215             it->second->m_in_use = false;
216         }
217         m_cond_session_ready.notify_all();
218     }
219 }
220
221 yf::Multi::FrontendSet::FrontendSet(std::string setname)
222     :  m_setname(setname)
223 {
224 }
225
226
227 yf::Multi::FrontendSet::FrontendSet()
228 {
229 }
230
231
232 yf::Multi::FrontendSet::~FrontendSet()
233 {
234 }
235
236 yf::Multi::Multi() : m_p(new Multi::Rep)
237 {
238 }
239
240 yf::Multi::~Multi() {
241 }
242
243
244 void yf::Multi::Backend::operator() (void)
245 {
246     m_package->move(m_route);
247 }
248
249
250 void yf::Multi::Frontend::close(mp::Package &package)
251 {
252     std::list<BackendPtr>::const_iterator bit;
253     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
254     {
255         BackendPtr b = *bit;
256
257         b->m_package->copy_filter(package);
258         b->m_package->request() = (Z_GDU *) 0;
259         b->m_package->session().close();
260         b->m_package->move(b->m_route);
261     }
262 }
263
264 void yf::Multi::Frontend::multi_move(std::list<BackendPtr> &blist)
265 {
266     std::list<BackendPtr>::const_iterator bit;
267     boost::thread_group g;
268     for (bit = blist.begin(); bit != blist.end(); bit++)
269     {
270         g.add_thread(new boost::thread(**bit));
271     }
272     g.join_all();
273 }
274
275 void yf::Multi::FrontendSet::serve_order(int start, int number,
276                                          std::list<PresentJob> &jobs)
277 {
278     int i;
279     for (i = 0; i < number; i++)
280     {
281         std::list<BackendSet>::const_iterator bsit;
282         int voffset = 0;
283         int offset = start + i - 1;
284         for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end();
285              bsit++)
286         {
287             if (offset >= voffset && offset < voffset + bsit->m_count)
288             {
289                 PresentJob job(bsit->m_backend, offset - voffset + 1);
290                 jobs.push_back(job);
291                 break;
292             }
293             voffset += bsit->m_count;
294         }
295     }
296 }
297
298 void yf::Multi::FrontendSet::round_robin(int start, int number,
299                                          std::list<PresentJob> &jobs)
300 {
301     std::list<int> pos;
302     std::list<BackendSet>::const_iterator bsit;
303     for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end(); bsit++)
304     {
305         pos.push_back(1);
306     }
307
308     int p = 1;
309 #if 1
310     // optimization step!
311     int omin = 0;
312     while(true)
313     {
314         int min = 0;
315         int no_left = 0;
316         // find min count for each set which is > omin
317         for (bsit = m_backend_sets.begin(); bsit != m_backend_sets.end(); bsit++)
318         {
319             if (bsit->m_count > omin)
320             {
321                 if (no_left == 0 || bsit->m_count < min)
322                     min = bsit->m_count;
323                 no_left++;
324             }
325         }
326         if (no_left == 0) // if nothing greater than omin, bail out.
327             break;
328         int skip = no_left * min;
329         if (p + skip > start)  // step gets us "into" present range?
330         {
331             // Yes. skip until start.. Rounding off is deliberate!
332             min = (start-p) / no_left;
333             p += no_left * min;
334
335             // update positions in each set..
336             std::list<int>::iterator psit = pos.begin();
337             for (psit = pos.begin(); psit != pos.end(); psit++)
338                 *psit += min;
339             break;
340         }
341         // skip on each set.. before "present range"..
342         p = p + skip;
343
344         std::list<int>::iterator psit = pos.begin();
345         for (psit = pos.begin(); psit != pos.end(); psit++)
346             *psit += min;
347
348         omin = min; // update so we consider next class (with higher count)
349     }
350 #endif
351     int fetched = 0;
352     bool more = true;
353     while (more)
354     {
355         more = false;
356         std::list<int>::iterator psit = pos.begin();
357         bsit = m_backend_sets.begin();
358
359         for (; bsit != m_backend_sets.end(); psit++,bsit++)
360         {
361             if (fetched >= number)
362             {
363                 more = false;
364                 break;
365             }
366             if (*psit <= bsit->m_count)
367             {
368                 if (p >= start)
369                 {
370                     PresentJob job(bsit->m_backend, *psit);
371                     jobs.push_back(job);
372                     fetched++;
373                 }
374                 (*psit)++;
375                 p++;
376                 more = true;
377             }
378         }
379     }
380 }
381
382 void yf::Multi::Frontend::init(mp::Package &package, Z_GDU *gdu)
383 {
384     Z_InitRequest *req = gdu->u.z3950->u.initRequest;
385
386     std::list<std::string> targets;
387
388     mp::util::get_vhost_otherinfo(req->otherInfo, targets);
389
390     if (targets.size() < 1)
391     {
392         package.move();
393         return;
394     }
395
396     std::list<std::string>::const_iterator t_it = targets.begin();
397     for (; t_it != targets.end(); t_it++)
398     {
399         Session s;
400         Backend *b = new Backend;
401         b->m_vhost = *t_it;
402
403         std::list<Multi::Map>::const_iterator it =
404             m_p->m_route_patterns.begin();
405         while (it != m_p->m_route_patterns.end()) {
406             if (it->match(*t_it, &b->m_route, &b->m_auth))
407                 break;
408             it++;
409         }
410         // b->m_route = m_p->m_target_route[*t_it];
411         // b->m_route unset
412         b->m_package = PackagePtr(new Package(s, package.origin()));
413
414         m_backend_list.push_back(BackendPtr(b));
415     }
416     m_is_multi = true;
417
418     // create init request
419     std::list<BackendPtr>::iterator bit;
420     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
421     {
422         mp::odr odr;
423         BackendPtr b = *bit;
424         Z_APDU *init_apdu = zget_APDU(odr, Z_APDU_initRequest);
425
426         std::list<std::string>vhost_one;
427         vhost_one.push_back(b->m_vhost);
428         mp::util::set_vhost_otherinfo(&init_apdu->u.initRequest->otherInfo,
429                                        odr, vhost_one);
430
431
432         Z_InitRequest *breq = init_apdu->u.initRequest;
433
434         if (b->m_auth.length())
435         {
436             breq->idAuthentication =
437                 (Z_IdAuthentication *)
438                 odr_malloc(odr, sizeof(*breq->idAuthentication));
439             breq->idAuthentication->which = Z_IdAuthentication_open;
440             breq->idAuthentication->u.open = odr_strdup(odr, b->m_auth.c_str());
441         }
442         else
443             breq->idAuthentication = req->idAuthentication;
444
445         *breq->preferredMessageSize = *req->preferredMessageSize;
446         *breq->maximumRecordSize = *req->maximumRecordSize;
447
448         ODR_MASK_SET(breq->options, Z_Options_search);
449         ODR_MASK_SET(breq->options, Z_Options_present);
450         ODR_MASK_SET(breq->options, Z_Options_namedResultSets);
451         ODR_MASK_SET(breq->options, Z_Options_scan);
452
453         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_1);
454         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_2);
455         ODR_MASK_SET(breq->protocolVersion, Z_ProtocolVersion_3);
456
457         b->m_package->request() = init_apdu;
458
459         b->m_package->copy_filter(package);
460     }
461     multi_move(m_backend_list);
462
463     // create the frontend init response based on each backend init response
464     mp::odr odr;
465
466     Z_APDU *f_apdu = odr.create_initResponse(gdu->u.z3950, 0, 0);
467     Z_InitResponse *f_resp = f_apdu->u.initResponse;
468
469     ODR_MASK_SET(f_resp->options, Z_Options_search);
470     ODR_MASK_SET(f_resp->options, Z_Options_present);
471     ODR_MASK_SET(f_resp->options, Z_Options_namedResultSets);
472     ODR_MASK_SET(f_resp->options, Z_Options_scan);
473
474     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_1);
475     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_2);
476     ODR_MASK_SET(f_resp->protocolVersion, Z_ProtocolVersion_3);
477
478     int no_failed = 0;
479     int no_succeeded = 0;
480
481     Odr_int preferredMessageSize = *req->preferredMessageSize;
482     Odr_int maximumRecordSize = *req->maximumRecordSize;
483     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); )
484     {
485         PackagePtr p = (*bit)->m_package;
486
487         if (p->session().is_closed())
488         {
489             // failed. Remove from list and increment number of failed
490             no_failed++;
491             bit = m_backend_list.erase(bit);
492             continue;
493         }
494         Z_GDU *gdu = p->response().get();
495         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
496             Z_APDU_initResponse)
497         {
498             int i;
499             Z_APDU *b_apdu = gdu->u.z3950;
500             Z_InitResponse *b_resp = b_apdu->u.initResponse;
501
502             // common options for all backends
503             for (i = 0; i <= Z_Options_stringSchema; i++)
504             {
505                 if (!ODR_MASK_GET(b_resp->options, i))
506                     ODR_MASK_CLEAR(f_resp->options, i);
507             }
508             // common protocol version
509             for (i = 0; i <= Z_ProtocolVersion_3; i++)
510                 if (!ODR_MASK_GET(b_resp->protocolVersion, i))
511                     ODR_MASK_CLEAR(f_resp->protocolVersion, i);
512             if (*b_resp->result)
513             {
514                 no_succeeded++;
515                 if (preferredMessageSize > *b_resp->preferredMessageSize)
516                     preferredMessageSize = *b_resp->preferredMessageSize;
517                 if (maximumRecordSize > *b_resp->maximumRecordSize)
518                     maximumRecordSize = *b_resp->maximumRecordSize;
519             }
520             else
521             {
522                 if (!f_resp->userInformationField
523                     && b_resp->userInformationField)
524                     f_resp->userInformationField = b_resp->userInformationField;
525                 no_failed++;
526             }
527         }
528         else
529             no_failed++;
530         bit++;
531     }
532     *f_resp->preferredMessageSize = preferredMessageSize;
533     *f_resp->maximumRecordSize = maximumRecordSize;
534
535     if (m_p->m_hide_unavailable)
536     {
537         if (no_succeeded == 0)
538         {
539             *f_resp->result = 0;
540             package.session().close();
541         }
542     }
543     else
544     {
545         if (no_failed)
546         {
547             *f_resp->result = 0;
548             package.session().close();
549         }
550     }
551     package.response() = f_apdu;
552 }
553
554 void yf::Multi::Frontend::record_diagnostics(Z_Records *records, 
555                                              Z_DiagRecs * &z_diag,
556                                              ODR odr,
557                                              int &no_successful)
558 {
559     // see we get any errors (AKA diagnstics)
560     if (records)
561     {
562         if (records->which == Z_Records_NSD)
563         {
564             if (!z_diag)
565             {
566                 z_diag = (Z_DiagRecs *)
567                     odr_malloc(odr, sizeof(*z_diag));
568                 z_diag->num_diagRecs = 0;
569                 z_diag->diagRecs = (Z_DiagRec**)
570                     odr_malloc(odr, sizeof(*z_diag->diagRecs));
571             }
572             else
573             {
574                 Z_DiagRec **n = (Z_DiagRec **)
575                     odr_malloc(odr,
576                                (1 + z_diag->num_diagRecs) * sizeof(*n));
577                 memcpy(n, z_diag->diagRecs, z_diag->num_diagRecs
578                        * sizeof(*n));
579                 z_diag->diagRecs = n;
580             }
581             Z_DiagRec *nr = (Z_DiagRec *) odr_malloc(odr, sizeof(*nr));
582             nr->which = Z_DiagRec_defaultFormat;
583             nr->u.defaultFormat =
584                 records->u.nonSurrogateDiagnostic;
585             z_diag->diagRecs[z_diag->num_diagRecs++] = nr;
586         }
587         else if (records->which == Z_Records_multipleNSD)
588         {
589             Z_DiagRecs * dr =records->u.multipleNonSurDiagnostics;
590
591             if (!z_diag)
592             {
593                 z_diag = (Z_DiagRecs *) odr_malloc(odr, sizeof(*z_diag));
594                 z_diag->num_diagRecs = 0;
595                 z_diag->diagRecs = 0;
596             }
597             Z_DiagRec **n = (Z_DiagRec **)
598                 odr_malloc(odr,
599                            (dr->num_diagRecs + z_diag->num_diagRecs) * 
600                            sizeof(*n));
601             if (z_diag->num_diagRecs)
602                 memcpy(n, z_diag->diagRecs, z_diag->num_diagRecs * sizeof(*n));
603             memcpy(n + z_diag->num_diagRecs,
604                    dr->diagRecs, dr->num_diagRecs * sizeof(*n));
605             z_diag->diagRecs = n;
606             z_diag->num_diagRecs += dr->num_diagRecs;
607         }
608         else
609             no_successful++; // probably piggyback
610     }
611     else
612         no_successful++;  // no records and no diagnostics
613 }
614
615 void yf::Multi::Frontend::search(mp::Package &package, Z_APDU *apdu_req)
616 {
617     // create search request
618     Z_SearchRequest *req = apdu_req->u.searchRequest;
619
620     int default_num_db = req->num_databaseNames;
621     char **default_db = req->databaseNames;
622
623     std::list<BackendPtr>::const_iterator bit;
624     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
625     {
626         PackagePtr p = (*bit)->m_package;
627         yazpp_1::GDU gdu1(apdu_req);
628         mp::odr odr;
629         Z_SearchRequest *req1 = gdu1.get()->u.z3950->u.searchRequest;
630
631         // they are altered now - to disable piggyback
632         *req1->smallSetUpperBound = 0;
633         *req1->largeSetLowerBound = 1;
634         *req1->mediumSetPresentNumber = 0;
635
636         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
637                                                &req1->num_databaseNames,
638                                                &req1->databaseNames))
639         {
640             req1->num_databaseNames = default_num_db;
641             req1->databaseNames = default_db;
642         }
643         p->request() = gdu1;
644         p->copy_filter(package);
645     }
646     multi_move(m_backend_list);
647
648     // look at each response
649     FrontendSet resultSet(std::string(req->resultSetName));
650
651     mp::odr odr;
652     Odr_int result_set_size = 0;
653     Z_DiagRecs *z_diag = 0;
654     int no_successful = 0;
655     PackagePtr close_p;
656
657     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
658     {
659         PackagePtr p = (*bit)->m_package;
660
661         // save closing package for at least one target
662         if (p->session().is_closed())
663             close_p = p;
664
665         Z_GDU *gdu = p->response().get();
666         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
667             Z_APDU_searchResponse)
668         {
669             Z_APDU *b_apdu = gdu->u.z3950;
670             Z_SearchResponse *b_resp = b_apdu->u.searchResponse;
671
672             record_diagnostics(b_resp->records, z_diag, odr, no_successful);
673
674             BackendSet backendSet;
675             backendSet.m_backend = *bit;
676             backendSet.m_count = *b_resp->resultCount;
677             result_set_size += *b_resp->resultCount;
678             resultSet.m_backend_sets.push_back(backendSet);
679         }
680     }
681
682     Z_APDU *f_apdu = odr.create_searchResponse(apdu_req, 0, 0);
683     Z_SearchResponse *f_resp = f_apdu->u.searchResponse;
684
685     yaz_log(YLOG_LOG, "no_successful=%d is_closed=%s hide_errors=%s",
686             no_successful,
687             close_p ? "true" : "false",
688             m_p->m_hide_errors ? "true" : "false");
689     *f_resp->resultCount = result_set_size;
690     if (close_p && (no_successful == 0 || !m_p->m_hide_errors))
691     {
692         package.session().close();
693         package.response() = close_p->response();
694         return;
695     }
696     if (z_diag && (no_successful == 0 || !m_p->m_hide_errors))
697     {
698         f_resp->records = (Z_Records *)
699             odr_malloc(odr, sizeof(*f_resp->records));
700         if (z_diag->num_diagRecs > 1)
701         {
702             f_resp->records->which = Z_Records_multipleNSD;
703             f_resp->records->u.multipleNonSurDiagnostics = z_diag;
704         }
705         else
706         {
707             f_resp->records->which = Z_Records_NSD;
708             f_resp->records->u.nonSurrogateDiagnostic =
709                 z_diag->diagRecs[0]->u.defaultFormat;
710         }
711     }
712     // assume OK
713     m_sets[resultSet.m_setname] = resultSet;
714
715     Odr_int number;
716     mp::util::piggyback(*req->smallSetUpperBound,
717                         *req->largeSetLowerBound,
718                         *req->mediumSetPresentNumber,
719                         0, 0,
720                         result_set_size,
721                         number, 0);
722     Package pp(package.session(), package.origin());
723     if (z_diag == 0 && number > 0)
724     {
725         pp.copy_filter(package);
726         Z_APDU *p_apdu = zget_APDU(odr, Z_APDU_presentRequest);
727         Z_PresentRequest *p_req = p_apdu->u.presentRequest;
728         p_req->preferredRecordSyntax = req->preferredRecordSyntax;
729         p_req->resultSetId = req->resultSetName;
730         *p_req->resultSetStartPoint = 1;
731         *p_req->numberOfRecordsRequested = number;
732         pp.request() = p_apdu;
733         present(pp, p_apdu);
734
735         if (pp.session().is_closed())
736             package.session().close();
737
738         Z_GDU *gdu = pp.response().get();
739         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
740             Z_APDU_presentResponse)
741         {
742             Z_PresentResponse *p_res = gdu->u.z3950->u.presentResponse;
743             f_resp->records = p_res->records;
744             *f_resp->numberOfRecordsReturned =
745                 *p_res->numberOfRecordsReturned;
746             *f_resp->nextResultSetPosition =
747                 *p_res->nextResultSetPosition;
748         }
749         else
750         {
751             package.response() = pp.response();
752             return;
753         }
754     }
755     package.response() = f_apdu; // in this scope because of p
756 }
757
758 void yf::Multi::Frontend::present(mp::Package &package, Z_APDU *apdu_req)
759 {
760     // create present request
761     Z_PresentRequest *req = apdu_req->u.presentRequest;
762
763     Sets_it it;
764     it = m_sets.find(std::string(req->resultSetId));
765     if (it == m_sets.end())
766     {
767         mp::odr odr;
768         Z_APDU *apdu =
769             odr.create_presentResponse(
770                 apdu_req,
771                 YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
772                 req->resultSetId);
773         package.response() = apdu;
774         return;
775     }
776     std::list<Multi::FrontendSet::PresentJob> jobs;
777     int start = *req->resultSetStartPoint;
778     int number = *req->numberOfRecordsRequested;
779
780     if (m_p->m_merge_type == round_robin)
781         it->second.round_robin(start, number, jobs);
782     else if (m_p->m_merge_type == serve_order)
783         it->second.serve_order(start, number, jobs);
784
785     if (0)
786     {
787         std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
788         for (jit = jobs.begin(); jit != jobs.end(); jit++)
789         {
790             yaz_log(YLOG_LOG, "job pos=%d", jit->m_pos);
791         }
792     }
793
794     std::list<BackendPtr> present_backend_list;
795
796     std::list<BackendSet>::const_iterator bsit;
797     bsit = it->second.m_backend_sets.begin();
798     for (; bsit != it->second.m_backend_sets.end(); bsit++)
799     {
800         int start = -1;
801         int end = -1;
802         {
803             std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
804             for (jit = jobs.begin(); jit != jobs.end(); jit++)
805             {
806                 if (jit->m_backend == bsit->m_backend)
807                 {
808                     if (start == -1 || jit->m_pos < start)
809                         start = jit->m_pos;
810                     if (end == -1 || jit->m_pos > end)
811                         end = jit->m_pos;
812                 }
813             }
814         }
815         if (start != -1)
816         {
817             std::list<Multi::FrontendSet::PresentJob>::iterator jit;
818             for (jit = jobs.begin(); jit != jobs.end(); jit++)
819             {
820                 if (jit->m_backend == bsit->m_backend)
821                 {
822                     if (jit->m_pos >= start && jit->m_pos <= end)
823                         jit->m_start = start;
824                 }
825             }
826
827             PackagePtr p = bsit->m_backend->m_package;
828
829             *req->resultSetStartPoint = start;
830             *req->numberOfRecordsRequested = end - start + 1;
831
832             p->request() = apdu_req;
833             p->copy_filter(package);
834
835             present_backend_list.push_back(bsit->m_backend);
836         }
837     }
838     multi_move(present_backend_list);
839
840     // look at each response
841     Z_DiagRecs *z_diag = 0;
842     int no_successful = 0;
843     mp::odr odr;
844     PackagePtr close_p;
845
846     std::list<BackendPtr>::const_iterator pbit = present_backend_list.begin();
847     for (; pbit != present_backend_list.end(); pbit++)
848     {
849         PackagePtr p = (*pbit)->m_package;
850
851         // save closing package for at least one target
852         if (p->session().is_closed())
853             close_p = p;
854
855         Z_GDU *gdu = p->response().get();
856         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
857             Z_APDU_presentResponse)
858         {
859             Z_APDU *b_apdu = gdu->u.z3950;
860             Z_PresentResponse *b_resp = b_apdu->u.presentResponse;
861
862             record_diagnostics(b_resp->records, z_diag, odr, no_successful);
863         }
864     }
865
866     Z_APDU *f_apdu = odr.create_presentResponse(apdu_req, 0, 0);
867     Z_PresentResponse *f_resp = f_apdu->u.presentResponse;
868
869     if (close_p && (no_successful == 0 || !m_p->m_hide_errors))
870     {
871         package.session().close();
872         package.response() = close_p->response();
873         return;
874     }
875     if (z_diag && (no_successful == 0 || !m_p->m_hide_errors))
876     {
877         f_resp->records = (Z_Records *)
878             odr_malloc(odr, sizeof(*f_resp->records));
879         if (z_diag->num_diagRecs > 1)
880         {
881             f_resp->records->which = Z_Records_multipleNSD;
882             f_resp->records->u.multipleNonSurDiagnostics = z_diag;
883         }
884         else
885         {
886             f_resp->records->which = Z_Records_NSD;
887             f_resp->records->u.nonSurrogateDiagnostic =
888                 z_diag->diagRecs[0]->u.defaultFormat;
889         }
890     }
891     else if (number < 0 || (size_t) number > jobs.size())
892     {
893         f_apdu =
894             odr.create_presentResponse(
895                 apdu_req,
896                 YAZ_BIB1_PRESENT_REQUEST_OUT_OF_RANGE,
897                 0);
898     }
899     else
900     {
901         f_resp->records = (Z_Records *) odr_malloc(odr, sizeof(Z_Records));
902         Z_Records * records = f_resp->records;
903         records->which = Z_Records_DBOSD;
904         records->u.databaseOrSurDiagnostics =
905             (Z_NamePlusRecordList *)
906             odr_malloc(odr, sizeof(Z_NamePlusRecordList));
907         Z_NamePlusRecordList *nprl = records->u.databaseOrSurDiagnostics;
908         nprl->num_records = jobs.size();
909         nprl->records = (Z_NamePlusRecord**)
910             odr_malloc(odr, sizeof(Z_NamePlusRecord *) * nprl->num_records);
911         int i = 0;
912         std::list<Multi::FrontendSet::PresentJob>::const_iterator jit;
913         for (jit = jobs.begin(); jit != jobs.end(); jit++)
914         {
915             PackagePtr p = jit->m_backend->m_package;
916
917             Z_GDU *gdu = p->response().get();
918             Z_APDU *b_apdu = gdu->u.z3950;
919             int inside_pos = jit->m_pos - jit->m_start;
920             Z_Records *records = b_apdu->u.presentResponse->records;
921
922             if (records && records->which == Z_Records_DBOSD
923                 && inside_pos < 
924                 records->u.databaseOrSurDiagnostics->num_records)
925             {
926                 nprl->records[i] = (Z_NamePlusRecord*)
927                     odr_malloc(odr, sizeof(Z_NamePlusRecord));
928                 *nprl->records[i] = *records->
929                     u.databaseOrSurDiagnostics->records[inside_pos];
930                 nprl->records[i]->databaseName =
931                     odr_strdup(odr, jit->m_backend->m_vhost.c_str());
932                 i++;
933             }
934         }
935         nprl->num_records = i; // usually same as jobs.size();
936         *f_resp->nextResultSetPosition = start + i;
937         *f_resp->numberOfRecordsReturned = i;
938     }
939     package.response() = f_apdu;
940 }
941
942 void yf::Multi::Frontend::scan1(mp::Package &package, Z_APDU *apdu_req)
943 {
944     if (m_backend_list.size() > 1)
945     {
946         mp::odr odr;
947         Z_APDU *f_apdu =
948             odr.create_scanResponse(
949                 apdu_req, YAZ_BIB1_COMBI_OF_SPECIFIED_DATABASES_UNSUPP, 0);
950         package.response() = f_apdu;
951         return;
952     }
953     Z_ScanRequest *req = apdu_req->u.scanRequest;
954
955     int default_num_db = req->num_databaseNames;
956     char **default_db = req->databaseNames;
957
958     std::list<BackendPtr>::const_iterator bit;
959     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
960     {
961         PackagePtr p = (*bit)->m_package;
962         mp::odr odr;
963
964         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
965                                                 &req->num_databaseNames,
966                                                 &req->databaseNames))
967         {
968             req->num_databaseNames = default_num_db;
969             req->databaseNames = default_db;
970         }
971         p->request() = apdu_req;
972         p->copy_filter(package);
973     }
974     multi_move(m_backend_list);
975
976     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
977     {
978         PackagePtr p = (*bit)->m_package;
979
980         if (p->session().is_closed()) // if any backend closes, close frontend
981             package.session().close();
982
983         Z_GDU *gdu = p->response().get();
984         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
985             Z_APDU_scanResponse)
986         {
987             package.response() = p->response();
988             break;
989         }
990         else
991         {
992             // if any target does not return scan response - return that
993             package.response() = p->response();
994             return;
995         }
996     }
997 }
998
999 bool yf::Multi::ScanTermInfo::operator < (const ScanTermInfo &k) const
1000 {
1001     return m_norm_term < k.m_norm_term;
1002 }
1003
1004 bool yf::Multi::ScanTermInfo::operator == (const ScanTermInfo &k) const
1005 {
1006     return m_norm_term == k.m_norm_term;
1007 }
1008
1009 Z_Entry *yf::Multi::ScanTermInfo::get_entry(ODR odr)
1010 {
1011     Z_Entry *e = (Z_Entry *)odr_malloc(odr, sizeof(*e));
1012     e->which = Z_Entry_termInfo;
1013     Z_TermInfo *t;
1014     t = e->u.termInfo = (Z_TermInfo *) odr_malloc(odr, sizeof(*t));
1015     t->suggestedAttributes = 0;
1016     t->displayTerm = 0;
1017     t->alternativeTerm = 0;
1018     t->byAttributes = 0;
1019     t->otherTermInfo = 0;
1020     t->globalOccurrences = odr_intdup(odr, m_count);
1021     t->term = (Z_Term *)
1022         odr_malloc(odr, sizeof(*t->term));
1023     t->term->which = Z_Term_general;
1024     Odr_oct *o;
1025     t->term->u.general = o = (Odr_oct *)odr_malloc(odr, sizeof(Odr_oct));
1026
1027     o->len = o->size = m_norm_term.size();
1028     o->buf = (unsigned char *) odr_malloc(odr, o->len);
1029     memcpy(o->buf, m_norm_term.c_str(), o->len);
1030     return e;
1031 }
1032
1033 void yf::Multi::Frontend::relay_apdu(mp::Package &package, Z_APDU *apdu_req)
1034 {
1035     std::list<BackendPtr>::const_iterator bit;
1036     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
1037     {
1038         PackagePtr p = (*bit)->m_package;
1039         mp::odr odr;
1040
1041         p->request() = apdu_req;
1042         p->copy_filter(package);
1043     }
1044     multi_move(m_backend_list);
1045     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
1046     {
1047         PackagePtr p = (*bit)->m_package;
1048
1049         if (p->session().is_closed()) // if any backend closes, close frontend
1050             package.session().close();
1051
1052         package.response() = p->response();
1053     }
1054 }
1055
1056
1057 void yf::Multi::Frontend::scan2(mp::Package &package, Z_APDU *apdu_req)
1058 {
1059     Z_ScanRequest *req = apdu_req->u.scanRequest;
1060
1061     int default_num_db = req->num_databaseNames;
1062     char **default_db = req->databaseNames;
1063
1064     std::list<BackendPtr>::const_iterator bit;
1065     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
1066     {
1067         PackagePtr p = (*bit)->m_package;
1068         mp::odr odr;
1069
1070         if (!mp::util::set_databases_from_zurl(odr, (*bit)->m_vhost,
1071                                                 &req->num_databaseNames,
1072                                                 &req->databaseNames))
1073         {
1074             req->num_databaseNames = default_num_db;
1075             req->databaseNames = default_db;
1076         }
1077         p->request() = apdu_req;
1078         p->copy_filter(package);
1079     }
1080     multi_move(m_backend_list);
1081
1082     ScanTermInfoList entries_before;
1083     ScanTermInfoList entries_after;
1084     int no_before = 0;
1085     int no_after = 0;
1086
1087     for (bit = m_backend_list.begin(); bit != m_backend_list.end(); bit++)
1088     {
1089         PackagePtr p = (*bit)->m_package;
1090
1091         if (p->session().is_closed()) // if any backend closes, close frontend
1092             package.session().close();
1093
1094         Z_GDU *gdu = p->response().get();
1095         if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1096             Z_APDU_scanResponse)
1097         {
1098             Z_ScanResponse *res = gdu->u.z3950->u.scanResponse;
1099
1100             if (res->entries && res->entries->nonsurrogateDiagnostics)
1101             {
1102                 // failure
1103                 mp::odr odr;
1104                 Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 1, 0);
1105                 Z_ScanResponse *f_res = f_apdu->u.scanResponse;
1106
1107                 f_res->entries->nonsurrogateDiagnostics =
1108                     res->entries->nonsurrogateDiagnostics;
1109                 f_res->entries->num_nonsurrogateDiagnostics =
1110                     res->entries->num_nonsurrogateDiagnostics;
1111
1112                 package.response() = f_apdu;
1113                 return;
1114             }
1115
1116             if (res->entries && res->entries->entries)
1117             {
1118                 Z_Entry **entries = res->entries->entries;
1119                 int num_entries = res->entries->num_entries;
1120                 int position = 1;
1121                 if (req->preferredPositionInResponse)
1122                     position = *req->preferredPositionInResponse;
1123                 if (res->positionOfTerm)
1124                     position = *res->positionOfTerm;
1125
1126                 // before
1127                 int i;
1128                 for (i = 0; i<position-1 && i<num_entries; i++)
1129                 {
1130                     Z_Entry *ent = entries[i];
1131
1132                     if (ent->which == Z_Entry_termInfo)
1133                     {
1134                         ScanTermInfo my;
1135
1136                         Odr_int *occur = ent->u.termInfo->globalOccurrences;
1137                         my.m_count = occur ? *occur : 0;
1138
1139                         if (ent->u.termInfo->term->which == Z_Term_general)
1140                         {
1141                             my.m_norm_term = std::string(
1142                                 (const char *)
1143                                 ent->u.termInfo->term->u.general->buf,
1144                                 ent->u.termInfo->term->u.general->len);
1145                         }
1146                         if (my.m_norm_term.length())
1147                         {
1148                             ScanTermInfoList::iterator it =
1149                                 entries_before.begin();
1150                             while (it != entries_before.end() && my <*it)
1151                                 it++;
1152                             if (it != entries_before.end() && my == *it)
1153                             {
1154                                 it->m_count += my.m_count;
1155                             }
1156                             else
1157                             {
1158                                 entries_before.insert(it, my);
1159                                 no_before++;
1160                             }
1161                         }
1162                     }
1163                 }
1164                 // after
1165                 if (position <= 0)
1166                     i = 0;
1167                 else
1168                     i = position-1;
1169                 for ( ; i<num_entries; i++)
1170                 {
1171                     Z_Entry *ent = entries[i];
1172
1173                     if (ent->which == Z_Entry_termInfo)
1174                     {
1175                         ScanTermInfo my;
1176
1177                         Odr_int *occur = ent->u.termInfo->globalOccurrences;
1178                         my.m_count = occur ? *occur : 0;
1179
1180                         if (ent->u.termInfo->term->which == Z_Term_general)
1181                         {
1182                             my.m_norm_term = std::string(
1183                                 (const char *)
1184                                 ent->u.termInfo->term->u.general->buf,
1185                                 ent->u.termInfo->term->u.general->len);
1186                         }
1187                         if (my.m_norm_term.length())
1188                         {
1189                             ScanTermInfoList::iterator it =
1190                                 entries_after.begin();
1191                             while (it != entries_after.end() && *it < my)
1192                                 it++;
1193                             if (it != entries_after.end() && my == *it)
1194                             {
1195                                 it->m_count += my.m_count;
1196                             }
1197                             else
1198                             {
1199                                 entries_after.insert(it, my);
1200                                 no_after++;
1201                             }
1202                         }
1203                     }
1204                 }
1205
1206             }
1207         }
1208         else
1209         {
1210             // if any target does not return scan response - return that
1211             package.response() = p->response();
1212             return;
1213         }
1214     }
1215
1216     if (false)
1217     {
1218         std::cout << "BEFORE\n";
1219         ScanTermInfoList::iterator it = entries_before.begin();
1220         for(; it != entries_before.end(); it++)
1221         {
1222             std::cout << " " << it->m_norm_term << " " << it->m_count << "\n";
1223         }
1224
1225         std::cout << "AFTER\n";
1226         it = entries_after.begin();
1227         for(; it != entries_after.end(); it++)
1228         {
1229             std::cout << " " << it->m_norm_term << " " << it->m_count << "\n";
1230         }
1231     }
1232
1233     if (false)
1234     {
1235         mp::odr odr;
1236         Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 1, "not implemented");
1237         package.response() = f_apdu;
1238     }
1239     else
1240     {
1241         mp::odr odr;
1242         Z_APDU *f_apdu = odr.create_scanResponse(apdu_req, 0, 0);
1243         Z_ScanResponse *resp = f_apdu->u.scanResponse;
1244
1245         int number_returned = *req->numberOfTermsRequested;
1246         int position_returned = *req->preferredPositionInResponse;
1247
1248         resp->entries->num_entries = number_returned;
1249         resp->entries->entries = (Z_Entry**)
1250             odr_malloc(odr, sizeof(Z_Entry*) * number_returned);
1251         int i;
1252
1253         int lbefore = entries_before.size();
1254         if (lbefore < position_returned-1)
1255             position_returned = lbefore+1;
1256
1257         ScanTermInfoList::iterator it = entries_before.begin();
1258         for (i = 0; i<position_returned-1 && it != entries_before.end(); i++, it++)
1259         {
1260             resp->entries->entries[position_returned-2-i] = it->get_entry(odr);
1261         }
1262
1263         it = entries_after.begin();
1264
1265         if (position_returned <= 0)
1266             i = 0;
1267         else
1268             i = position_returned-1;
1269         for (; i<number_returned && it != entries_after.end(); i++, it++)
1270         {
1271             resp->entries->entries[i] = it->get_entry(odr);
1272         }
1273
1274         number_returned = i;
1275
1276         resp->positionOfTerm = odr_intdup(odr, position_returned);
1277         resp->numberOfEntriesReturned = odr_intdup(odr, number_returned);
1278         resp->entries->num_entries = number_returned;
1279
1280         package.response() = f_apdu;
1281     }
1282 }
1283
1284
1285 void yf::Multi::process(mp::Package &package) const
1286 {
1287     FrontendPtr f = m_p->get_frontend(package);
1288
1289     Z_GDU *gdu = package.request().get();
1290
1291     if (gdu && gdu->which == Z_GDU_Z3950 && gdu->u.z3950->which ==
1292         Z_APDU_initRequest && !f->m_is_multi)
1293     {
1294         f->init(package, gdu);
1295     }
1296     else if (!f->m_is_multi)
1297         package.move();
1298     else if (gdu && gdu->which == Z_GDU_Z3950)
1299     {
1300         Z_APDU *apdu = gdu->u.z3950;
1301         if (apdu->which == Z_APDU_initRequest)
1302         {
1303             mp::odr odr;
1304
1305             package.response() = odr.create_close(
1306                 apdu,
1307                 Z_Close_protocolError,
1308                 "double init");
1309
1310             package.session().close();
1311         }
1312         else if (apdu->which == Z_APDU_searchRequest)
1313         {
1314             f->search(package, apdu);
1315         }
1316         else if (apdu->which == Z_APDU_presentRequest)
1317         {
1318             f->present(package, apdu);
1319         }
1320         else if (apdu->which == Z_APDU_scanRequest)
1321         {
1322             f->scan2(package, apdu);
1323         }
1324         else if (apdu->which == Z_APDU_close)
1325         {
1326             f->relay_apdu(package, apdu);
1327         }
1328         else
1329         {
1330             mp::odr odr;
1331
1332             package.response() = odr.create_close(
1333                 apdu, Z_Close_protocolError,
1334                 "unsupported APDU in filter multi");
1335
1336             package.session().close();
1337         }
1338     }
1339     m_p->release_frontend(package);
1340 }
1341
1342 void mp::filter::Multi::configure(const xmlNode * ptr, bool test_only,
1343                                   const char *path)
1344 {
1345     for (ptr = ptr->children; ptr; ptr = ptr->next)
1346     {
1347         if (ptr->type != XML_ELEMENT_NODE)
1348             continue;
1349         if (!strcmp((const char *) ptr->name, "target"))
1350         {
1351             std::string auth;
1352             std::string route = mp::xml::get_route(ptr, auth);
1353             std::string target = mp::xml::get_text(ptr);
1354             if (target.length() == 0)
1355                 target = route;
1356             m_p->m_route_patterns.push_back(Multi::Map(target, route, auth));
1357         }
1358         else if (!strcmp((const char *) ptr->name, "hideunavailable"))
1359         {
1360             m_p->m_hide_unavailable = true;
1361         }
1362         else if (!strcmp((const char *) ptr->name, "hideerrors"))
1363         {
1364             m_p->m_hide_errors = true;
1365         }
1366         else if (!strcmp((const char *) ptr->name, "mergetype"))
1367         {
1368             std::string mergetype = mp::xml::get_text(ptr);
1369             if (mergetype == "roundrobin")
1370                 m_p->m_merge_type = round_robin;
1371             else if (mergetype == "serveorder")
1372                 m_p->m_merge_type = serve_order;
1373             else
1374                 throw mp::filter::FilterException
1375                     ("Bad mergetype "  + mergetype + " in multi filter");
1376
1377         }
1378         else
1379         {
1380             throw mp::filter::FilterException
1381                 ("Bad element "
1382                  + std::string((const char *) ptr->name)
1383                  + " in multi filter");
1384         }
1385     }
1386 }
1387
1388 static mp::filter::Base* filter_creator()
1389 {
1390     return new mp::filter::Multi;
1391 }
1392
1393 extern "C" {
1394     struct metaproxy_1_filter_struct metaproxy_1_filter_multi = {
1395         0,
1396         "multi",
1397         filter_creator
1398     };
1399 }
1400
1401
1402 /*
1403  * Local variables:
1404  * c-basic-offset: 4
1405  * c-file-style: "Stroustrup"
1406  * indent-tabs-mode: nil
1407  * End:
1408  * vim: shiftwidth=4 tabstop=8 expandtab
1409  */
1410