Deal with OPAC records . These are converted to XML but character
[pazpar2-moved-to-github.git] / src / logic.c
1 /* $Id: logic.c,v 1.49 2007-07-06 14:31:06 adam Exp $
2    Copyright (c) 2006-2007, Index Data.
3
4 This file is part of Pazpar2.
5
6 Pazpar2 is free software; you can redistribute it and/or modify it under
7 the terms of the GNU General Public License as published by the Free
8 Software Foundation; either version 2, or (at your option) any later
9 version.
10
11 Pazpar2 is distributed in the hope that it will be useful, but WITHOUT ANY
12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
14 for more details.
15
16 You should have received a copy of the GNU General Public License
17 along with Pazpar2; see the file LICENSE.  If not, write to the
18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
19 02111-1307, USA.
20  */
21
22 /** \file logic.c
23     \brief high-level logic; mostly user sessions and settings
24 */
25
26 #include <stdlib.h>
27 #include <stdio.h>
28 #include <string.h>
29 #include <sys/time.h>
30 #include <unistd.h>
31 #include <sys/socket.h>
32 #include <netdb.h>
33 #include <signal.h>
34 #include <ctype.h>
35 #include <assert.h>
36
37 #include <yaz/marcdisp.h>
38 #include <yaz/comstack.h>
39 #include <yaz/tcpip.h>
40 #include <yaz/proto.h>
41 #include <yaz/readconf.h>
42 #include <yaz/pquery.h>
43 #include <yaz/otherinfo.h>
44 #include <yaz/yaz-util.h>
45 #include <yaz/nmem.h>
46 #include <yaz/query-charset.h>
47 #include <yaz/querytowrbuf.h>
48 #include <yaz/oid_db.h>
49
50 #if HAVE_CONFIG_H
51 #include "cconfig.h"
52 #endif
53
54 #define USE_TIMING 0
55 #if USE_TIMING
56 #include <yaz/timing.h>
57 #endif
58
59 #include <netinet/in.h>
60
61 #include "pazpar2.h"
62 #include "eventl.h"
63 #include "http.h"
64 #include "termlists.h"
65 #include "reclists.h"
66 #include "relevance.h"
67 #include "config.h"
68 #include "database.h"
69 #include "client.h"
70 #include "settings.h"
71 #include "normalize7bit.h"
72
73 #define TERMLIST_HIGH_SCORE 25
74
75 #define MAX_CHUNK 15
76
77 // Note: Some things in this structure will eventually move to configuration
78 struct parameters global_parameters = 
79 {
80     "",
81     "",
82     "", 
83     0,
84     0, /* dump_records */
85     0, /* debug_mode */
86     30,
87     "81",
88     "Index Data PazPar2",
89     VERSION,
90     600, // 10 minutes
91     60,
92     100,
93     MAX_CHUNK,
94     0,
95     0
96 };
97
98
99 // Recursively traverse query structure to extract terms.
100 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
101 {
102     char **words;
103     int numwords;
104     int i;
105
106     switch (n->kind)
107     {
108         case CCL_RPN_AND:
109         case CCL_RPN_OR:
110         case CCL_RPN_NOT:
111         case CCL_RPN_PROX:
112             pull_terms(nmem, n->u.p[0], termlist, num);
113             pull_terms(nmem, n->u.p[1], termlist, num);
114             break;
115         case CCL_RPN_TERM:
116             nmem_strsplit(nmem, " ", n->u.t.term, &words, &numwords);
117             for (i = 0; i < numwords; i++)
118                 termlist[(*num)++] = words[i];
119             break;
120         default: // NOOP
121             break;
122     }
123 }
124
125
126
127 static void add_facet(struct session *s, const char *type, const char *value)
128 {
129     int i;
130
131     if (!*value)
132         return;
133     for (i = 0; i < s->num_termlists; i++)
134         if (!strcmp(s->termlists[i].name, type))
135             break;
136     if (i == s->num_termlists)
137     {
138         if (i == SESSION_MAX_TERMLISTS)
139         {
140             yaz_log(YLOG_FATAL, "Too many termlists");
141             return;
142         }
143
144         s->termlists[i].name = nmem_strdup(s->nmem, type);
145         s->termlists[i].termlist 
146             = termlist_create(s->nmem, s->expected_maxrecs,
147                               TERMLIST_HIGH_SCORE);
148         s->num_termlists = i + 1;
149     }
150     termlist_insert(s->termlists[i].termlist, value);
151 }
152
153 xmlDoc *record_to_xml(struct session_database *sdb, Z_External *rec)
154 {
155     struct database *db = sdb->database;
156     xmlDoc *rdoc = 0;
157     const Odr_oid *oid = rec->direct_reference;
158
159     /* convert response record to XML somehow */
160     if (rec->which == Z_External_octet && oid
161         && !oid_oidcmp(oid, yaz_oid_recsyn_xml))
162     {
163         /* xml already */
164         rdoc = xmlParseMemory((char*) rec->u.octet_aligned->buf,
165                               rec->u.octet_aligned->len);
166         if (!rdoc)
167         {
168             yaz_log(YLOG_FATAL, "Non-wellformed XML received from %s",
169                     db->url);
170             return 0;
171         }
172     }
173     else if (rec->which == Z_External_OPAC)
174     {
175         /* OPAC gets converted to XML too */
176         WRBUF wrbuf_opac = wrbuf_alloc();
177         yaz_display_OPAC(wrbuf_opac, rec->u.opac, 0);
178
179         rdoc = xmlParseMemory((char*) wrbuf_buf(wrbuf_opac),
180                               wrbuf_len(wrbuf_opac));
181         if (!rdoc)
182             yaz_log(YLOG_WARN, "Unable to parse OPAC XML");
183         wrbuf_destroy(wrbuf_opac);
184     }
185     else if (oid && yaz_oid_is_iso2709(oid))
186     {
187         /* ISO2709 gets converted to MARCXML */
188         if (!sdb->yaz_marc)
189         {
190             yaz_log(YLOG_FATAL, "Unable to handle ISO2709 record");
191             return 0;
192         }
193         else
194         {
195             xmlNode *res;
196             char *buf;
197             int len;
198             
199             if (rec->which != Z_External_octet)
200             {
201                 yaz_log(YLOG_WARN, "Unexpected external branch, probably BER %s",
202                         db->url);
203                 return 0;
204             }
205             buf = (char*) rec->u.octet_aligned->buf;
206             len = rec->u.octet_aligned->len;
207             if (yaz_marc_read_iso2709(sdb->yaz_marc, buf, len) < 0)
208             {
209                 yaz_log(YLOG_WARN, "Failed to decode MARC %s", db->url);
210                 return 0;
211             }
212             
213             yaz_marc_write_using_libxml2(sdb->yaz_marc, 1);
214             if (yaz_marc_write_xml(sdb->yaz_marc, &res,
215                                    "http://www.loc.gov/MARC21/slim", 0, 0) < 0)
216             {
217                 yaz_log(YLOG_WARN, "Failed to encode as XML %s",
218                         db->url);
219                 return 0;
220             }
221             rdoc = xmlNewDoc((xmlChar *) "1.0");
222             xmlDocSetRootElement(rdoc, res);
223         }
224     }
225     else
226     {
227         char oid_name_buf[OID_STR_MAX];
228         const char *oid_name = yaz_oid_to_string_buf(oid, 0, oid_name_buf);
229         yaz_log(YLOG_FATAL, 
230                 "Unable to handle record of type %s from %s", 
231                 oid_name, db->url);
232         return 0;
233     }
234     
235     if (global_parameters.dump_records){
236         fprintf(stderr, 
237                 "Input Record (normalized) from %s\n----------------\n",
238                 db->url);
239 #if LIBXML_VERSION >= 20600
240         xmlDocFormatDump(stderr, rdoc, 1);
241 #else
242         xmlDocDump(stderr, rdoc);
243 #endif
244     }
245     return rdoc;
246 }
247
248 xmlDoc *normalize_record(struct session_database *sdb, Z_External *rec)
249 {
250     struct database_retrievalmap *m;
251     xmlDoc *rdoc = record_to_xml(sdb, rec);
252     if (rdoc)
253     {
254         for (m = sdb->map; m; m = m->next){
255             xmlDoc *new = 0;
256             
257             {
258                 xmlNodePtr root = 0;
259                 new = xsltApplyStylesheet(m->stylesheet, rdoc, 0);
260                 root= xmlDocGetRootElement(new);
261                 if (!new || !root || !(root->children))
262                 {
263                     yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
264                             sdb->database->url);
265                     xmlFreeDoc(new);
266                     xmlFreeDoc(rdoc);
267                     return 0;
268                 }
269             }
270             
271             xmlFreeDoc(rdoc);
272             rdoc = new;
273         }
274         if (global_parameters.dump_records)
275         {
276             fprintf(stderr, "Record from %s\n----------------\n", 
277                     sdb->database->url);
278 #if LIBXML_VERSION >= 20600
279             xmlDocFormatDump(stderr, rdoc, 1);
280 #else
281             xmlDocDump(stderr, rdoc);
282 #endif
283         }
284     }
285     return rdoc;
286 }
287
288 // Retrieve first defined value for 'name' for given database.
289 // Will be extended to take into account user associated with session
290 char *session_setting_oneval(struct session_database *db, int offset)
291 {
292     if (!db->settings[offset])
293         return "";
294     return db->settings[offset]->value;
295 }
296
297
298
299 // Initialize YAZ Map structures for MARC-based targets
300 static int prepare_yazmarc(struct session_database *sdb)
301 {
302     char *s;
303
304     if (!sdb->settings)
305     {
306         yaz_log(YLOG_WARN, "No settings for %s", sdb->database->url);
307         return -1;
308     }
309     if ((s = session_setting_oneval(sdb, PZ_NATIVESYNTAX)) 
310         && !strncmp(s, "iso2709", 7))
311     {
312         char *encoding = "marc-8s", *e;
313         yaz_iconv_t cm;
314
315         // See if a native encoding is specified
316         if ((e = strchr(s, ';')))
317             encoding = e + 1;
318
319         sdb->yaz_marc = yaz_marc_create();
320         yaz_marc_subfield_str(sdb->yaz_marc, "\t");
321         
322         cm = yaz_iconv_open("utf-8", encoding);
323         if (!cm)
324         {
325             yaz_log(YLOG_FATAL, 
326                     "Unable to map from %s to UTF-8 for target %s", 
327                     encoding, sdb->database->url);
328             return -1;
329         }
330         yaz_marc_iconv(sdb->yaz_marc, cm);
331     }
332     return 0;
333 }
334
335 // Prepare XSLT stylesheets for record normalization
336 // Structures are allocated on the session_wide nmem to avoid having
337 // to recompute this for every search. This would lead
338 // to leaking if a single session was to repeatedly change the PZ_XSLT
339 // setting. However, this is not a realistic use scenario.
340 static int prepare_map(struct session *se, struct session_database *sdb)
341 {
342    char *s;
343
344     if (!sdb->settings)
345     {
346         yaz_log(YLOG_WARN, "No settings on %s", sdb->database->url);
347         return -1;
348     }
349     if ((s = session_setting_oneval(sdb, PZ_XSLT)))
350     {
351         char **stylesheets;
352         struct database_retrievalmap **m = &sdb->map;
353         int num, i;
354
355         nmem_strsplit(se->session_nmem, ",", s, &stylesheets, &num);
356         for (i = 0; i < num; i++)
357         {
358             (*m) = nmem_malloc(se->session_nmem, sizeof(**m));
359             (*m)->next = 0;
360             if (!((*m)->stylesheet = conf_load_stylesheet(stylesheets[i])))
361             {
362                 yaz_log(YLOG_FATAL|YLOG_ERRNO, "Unable to load stylesheet: %s",
363                         stylesheets[i]);
364                 return -1;
365             }
366             m = &(*m)->next;
367         }
368     }
369     if (!sdb->map)
370         yaz_log(YLOG_WARN, "No Normalization stylesheet for target %s",
371                 sdb->database->url);
372     return 0;
373 }
374
375 // This analyzes settings and recomputes any supporting data structures
376 // if necessary.
377 static int prepare_session_database(struct session *se, 
378                                     struct session_database *sdb)
379 {
380     if (!sdb->settings)
381     {
382         yaz_log(YLOG_WARN, 
383                 "No settings associated with %s", sdb->database->url);
384         return -1;
385     }
386     if (sdb->settings[PZ_NATIVESYNTAX] && !sdb->yaz_marc)
387     {
388         if (prepare_yazmarc(sdb) < 0)
389             return -1;
390     }
391     if (sdb->settings[PZ_XSLT] && !sdb->map)
392     {
393         if (prepare_map(se, sdb) < 0)
394             return -1;
395     }
396     return 0;
397 }
398
399 // called if watch should be removed because http_channel is to be destroyed
400 static void session_watch_cancel(void *data, struct http_channel *c)
401 {
402     struct session_watchentry *ent = data;
403
404     ent->fun = 0;
405     ent->data = 0;
406     ent->obs = 0;
407 }
408
409 // set watch. Returns 0=OK, -1 if watch is already set
410 int session_set_watch(struct session *s, int what, 
411                       session_watchfun fun, void *data,
412                       struct http_channel *chan)
413 {
414     if (s->watchlist[what].fun)
415         return -1;
416     s->watchlist[what].fun = fun;
417     s->watchlist[what].data = data;
418     s->watchlist[what].obs = http_add_observer(chan, &s->watchlist[what],
419                                                session_watch_cancel);
420     return 0;
421 }
422
423 void session_alert_watch(struct session *s, int what)
424 {
425     if (!s->watchlist[what].fun)
426         return;
427     http_remove_observer(s->watchlist[what].obs);
428     (*s->watchlist[what].fun)(s->watchlist[what].data);
429     s->watchlist[what].fun = 0;
430     s->watchlist[what].data = 0;
431     s->watchlist[what].obs = 0;
432 }
433
434 //callback for grep_databases
435 static void select_targets_callback(void *context, struct session_database *db)
436 {
437     struct session *se = (struct session*) context;
438     struct client *cl = client_create();
439     client_set_database(cl, db);
440     client_set_session(cl, se);
441 }
442
443 // Associates a set of clients with a session;
444 // Note: Session-databases represent databases with per-session 
445 // setting overrides
446 int select_targets(struct session *se, struct database_criterion *crit)
447 {
448     while (se->clients)
449         client_destroy(se->clients);
450
451     return session_grep_databases(se, crit, select_targets_callback);
452 }
453
454 int session_active_clients(struct session *s)
455 {
456     struct client *c;
457     int res = 0;
458
459     for (c = s->clients; c; c = client_next_in_session(c))
460         if (client_is_active(c))
461             res++;
462
463     return res;
464 }
465
466 // parses crit1=val1,crit2=val2|val3,...
467 static struct database_criterion *parse_filter(NMEM m, const char *buf)
468 {
469     struct database_criterion *res = 0;
470     char **values;
471     int num;
472     int i;
473
474     if (!buf || !*buf)
475         return 0;
476     nmem_strsplit(m, ",", buf,  &values, &num);
477     for (i = 0; i < num; i++)
478     {
479         char **subvalues;
480         int subnum;
481         int subi;
482         struct database_criterion *new = nmem_malloc(m, sizeof(*new));
483         char *eq = strchr(values[i], '=');
484         if (!eq)
485         {
486             yaz_log(YLOG_WARN, "Missing equal-sign in filter");
487             return 0;
488         }
489         *(eq++) = '\0';
490         new->name = values[i];
491         nmem_strsplit(m, "|", eq, &subvalues, &subnum);
492         new->values = 0;
493         for (subi = 0; subi < subnum; subi++)
494         {
495             struct database_criterion_value *newv
496                 = nmem_malloc(m, sizeof(*newv));
497             newv->value = subvalues[subi];
498             newv->next = new->values;
499             new->values = newv;
500         }
501         new->next = res;
502         res = new;
503     }
504     return res;
505 }
506
507 enum pazpar2_error_code search(struct session *se,
508                                char *query, char *filter,
509                                const char **addinfo)
510 {
511     int live_channels = 0;
512     int no_working = 0;
513     int no_failed = 0;
514     struct client *cl;
515     struct database_criterion *criteria;
516
517     yaz_log(YLOG_DEBUG, "Search");
518
519     *addinfo = 0;
520     nmem_reset(se->nmem);
521     criteria = parse_filter(se->nmem, filter);
522     se->requestid++;
523     live_channels = select_targets(se, criteria);
524     if (live_channels)
525     {
526         int maxrecs = live_channels * global_parameters.toget;
527         se->num_termlists = 0;
528         se->reclist = reclist_create(se->nmem, maxrecs);
529         // This will be initialized in send_search()
530         se->total_records = se->total_hits = se->total_merged = 0;
531         se->expected_maxrecs = maxrecs;
532     }
533     else
534         return PAZPAR2_NO_TARGETS;
535
536     se->relevance = 0;
537
538     for (cl = se->clients; cl; cl = client_next_in_session(cl))
539     {
540         if (prepare_session_database(se, client_get_database(cl)) < 0)
541         {
542             *addinfo = client_get_database(cl)->database->url;
543             return PAZPAR2_CONFIG_TARGET;
544         }
545         // Parse query for target
546         if (client_parse_query(cl, query) < 0)
547             no_failed++;
548         else
549         {
550             no_working++;
551             client_prep_connection(cl);
552         }
553     }
554
555     // If no queries could be mapped, we signal an error
556     if (no_working == 0)
557     {
558         *addinfo = "query";
559         return PAZPAR2_MALFORMED_PARAMETER_VALUE;
560     }
561     return PAZPAR2_NO_ERROR;
562 }
563
564 // Creates a new session_database object for a database
565 static void session_init_databases_fun(void *context, struct database *db)
566 {
567     struct session *se = (struct session *) context;
568     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
569     int num = settings_num();
570     int i;
571
572     new->database = db;
573     new->yaz_marc = 0;
574     
575 #ifdef HAVE_ICU
576     if (global_parameters.server && global_parameters.server->icu_chn)
577         new->pct = pp2_charset_create(global_parameters.server->icu_chn);
578     else
579         new->pct = pp2_charset_create(0);
580 #else // HAVE_ICU
581     new->pct = pp2_charset_create(0);
582 #endif // HAVE_ICU
583
584     new->map = 0;
585     new->settings 
586         = nmem_malloc(se->session_nmem, sizeof(struct settings *) * num);
587     memset(new->settings, 0, sizeof(struct settings*) * num);
588
589     if (db->settings)
590     {
591         for (i = 0; i < num; i++)
592             new->settings[i] = db->settings[i];
593     }
594     new->next = se->databases;
595     se->databases = new;
596 }
597
598 // Doesn't free memory associated with sdb -- nmem takes care of that
599 static void session_database_destroy(struct session_database *sdb)
600 {
601     struct database_retrievalmap *m;
602
603     for (m = sdb->map; m; m = m->next)
604         xsltFreeStylesheet(m->stylesheet);
605     if (sdb->yaz_marc)
606         yaz_marc_destroy(sdb->yaz_marc);
607     if (sdb->pct)
608         pp2_charset_destroy(sdb->pct);
609 }
610
611 // Initialize session_database list -- this represents this session's view
612 // of the database list -- subject to modification by the settings ws command
613 void session_init_databases(struct session *se)
614 {
615     se->databases = 0;
616     predef_grep_databases(se, 0, session_init_databases_fun);
617 }
618
619 // Probably session_init_databases_fun should be refactored instead of
620 // called here.
621 static struct session_database *load_session_database(struct session *se, 
622                                                       char *id)
623 {
624     struct database *db = find_database(id, 0);
625
626     session_init_databases_fun((void*) se, db);
627     // New sdb is head of se->databases list
628     return se->databases;
629 }
630
631 // Find an existing session database. If not found, load it
632 static struct session_database *find_session_database(struct session *se, 
633                                                       char *id)
634 {
635     struct session_database *sdb;
636
637     for (sdb = se->databases; sdb; sdb = sdb->next)
638         if (!strcmp(sdb->database->url, id))
639             return sdb;
640     return load_session_database(se, id);
641 }
642
643 // Apply a session override to a database
644 void session_apply_setting(struct session *se, char *dbname, char *setting,
645                            char *value)
646 {
647     struct session_database *sdb = find_session_database(se, dbname);
648     struct setting *new = nmem_malloc(se->session_nmem, sizeof(*new));
649     int offset = settings_offset_cprefix(setting);
650
651     if (offset < 0)
652     {
653         yaz_log(YLOG_WARN, "Unknown setting %s", setting);
654         return;
655     }
656     // Jakub: This breaks the filter setting.
657     /*if (offset == PZ_ID)
658     {
659         yaz_log(YLOG_WARN, "No need to set pz:id setting. Ignoring");
660         return;
661     }*/
662     new->precedence = 0;
663     new->target = dbname;
664     new->name = setting;
665     new->value = value;
666     new->next = sdb->settings[offset];
667     sdb->settings[offset] = new;
668
669     // Force later recompute of settings-driven data structures
670     // (happens when a search starts and client connections are prepared)
671     switch (offset)
672     {
673         case PZ_NATIVESYNTAX:
674             if (sdb->yaz_marc)
675             {
676                 yaz_marc_destroy(sdb->yaz_marc);
677                 sdb->yaz_marc = 0;
678             }
679             break;
680         case PZ_XSLT:
681             if (sdb->map)
682             {
683                 struct database_retrievalmap *m;
684                 // We don't worry about the map structure -- it's in nmem
685                 for (m = sdb->map; m; m = m->next)
686                     xsltFreeStylesheet(m->stylesheet);
687                 sdb->map = 0;
688             }
689             break;
690     }
691 }
692
693 void destroy_session(struct session *s)
694 {
695     struct session_database *sdb;
696
697     while (s->clients)
698         client_destroy(s->clients);
699     for (sdb = s->databases; sdb; sdb = sdb->next)
700         session_database_destroy(sdb);
701     nmem_destroy(s->nmem);
702     wrbuf_destroy(s->wrbuf);
703 }
704
705 struct session *new_session(NMEM nmem) 
706 {
707     int i;
708     struct session *session = nmem_malloc(nmem, sizeof(*session));
709
710     yaz_log(YLOG_DEBUG, "New Pazpar2 session");
711     
712     session->relevance = 0;
713     session->total_hits = 0;
714     session->total_records = 0;
715     session->num_termlists = 0;
716     session->reclist = 0;
717     session->requestid = -1;
718     session->clients = 0;
719     session->expected_maxrecs = 0;
720     session->session_nmem = nmem;
721     session->nmem = nmem_create();
722     session->wrbuf = wrbuf_alloc();
723     session->databases = 0;
724     for (i = 0; i <= SESSION_WATCH_MAX; i++)
725     {
726         session->watchlist[i].data = 0;
727         session->watchlist[i].fun = 0;
728     }
729
730     return session;
731 }
732
733 struct hitsbytarget *hitsbytarget(struct session *se, int *count)
734 {
735     static struct hitsbytarget res[1000]; // FIXME MM
736     struct client *cl;
737
738     *count = 0;
739     for (cl = se->clients; cl; cl = client_next_in_session(cl))
740     {
741         char *name = session_setting_oneval(client_get_database(cl), PZ_NAME);
742
743         res[*count].id = client_get_database(cl)->database->url;
744         res[*count].name = *name ? name : "Unknown";
745         res[*count].hits = client_get_hits(cl);
746         res[*count].records = client_get_num_records(cl);
747         res[*count].diagnostic = client_get_diagnostic(cl);
748         res[*count].state = client_get_state_str(cl);
749         res[*count].connected  = client_get_connection(cl) ? 1 : 0;
750         (*count)++;
751     }
752
753     return res;
754 }
755
756 struct termlist_score **termlist(struct session *s, const char *name, int *num)
757 {
758     int i;
759
760     for (i = 0; i < s->num_termlists; i++)
761         if (!strcmp((const char *) s->termlists[i].name, name))
762             return termlist_highscore(s->termlists[i].termlist, num);
763     return 0;
764 }
765
766 #ifdef MISSING_HEADERS
767 void report_nmem_stats(void)
768 {
769     size_t in_use, is_free;
770
771     nmem_get_memory_in_use(&in_use);
772     nmem_get_memory_free(&is_free);
773
774     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld", 
775             (long) in_use, (long) is_free);
776 }
777 #endif
778
779 struct record_cluster *show_single(struct session *s, int id)
780 {
781     struct record_cluster *r;
782
783     reclist_rewind(s->reclist);
784     while ((r = reclist_read_record(s->reclist)))
785         if (r->recid == id)
786             return r;
787     return 0;
788 }
789
790 struct record_cluster **show(struct session *s, struct reclist_sortparms *sp, 
791                              int start, int *num, int *total, int *sumhits, 
792                              NMEM nmem_show)
793 {
794     struct record_cluster **recs = nmem_malloc(nmem_show, *num 
795                                        * sizeof(struct record_cluster *));
796     struct reclist_sortparms *spp;
797     int i;
798 #if USE_TIMING    
799     yaz_timing_t t = yaz_timing_create();
800 #endif
801
802     if (!s->relevance)
803     {
804         *num = 0;
805         *total = 0;
806         *sumhits = 0;
807         recs = 0;
808     }
809     else
810     {
811         for (spp = sp; spp; spp = spp->next)
812             if (spp->type == Metadata_sortkey_relevance)
813             {
814                 relevance_prepare_read(s->relevance, s->reclist);
815                 break;
816             }
817         reclist_sort(s->reclist, sp);
818         
819         *total = s->reclist->num_records;
820         *sumhits = s->total_hits;
821         
822         for (i = 0; i < start; i++)
823             if (!reclist_read_record(s->reclist))
824             {
825                 *num = 0;
826                 recs = 0;
827                 break;
828             }
829         
830         for (i = 0; i < *num; i++)
831         {
832             struct record_cluster *r = reclist_read_record(s->reclist);
833             if (!r)
834             {
835                 *num = i;
836                 break;
837             }
838             recs[i] = r;
839         }
840     }
841 #if USE_TIMING
842     yaz_timing_stop(t);
843     yaz_log(YLOG_LOG, "show %6.5f %3.2f %3.2f", 
844             yaz_timing_get_real(t), yaz_timing_get_user(t),
845             yaz_timing_get_sys(t));
846     yaz_timing_destroy(&t);
847 #endif
848     return recs;
849 }
850
851 void statistics(struct session *se, struct statistics *stat)
852 {
853     struct client *cl;
854     int count = 0;
855
856     memset(stat, 0, sizeof(*stat));
857     for (cl = se->clients; cl; cl = client_next_in_session(cl))
858     {
859         if (!client_get_connection(cl))
860             stat->num_no_connection++;
861         switch (client_get_state(cl))
862         {
863             case Client_Connecting: stat->num_connecting++; break;
864             case Client_Initializing: stat->num_initializing++; break;
865             case Client_Searching: stat->num_searching++; break;
866             case Client_Presenting: stat->num_presenting++; break;
867             case Client_Idle: stat->num_idle++; break;
868             case Client_Failed: stat->num_failed++; break;
869             case Client_Error: stat->num_error++; break;
870             default: break;
871         }
872         count++;
873     }
874     stat->num_hits = se->total_hits;
875     stat->num_records = se->total_records;
876
877     stat->num_clients = count;
878 }
879
880 void start_http_listener(void)
881 {
882     char hp[128] = "";
883     struct conf_server *ser = global_parameters.server;
884
885     if (*global_parameters.listener_override)
886         strcpy(hp, global_parameters.listener_override);
887     else
888     {
889         strcpy(hp, ser->host ? ser->host : "");
890         if (ser->port)
891         {
892             if (*hp)
893                 strcat(hp, ":");
894             sprintf(hp + strlen(hp), "%d", ser->port);
895         }
896     }
897     http_init(hp);
898 }
899
900 void start_proxy(void)
901 {
902     char hp[128] = "";
903     struct conf_server *ser = global_parameters.server;
904
905     if (*global_parameters.proxy_override)
906         strcpy(hp, global_parameters.proxy_override);
907     else if (ser->proxy_host || ser->proxy_port)
908     {
909         strcpy(hp, ser->proxy_host ? ser->proxy_host : "");
910         if (ser->proxy_port)
911         {
912             if (*hp)
913                 strcat(hp, ":");
914             sprintf(hp + strlen(hp), "%d", ser->proxy_port);
915         }
916     }
917     else
918         return;
919
920     http_set_proxyaddr(hp, ser->myurl ? ser->myurl : "");
921 }
922
923
924 // Master list of connections we're handling events to
925 static IOCHAN channel_list = 0; 
926 void pazpar2_add_channel(IOCHAN chan)
927 {
928     chan->next = channel_list;
929     channel_list = chan;
930 }
931
932 void pazpar2_event_loop()
933 {
934     event_loop(&channel_list);
935 }
936
937 struct record *ingest_record(struct client *cl, Z_External *rec,
938                              int record_no)
939 {
940     xmlDoc *xdoc = normalize_record(client_get_database(cl), rec);
941     xmlNode *root, *n;
942     struct record *record;
943     struct record_cluster *cluster;
944     struct session *se = client_get_session(cl);
945     xmlChar *mergekey, *mergekey_norm;
946     xmlChar *type = 0;
947     xmlChar *value = 0;
948     struct conf_service *service = global_parameters.server->service;
949
950     if (!xdoc)
951         return 0;
952
953     root = xmlDocGetRootElement(xdoc);
954     if (!(mergekey = xmlGetProp(root, (xmlChar *) "mergekey")))
955     {
956         yaz_log(YLOG_WARN, "No mergekey found in record");
957         xmlFreeDoc(xdoc);
958         return 0;
959     }
960
961     record = record_create(se->nmem, 
962                            service->num_metadata, service->num_sortkeys, cl,
963                            record_no);
964
965     mergekey_norm = (xmlChar *) nmem_strdup(se->nmem, (char*) mergekey);
966     xmlFree(mergekey);
967     normalize7bit_mergekey((char *) mergekey_norm, 0);
968
969     cluster = reclist_insert(se->reclist, 
970                              global_parameters.server->service, 
971                              record, (char *) mergekey_norm, 
972                              &se->total_merged);
973     if (global_parameters.dump_records)
974         yaz_log(YLOG_LOG, "Cluster id %d from %s (#%d)", cluster->recid,
975                 client_get_database(cl)->database->url, record_no);
976     if (!cluster)
977     {
978         /* no room for record */
979         xmlFreeDoc(xdoc);
980         return 0;
981     }
982     relevance_newrec(se->relevance, cluster);
983
984
985     // now parsing XML record and adding data to cluster or record metadata
986     for (n = root->children; n; n = n->next)
987     {
988         if (type)
989             xmlFree(type);
990         if (value)
991             xmlFree(value);
992         type = value = 0;
993
994         if (n->type != XML_ELEMENT_NODE)
995             continue;
996         if (!strcmp((const char *) n->name, "metadata"))
997         {
998             struct conf_metadata *ser_md = 0;
999             struct conf_sortkey *ser_sk = 0;
1000             struct record_metadata **wheretoput = 0;
1001             struct record_metadata *rec_md = 0;
1002             int md_field_id = -1;
1003             int sk_field_id = -1;
1004             int first, last;
1005
1006             type = xmlGetProp(n, (xmlChar *) "type");
1007             value = xmlNodeListGetString(xdoc, n->children, 1);
1008
1009             if (!type || !value)
1010                 continue;
1011
1012             md_field_id 
1013                 = conf_service_metadata_field_id(service, (const char *) type);
1014             if (md_field_id < 0)
1015             {
1016                 yaz_log(YLOG_WARN, 
1017                         "Ignoring unknown metadata element: %s", type);
1018                 continue;
1019             }
1020
1021             ser_md = &service->metadata[md_field_id];
1022
1023             if (ser_md->sortkey_offset >= 0){
1024                 sk_field_id = ser_md->sortkey_offset;
1025                 ser_sk = &service->sortkeys[sk_field_id];
1026             }
1027
1028             // Find out where we are putting it - based on merge or not
1029             if (ser_md->merge == Metadata_merge_no)
1030                 wheretoput = &record->metadata[md_field_id];
1031             else
1032                 wheretoput = &cluster->metadata[md_field_id];
1033             
1034             // create new record_metadata
1035             rec_md = record_metadata_create(se->nmem);
1036
1037             // and polulate with data:
1038             // type based charmapping decisions follow here
1039             if (ser_md->type == Metadata_type_generic)
1040             {
1041
1042                 char * p = (char *) value;
1043                 p = normalize7bit_generic(p, " ,/.:([");
1044                 
1045                 rec_md->data.text = nmem_strdup(se->nmem, p);
1046
1047             }
1048             else if (ser_md->type == Metadata_type_year)
1049             {
1050                 if (extract7bit_years((char *) value, &first, &last) < 0)
1051                     continue;
1052             }
1053             else
1054             {
1055                 yaz_log(YLOG_WARN, 
1056                         "Unknown type in metadata element %s", type);
1057                 continue;
1058             }
1059
1060             // and polulate with data:
1061             // assign cluster or record based on merge action
1062             if (ser_md->merge == Metadata_merge_unique)
1063             {
1064                 struct record_metadata *mnode;
1065                 for (mnode = *wheretoput; mnode; mnode = mnode->next)
1066                     if (!strcmp((const char *) mnode->data.text, 
1067                                 rec_md->data.text))
1068                         break;
1069                 if (!mnode)
1070                 {
1071                     rec_md->next = *wheretoput;
1072                     *wheretoput = rec_md;
1073                 }
1074             }
1075             else if (ser_md->merge == Metadata_merge_longest)
1076             {
1077                 if (!*wheretoput 
1078                     || strlen(rec_md->data.text) 
1079                        > strlen((*wheretoput)->data.text))
1080                 {
1081                     *wheretoput = rec_md;
1082                     if (ser_sk)
1083                     {
1084                         char *s = nmem_strdup(se->nmem, rec_md->data.text);
1085                         if (!cluster->sortkeys[sk_field_id])
1086                             cluster->sortkeys[sk_field_id] = 
1087                                 nmem_malloc(se->nmem, 
1088                                             sizeof(union data_types));
1089                         normalize7bit_mergekey(s,
1090                              (ser_sk->type == Metadata_sortkey_skiparticle));
1091                         cluster->sortkeys[sk_field_id]->text = s;
1092                     }
1093                 }
1094             }
1095             else if (ser_md->merge == Metadata_merge_all 
1096                      || ser_md->merge == Metadata_merge_no)
1097             {
1098                 rec_md->next = *wheretoput;
1099                 *wheretoput = rec_md;
1100             }
1101             else if (ser_md->merge == Metadata_merge_range)
1102             {
1103                 if (!*wheretoput)
1104                 {
1105                     *wheretoput = rec_md;
1106                     (*wheretoput)->data.number.min = first;
1107                     (*wheretoput)->data.number.max = last;
1108                     if (ser_sk)
1109                         cluster->sortkeys[sk_field_id] 
1110                             = &rec_md->data;
1111                 }
1112                 else
1113                 {
1114                     if (first < (*wheretoput)->data.number.min)
1115                         (*wheretoput)->data.number.min = first;
1116                     if (last > (*wheretoput)->data.number.max)
1117                         (*wheretoput)->data.number.max = last;
1118                 }
1119 #ifdef GAGA
1120                 if (ser_sk)
1121                 {
1122                     union data_types *sdata 
1123                         = cluster->sortkeys[sk_field_id];
1124                     yaz_log(YLOG_LOG, "SK range: %d-%d",
1125                             sdata->number.min, sdata->number.max);
1126                 }
1127 #endif
1128             }
1129
1130
1131             // ranking of _all_ fields enabled ... 
1132             if (ser_md->rank)
1133                 relevance_countwords(se->relevance, cluster, 
1134                                      (char *) value, ser_md->rank);
1135
1136             // construct facets ... 
1137             if (ser_md->termlist)
1138             {
1139                 if (ser_md->type == Metadata_type_year)
1140                 {
1141                     char year[64];
1142                     sprintf(year, "%d", last);
1143                     add_facet(se, (char *) type, year);
1144                     if (first != last)
1145                     {
1146                         sprintf(year, "%d", first);
1147                         add_facet(se, (char *) type, year);
1148                     }
1149                 }
1150                 else
1151                     add_facet(se, (char *) type, (char *) value);
1152             }
1153
1154             // cleaning up
1155             xmlFree(type);
1156             xmlFree(value);
1157             type = value = 0;
1158         }
1159         else
1160             yaz_log(YLOG_WARN,
1161                     "Unexpected element %s in internal record", n->name);
1162     }
1163     if (type)
1164         xmlFree(type);
1165     if (value)
1166         xmlFree(value);
1167
1168     xmlFreeDoc(xdoc);
1169
1170     relevance_donerecord(se->relevance, cluster);
1171     se->total_records++;
1172
1173     return record;
1174 }
1175
1176
1177
1178 /*
1179  * Local variables:
1180  * c-basic-offset: 4
1181  * indent-tabs-mode: nil
1182  * End:
1183  * vim: shiftwidth=4 tabstop=8 expandtab
1184  */