Introduce client_locking
[pazpar2-moved-to-github.git] / src / session.c
1 /* This file is part of Pazpar2.
2    Copyright (C) 2006-2010 Index Data
3
4 Pazpar2 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 Pazpar2 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
20 /** \file session.c
21     \brief high-level logic; mostly user sessions and settings
22 */
23
24 #if HAVE_CONFIG_H
25 #include <config.h>
26 #endif
27
28 #include <stdlib.h>
29 #include <stdio.h>
30 #include <string.h>
31 #if HAVE_SYS_TIME_H
32 #include <sys/time.h>
33 #endif
34 #if HAVE_UNISTD_H
35 #include <unistd.h>
36 #endif
37 #include <signal.h>
38 #include <ctype.h>
39 #include <assert.h>
40
41 #include <yaz/marcdisp.h>
42 #include <yaz/comstack.h>
43 #include <yaz/tcpip.h>
44 #include <yaz/proto.h>
45 #include <yaz/readconf.h>
46 #include <yaz/pquery.h>
47 #include <yaz/otherinfo.h>
48 #include <yaz/yaz-util.h>
49 #include <yaz/nmem.h>
50 #include <yaz/query-charset.h>
51 #include <yaz/querytowrbuf.h>
52 #include <yaz/oid_db.h>
53 #include <yaz/snprintf.h>
54
55 #define USE_TIMING 0
56 #if USE_TIMING
57 #include <yaz/timing.h>
58 #endif
59
60 #include "ppmutex.h"
61 #include "parameters.h"
62 #include "session.h"
63 #include "eventl.h"
64 #include "http.h"
65 #include "termlists.h"
66 #include "reclists.h"
67 #include "relevance.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     0,   // dump_records
81     0    // debug_mode
82 };
83
84 struct client_list {
85     struct client *client;
86     struct client_list *next;
87 };
88
89 static void log_xml_doc(xmlDoc *doc)
90 {
91     FILE *lf = yaz_log_file();
92     xmlChar *result = 0;
93     int len = 0;
94 #if LIBXML_VERSION >= 20600
95     xmlDocDumpFormatMemory(doc, &result, &len, 1);
96 #else
97     xmlDocDumpMemory(doc, &result, &len);
98 #endif
99     if (lf && len)
100     {
101         (void) fwrite(result, 1, len, lf);
102         fprintf(lf, "\n");
103     }
104     xmlFree(result);
105 }
106
107 static void session_enter(struct session *s)
108 {
109     yaz_mutex_enter(s->session_mutex);
110 }
111
112 static void session_leave(struct session *s)
113 {
114     yaz_mutex_leave(s->session_mutex);
115 }
116
117 // Recursively traverse query structure to extract terms.
118 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
119 {
120     char **words;
121     int numwords;
122     int i;
123
124     switch (n->kind)
125     {
126     case CCL_RPN_AND:
127     case CCL_RPN_OR:
128     case CCL_RPN_NOT:
129     case CCL_RPN_PROX:
130         pull_terms(nmem, n->u.p[0], termlist, num);
131         pull_terms(nmem, n->u.p[1], termlist, num);
132         break;
133     case CCL_RPN_TERM:
134         nmem_strsplit(nmem, " ", n->u.t.term, &words, &numwords);
135         for (i = 0; i < numwords; i++)
136             termlist[(*num)++] = words[i];
137         break;
138     default: // NOOP
139         break;
140     }
141 }
142
143
144 static void add_facet(struct session *s, const char *type, const char *value)
145 {
146     int i;
147
148     if (!*value)
149         return;
150     for (i = 0; i < s->num_termlists; i++)
151         if (!strcmp(s->termlists[i].name, type))
152             break;
153     if (i == s->num_termlists)
154     {
155         if (i == SESSION_MAX_TERMLISTS)
156         {
157             yaz_log(YLOG_FATAL, "Too many termlists");
158             return;
159         }
160
161         s->termlists[i].name = nmem_strdup(s->nmem, type);
162         s->termlists[i].termlist 
163             = termlist_create(s->nmem, TERMLIST_HIGH_SCORE);
164         s->num_termlists = i + 1;
165     }
166     termlist_insert(s->termlists[i].termlist, value);
167 }
168
169 static xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
170 {
171     struct database *db = sdb->database;
172     xmlDoc *rdoc = 0;
173
174     rdoc = xmlParseMemory(rec, strlen(rec));
175
176     if (!rdoc)
177     {
178         yaz_log(YLOG_FATAL, "Non-wellformed XML received from %s",
179                 db->url);
180         return 0;
181     }
182
183     if (global_parameters.dump_records)
184     {
185         yaz_log(YLOG_LOG, "Un-normalized record from %s", db->url);
186         log_xml_doc(rdoc);
187     }
188
189     return rdoc;
190 }
191
192 #define MAX_XSLT_ARGS 16
193
194 // Add static values from session database settings if applicable
195 static void insert_settings_parameters(struct session_database *sdb,
196                                        struct conf_service *service,
197                                        char **parms,
198                                        NMEM nmem)
199 {
200     int i;
201     int nparms = 0;
202     int offset = 0;
203
204     for (i = 0; i < service->num_metadata; i++)
205     {
206         struct conf_metadata *md = &service->metadata[i];
207         int setting;
208
209         if (md->setting == Metadata_setting_parameter &&
210             (setting = settings_lookup_offset(service, md->name)) >= 0)
211         {
212             const char *val = session_setting_oneval(sdb, setting);
213             if (val && nparms < MAX_XSLT_ARGS)
214             {
215                 char *buf;
216                 int len = strlen(val);
217                 buf = nmem_malloc(nmem, len + 3);
218                 buf[0] = '\'';
219                 strcpy(buf + 1, val);
220                 buf[len+1] = '\'';
221                 buf[len+2] = '\0';
222                 parms[offset++] = md->name;
223                 parms[offset++] = buf;
224                 nparms++;
225             }
226         }
227     }
228     parms[offset] = 0;
229 }
230
231 // Add static values from session database settings if applicable
232 static void insert_settings_values(struct session_database *sdb, xmlDoc *doc,
233     struct conf_service *service)
234 {
235     int i;
236
237     for (i = 0; i < service->num_metadata; i++)
238     {
239         struct conf_metadata *md = &service->metadata[i];
240         int offset;
241
242         if (md->setting == Metadata_setting_postproc &&
243             (offset = settings_lookup_offset(service, md->name)) >= 0)
244         {
245             const char *val = session_setting_oneval(sdb, offset);
246             if (val)
247             {
248                 xmlNode *r = xmlDocGetRootElement(doc);
249                 xmlNode *n = xmlNewTextChild(r, 0, (xmlChar *) "metadata",
250                                              (xmlChar *) val);
251                 xmlSetProp(n, (xmlChar *) "type", (xmlChar *) md->name);
252             }
253         }
254     }
255 }
256
257 static xmlDoc *normalize_record(struct session_database *sdb,
258                                 struct conf_service *service,
259                                 const char *rec, NMEM nmem)
260 {
261     xmlDoc *rdoc = record_to_xml(sdb, rec);
262
263     if (rdoc)
264     {
265         char *parms[MAX_XSLT_ARGS*2+1];
266         
267         insert_settings_parameters(sdb, service, parms, nmem);
268         
269         if (normalize_record_transform(sdb->map, &rdoc, (const char **)parms))
270         {
271             yaz_log(YLOG_WARN, "Normalize failed from %s", sdb->database->url);
272         }
273         else
274         {
275             insert_settings_values(sdb, rdoc, service);
276             
277             if (global_parameters.dump_records)
278             {
279                 yaz_log(YLOG_LOG, "Normalized record from %s", 
280                         sdb->database->url);
281                 log_xml_doc(rdoc);
282             }
283         }
284     }
285     return rdoc;
286 }
287
288 void session_settings_dump(struct session *se,
289                            struct session_database *db,
290                            WRBUF w)
291 {
292     if (db->settings)
293     {
294         int i, num = db->num_settings;
295         for (i = 0; i < num; i++)
296         {
297             struct setting *s = db->settings[i];
298             for (;s ; s = s->next)
299             {
300                 wrbuf_puts(w, "<set name=\"");
301                 wrbuf_xmlputs(w, s->name);
302                 wrbuf_puts(w, "\" value=\"");
303                 wrbuf_xmlputs(w, s->value);
304                 wrbuf_puts(w, "\"/>");
305             }
306             if (db->settings[i])
307                 wrbuf_puts(w, "\n");
308         }
309     }
310 }
311
312 // Retrieve first defined value for 'name' for given database.
313 // Will be extended to take into account user associated with session
314 const char *session_setting_oneval(struct session_database *db, int offset)
315 {
316     if (offset >= db->num_settings || !db->settings[offset])
317         return "";
318     return db->settings[offset]->value;
319 }
320
321 // Prepare XSLT stylesheets for record normalization
322 // Structures are allocated on the session_wide nmem to avoid having
323 // to recompute this for every search. This would lead
324 // to leaking if a single session was to repeatedly change the PZ_XSLT
325 // setting. However, this is not a realistic use scenario.
326 static int prepare_map(struct session *se, struct session_database *sdb)
327 {
328     const char *s;
329
330     if (!sdb->settings)
331     {
332         yaz_log(YLOG_WARN, "No settings on %s", sdb->database->url);
333         return -1;
334     }
335     if ((s = session_setting_oneval(sdb, PZ_XSLT)))
336     {
337         char auto_stylesheet[256];
338
339         if (!strcmp(s, "auto"))
340         {
341             const char *request_syntax = session_setting_oneval(
342                 sdb, PZ_REQUESTSYNTAX);
343             if (request_syntax)
344             {
345                 char *cp;
346                 yaz_snprintf(auto_stylesheet, sizeof(auto_stylesheet),
347                              "%s.xsl", request_syntax);
348                 for (cp = auto_stylesheet; *cp; cp++)
349                 {
350                     /* deliberately only consider ASCII */
351                     if (*cp > 32 && *cp < 127)
352                         *cp = tolower(*cp);
353                 }
354                 s = auto_stylesheet;
355             }
356             else
357             {
358                 yaz_log(YLOG_WARN, "No pz:requestsyntax for auto stylesheet");
359             }
360         }
361         sdb->map = normalize_cache_get(se->normalize_cache,
362                                        se->service, s);
363         if (!sdb->map)
364             return -1;
365     }
366     return 0;
367 }
368
369 // This analyzes settings and recomputes any supporting data structures
370 // if necessary.
371 static int prepare_session_database(struct session *se, 
372                                     struct session_database *sdb)
373 {
374     if (!sdb->settings)
375     {
376         yaz_log(YLOG_WARN, 
377                 "No settings associated with %s", sdb->database->url);
378         return -1;
379     }
380     if (sdb->settings[PZ_XSLT] && !sdb->map)
381     {
382         if (prepare_map(se, sdb) < 0)
383             return -1;
384     }
385     return 0;
386 }
387
388 // called if watch should be removed because http_channel is to be destroyed
389 static void session_watch_cancel(void *data, struct http_channel *c,
390                                  void *data2)
391 {
392     struct session_watchentry *ent = data;
393
394     ent->fun = 0;
395     ent->data = 0;
396     ent->obs = 0;
397 }
398
399 // set watch. Returns 0=OK, -1 if watch is already set
400 int session_set_watch(struct session *s, int what, 
401                       session_watchfun fun, void *data,
402                       struct http_channel *chan)
403 {
404     int ret;
405     session_enter(s);
406     if (s->watchlist[what].fun)
407         ret = -1;
408     else
409     {
410         
411         s->watchlist[what].fun = fun;
412         s->watchlist[what].data = data;
413         s->watchlist[what].obs = http_add_observer(chan, &s->watchlist[what],
414                                                    session_watch_cancel);
415         ret = 0;
416     }
417     session_leave(s);
418     return 0;
419 }
420
421 void session_alert_watch(struct session *s, int what)
422 {
423     assert(s);
424     session_enter(s);
425     if (s->watchlist[what].fun)
426     {
427         /* our watch is no longer associated with http_channel */
428         void *data;
429         session_watchfun fun;
430
431         http_remove_observer(s->watchlist[what].obs);
432         fun = s->watchlist[what].fun;
433         data = s->watchlist[what].data;
434
435         /* reset watch before fun is invoked - in case fun wants to set
436            it again */
437         s->watchlist[what].fun = 0;
438         s->watchlist[what].data = 0;
439         s->watchlist[what].obs = 0;
440
441         session_leave(s);
442         fun(data);
443     }
444     else
445         session_leave(s);
446 }
447
448 //callback for grep_databases
449 static void select_targets_callback(void *context, struct session_database *db)
450 {
451     struct session *se = (struct session*) context;
452     struct client *cl = client_create();
453     struct client_list *l;
454     client_set_database(cl, db);
455
456     client_set_session(cl, se);
457     l = xmalloc(sizeof(*l));
458     l->client = cl;
459     l->next = se->clients;
460     se->clients = l;
461 }
462
463 static void session_remove_clients(struct session *se)
464 {
465     struct client_list *l;
466
467     session_enter(se);
468     l = se->clients;
469     se->clients = 0;
470     session_leave(se);
471
472     while (l)
473     {
474         struct client_list *l_next = l->next;
475         client_lock(l->client);
476         client_set_session(l->client, 0);
477         client_set_database(l->client, 0);
478         client_unlock(l->client);
479         client_destroy(l->client);
480         xfree(l);
481         l = l_next;
482     }
483 }
484
485 // Associates a set of clients with a session;
486 // Note: Session-databases represent databases with per-session 
487 // setting overrides
488 static int select_targets(struct session *se, const char *filter)
489 {
490     return session_grep_databases(se, filter, select_targets_callback);
491 }
492
493 int session_active_clients(struct session *s)
494 {
495     struct client_list *l;
496     int res = 0;
497
498     for (l = s->clients; l; l = l->next)
499         if (client_is_active(l->client))
500             res++;
501
502     return res;
503 }
504
505
506 enum pazpar2_error_code search(struct session *se,
507                                const char *query,
508                                const char *startrecs, const char *maxrecs,
509                                const char *filter,
510                                const char **addinfo)
511 {
512     int live_channels = 0;
513     int no_working = 0;
514     int no_failed = 0;
515     struct client_list *l;
516
517     yaz_log(YLOG_DEBUG, "Search");
518
519     *addinfo = 0;
520
521     session_remove_clients(se);
522     
523     session_enter(se);
524     reclist_destroy(se->reclist);
525     se->reclist = 0;
526     nmem_reset(se->nmem);
527     relevance_destroy(&se->relevance);
528     se->total_records = se->total_hits = se->total_merged = 0;
529     se->num_termlists = 0;
530     live_channels = select_targets(se, filter);
531     if (!live_channels)
532     {
533         session_leave(se);
534         return PAZPAR2_NO_TARGETS;
535     }
536     se->reclist = reclist_create(se->nmem);
537
538     for (l = se->clients; l; l = l->next)
539     {
540         struct client *cl = l->client;
541
542         if (maxrecs)
543             client_set_maxrecs(cl, atoi(maxrecs));
544         if (startrecs)
545             client_set_startrecs(cl, atoi(startrecs));
546         if (prepare_session_database(se, client_get_database(cl)) < 0)
547             ;
548         else if (client_parse_query(cl, query) < 0)
549             no_failed++;
550         else
551         {
552             no_working++;
553             if (client_prep_connection(cl, se->service->z3950_operation_timeout,
554                                        se->service->z3950_session_timeout,
555                                        se->service->server->iochan_man))
556                 client_start_search(cl);
557         }
558     }
559     session_leave(se);
560     if (no_working == 0)
561     {
562         if (no_failed > 0)
563         {
564             *addinfo = "query";
565             return PAZPAR2_MALFORMED_PARAMETER_VALUE;
566         }
567         else
568             return PAZPAR2_NO_TARGETS;
569     }
570     return PAZPAR2_NO_ERROR;
571 }
572
573 // Creates a new session_database object for a database
574 static void session_init_databases_fun(void *context, struct database *db)
575 {
576     struct session *se = (struct session *) context;
577     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
578     int i;
579
580     new->database = db;
581     
582     new->map = 0;
583     assert(db->settings);
584     new->settings = nmem_malloc(se->session_nmem,
585                                 sizeof(struct settings *) * db->num_settings);
586     new->num_settings = db->num_settings;
587     for (i = 0; i < db->num_settings; i++)
588     {
589         struct setting *setting = db->settings[i];
590         new->settings[i] = setting;
591     }
592     new->next = se->databases;
593     se->databases = new;
594 }
595
596 // Doesn't free memory associated with sdb -- nmem takes care of that
597 static void session_database_destroy(struct session_database *sdb)
598 {
599     sdb->map = 0;
600 }
601
602 // Initialize session_database list -- this represents this session's view
603 // of the database list -- subject to modification by the settings ws command
604 void session_init_databases(struct session *se)
605 {
606     se->databases = 0;
607     predef_grep_databases(se, se->service, session_init_databases_fun);
608 }
609
610 // Probably session_init_databases_fun should be refactored instead of
611 // called here.
612 static struct session_database *load_session_database(struct session *se, 
613                                                       char *id)
614 {
615     struct database *db = new_database(id, se->session_nmem);
616
617     resolve_database(se->service, db);
618
619     session_init_databases_fun((void*) se, db);
620
621     // New sdb is head of se->databases list
622     return se->databases;
623 }
624
625 // Find an existing session database. If not found, load it
626 static struct session_database *find_session_database(struct session *se, 
627                                                       char *id)
628 {
629     struct session_database *sdb;
630
631     for (sdb = se->databases; sdb; sdb = sdb->next)
632         if (!strcmp(sdb->database->url, id))
633             return sdb;
634     return load_session_database(se, id);
635 }
636
637 // Apply a session override to a database
638 void session_apply_setting(struct session *se, char *dbname, char *setting,
639                            char *value)
640 {
641     struct session_database *sdb = find_session_database(se, dbname);
642     struct conf_service *service = se->service;
643     struct setting *new = nmem_malloc(se->session_nmem, sizeof(*new));
644     int offset = settings_create_offset(service, setting);
645
646     expand_settings_array(&sdb->settings, &sdb->num_settings, offset,
647                           se->session_nmem);
648     new->precedence = 0;
649     new->target = dbname;
650     new->name = setting;
651     new->value = value;
652     new->next = sdb->settings[offset];
653     sdb->settings[offset] = new;
654
655     // Force later recompute of settings-driven data structures
656     // (happens when a search starts and client connections are prepared)
657     switch (offset)
658     {
659     case PZ_XSLT:
660         if (sdb->map)
661         {
662             sdb->map = 0;
663         }
664         break;
665     }
666 }
667
668 void destroy_session(struct session *s)
669 {
670     struct session_database *sdb;
671
672     session_remove_clients(s);
673
674     for (sdb = s->databases; sdb; sdb = sdb->next)
675         session_database_destroy(sdb);
676     normalize_cache_destroy(s->normalize_cache);
677     relevance_destroy(&s->relevance);
678     reclist_destroy(s->reclist);
679     nmem_destroy(s->nmem);
680     service_destroy(s->service);
681     yaz_mutex_destroy(&s->session_mutex);
682     wrbuf_destroy(s->wrbuf);
683 }
684
685 struct session *new_session(NMEM nmem, struct conf_service *service,
686                             const char *name)
687 {
688     int i;
689     struct session *session = nmem_malloc(nmem, sizeof(*session));
690
691     yaz_log(YLOG_DEBUG, "New Pazpar2 session");
692
693     session->service = service;
694     session->relevance = 0;
695     session->total_hits = 0;
696     session->total_records = 0;
697     session->number_of_warnings_unknown_elements = 0;
698     session->number_of_warnings_unknown_metadata = 0;
699     session->num_termlists = 0;
700     session->reclist = 0;
701     session->clients = 0;
702     session->session_nmem = nmem;
703     session->nmem = nmem_create();
704     session->wrbuf = wrbuf_alloc();
705     session->databases = 0;
706     for (i = 0; i <= SESSION_WATCH_MAX; i++)
707     {
708         session->watchlist[i].data = 0;
709         session->watchlist[i].fun = 0;
710     }
711     session->normalize_cache = normalize_cache_create();
712     session->session_mutex = 0;
713     pazpar2_mutex_create(&session->session_mutex, name);
714
715     return session;
716 }
717
718 struct hitsbytarget *hitsbytarget(struct session *se, int *count, NMEM nmem)
719 {
720     struct hitsbytarget *res = 0;
721     struct client_list *l;
722     size_t sz = 0;
723
724     session_enter(se);
725     for (l = se->clients; l; l = l->next)
726         sz++;
727
728     res = nmem_malloc(nmem, sizeof(*res) * sz);
729     *count = 0;
730     for (l = se->clients; l; l = l->next)
731     {
732         struct client *cl = l->client;
733         WRBUF w = wrbuf_alloc();
734         const char *name = session_setting_oneval(client_get_database(cl),
735                                                   PZ_NAME);
736
737         res[*count].id = client_get_database(cl)->database->url;
738         res[*count].name = *name ? name : "Unknown";
739         res[*count].hits = client_get_hits(cl);
740         res[*count].records = client_get_num_records(cl);
741         res[*count].diagnostic = client_get_diagnostic(cl);
742         res[*count].state = client_get_state_str(cl);
743         res[*count].connected  = client_get_connection(cl) ? 1 : 0;
744         session_settings_dump(se, client_get_database(cl), w);
745         res[*count].settings_xml = w;
746         (*count)++;
747     }
748     session_leave(se);
749     return res;
750 }
751
752 struct termlist_score **termlist(struct session *s, const char *name, int *num)
753 {
754     int i;
755     struct termlist_score **tl = 0;
756
757     session_enter(s);
758     for (i = 0; i < s->num_termlists; i++)
759         if (!strcmp((const char *) s->termlists[i].name, name))
760         {
761             tl = termlist_highscore(s->termlists[i].termlist, num);
762             break;
763         }
764     session_leave(s);
765     return tl;
766 }
767
768 #ifdef MISSING_HEADERS
769 void report_nmem_stats(void)
770 {
771     size_t in_use, is_free;
772
773     nmem_get_memory_in_use(&in_use);
774     nmem_get_memory_free(&is_free);
775
776     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld", 
777             (long) in_use, (long) is_free);
778 }
779 #endif
780
781 struct record_cluster *show_single_start(struct session *s, const char *id,
782                                          struct record_cluster **prev_r,
783                                          struct record_cluster **next_r)
784 {
785     struct record_cluster *r;
786
787     session_enter(s);
788     reclist_enter(s->reclist);
789     *prev_r = 0;
790     *next_r = 0;
791     while ((r = reclist_read_record(s->reclist)))
792     {
793         if (!strcmp(r->recid, id))
794         {
795             *next_r = reclist_read_record(s->reclist);
796             break;
797         }
798         *prev_r = r;
799     }
800     reclist_leave(s->reclist);
801     if (!r)
802         session_leave(s);
803     return r;
804 }
805
806 void show_single_stop(struct session *s, struct record_cluster *rec)
807 {
808     session_leave(s);
809 }
810
811 struct record_cluster **show_range_start(struct session *s,
812                                          struct reclist_sortparms *sp, 
813                                          int start, int *num, int *total, Odr_int *sumhits)
814 {
815     struct record_cluster **recs = nmem_malloc(s->nmem, *num 
816                                                * sizeof(struct record_cluster *));
817     struct reclist_sortparms *spp;
818     int i;
819 #if USE_TIMING    
820     yaz_timing_t t = yaz_timing_create();
821 #endif
822
823     session_enter(s);
824     if (!s->relevance)
825     {
826         *num = 0;
827         *total = 0;
828         *sumhits = 0;
829         recs = 0;
830     }
831     else
832     {
833         for (spp = sp; spp; spp = spp->next)
834             if (spp->type == Metadata_sortkey_relevance)
835             {
836                 relevance_prepare_read(s->relevance, s->reclist);
837                 break;
838             }
839         reclist_sort(s->reclist, sp);
840         
841         reclist_enter(s->reclist);
842         *total = reclist_get_num_records(s->reclist);
843         *sumhits = s->total_hits;
844         
845         for (i = 0; i < start; i++)
846             if (!reclist_read_record(s->reclist))
847             {
848                 *num = 0;
849                 recs = 0;
850                 break;
851             }
852         
853         for (i = 0; i < *num; i++)
854         {
855             struct record_cluster *r = reclist_read_record(s->reclist);
856             if (!r)
857             {
858                 *num = i;
859                 break;
860             }
861             recs[i] = r;
862         }
863         reclist_leave(s->reclist);
864     }
865 #if USE_TIMING
866     yaz_timing_stop(t);
867     yaz_log(YLOG_LOG, "show %6.5f %3.2f %3.2f", 
868             yaz_timing_get_real(t), yaz_timing_get_user(t),
869             yaz_timing_get_sys(t));
870     yaz_timing_destroy(&t);
871 #endif
872     return recs;
873 }
874
875 void show_range_stop(struct session *s, struct record_cluster **recs)
876 {
877     session_leave(s);
878 }
879
880 void statistics(struct session *se, struct statistics *stat)
881 {
882     struct client_list *l;
883     int count = 0;
884
885     memset(stat, 0, sizeof(*stat));
886     for (l = se->clients; l; l = l->next)
887     {
888         struct client *cl = l->client;
889         if (!client_get_connection(cl))
890             stat->num_no_connection++;
891         switch (client_get_state(cl))
892         {
893         case Client_Connecting: stat->num_connecting++; break;
894         case Client_Working: stat->num_working++; break;
895         case Client_Idle: stat->num_idle++; break;
896         case Client_Failed: stat->num_failed++; break;
897         case Client_Error: stat->num_error++; break;
898         default: break;
899         }
900         count++;
901     }
902     stat->num_hits = se->total_hits;
903     stat->num_records = se->total_records;
904
905     stat->num_clients = count;
906 }
907
908 static struct record_metadata *record_metadata_init(
909     NMEM nmem, const char *value, enum conf_metadata_type type,
910     struct _xmlAttr *attr)
911 {
912     struct record_metadata *rec_md = record_metadata_create(nmem);
913     struct record_metadata_attr **attrp = &rec_md->attributes;
914     
915     for (; attr; attr = attr->next)
916     {
917         if (attr->children && attr->children->content)
918         {
919             if (strcmp((const char *) attr->name, "type"))
920             {  /* skip the "type" attribute.. Its value is already part of
921                   the element in output (md-%s) and so repeating it here
922                   is redundant */
923                 *attrp = nmem_malloc(nmem, sizeof(**attrp));
924                 (*attrp)->name =
925                     nmem_strdup(nmem, (const char *) attr->name);
926                 (*attrp)->value =
927                     nmem_strdup(nmem, (const char *) attr->children->content);
928                 attrp = &(*attrp)->next;
929             }
930         }
931     }
932     *attrp = 0;
933
934     if (type == Metadata_type_generic)
935     {
936         char *p = nmem_strdup(nmem, value);
937
938         p = normalize7bit_generic(p, " ,/.:([");
939         
940         rec_md->data.text.disp = p;
941         rec_md->data.text.sort = 0;
942     }
943     else if (type == Metadata_type_year || type == Metadata_type_date)
944     {
945         int first, last;
946         int longdate = 0;
947
948         if (type == Metadata_type_date)
949             longdate = 1;
950         if (extract7bit_dates((char *) value, &first, &last, longdate) < 0)
951             return 0;
952
953         rec_md->data.number.min = first;
954         rec_md->data.number.max = last;
955     }
956     else
957         return 0;
958     return rec_md;
959 }
960
961 static int get_mergekey_from_doc(xmlDoc *doc, xmlNode *root, const char *name,
962                                  struct conf_service *service, WRBUF norm_wr)
963 {
964     xmlNode *n;
965     int no_found = 0;
966     for (n = root->children; n; n = n->next)
967     {
968         if (n->type != XML_ELEMENT_NODE)
969             continue;
970         if (!strcmp((const char *) n->name, "metadata"))
971         {
972             xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
973             if (!strcmp(name, (const char *) type))
974             {
975                 xmlChar *value = xmlNodeListGetString(doc, n->children, 1);
976                 if (value)
977                 {
978                     const char *norm_str;
979                     pp2_relevance_token_t prt =
980                         pp2_relevance_tokenize(service->mergekey_pct);
981                     
982                     pp2_relevance_first(prt, (const char *) value, 0);
983                     if (wrbuf_len(norm_wr) > 0)
984                         wrbuf_puts(norm_wr, " ");
985                     wrbuf_puts(norm_wr, name);
986                     while ((norm_str =
987                             pp2_relevance_token_next(prt)))
988                     {
989                         if (*norm_str)
990                         {
991                             wrbuf_puts(norm_wr, " ");
992                             wrbuf_puts(norm_wr, norm_str);
993                         }
994                     }
995                     xmlFree(value);
996                     pp2_relevance_token_destroy(prt);
997                     no_found++;
998                 }
999             }
1000             xmlFree(type);
1001         }
1002     }
1003     return no_found;
1004 }
1005
1006 static const char *get_mergekey(xmlDoc *doc, struct client *cl, int record_no,
1007                                 struct conf_service *service, NMEM nmem)
1008 {
1009     char *mergekey_norm = 0;
1010     xmlNode *root = xmlDocGetRootElement(doc);
1011     WRBUF norm_wr = wrbuf_alloc();
1012
1013     /* consider mergekey from XSL first */
1014     xmlChar *mergekey = xmlGetProp(root, (xmlChar *) "mergekey");
1015     if (mergekey)
1016     {
1017         const char *norm_str;
1018         pp2_relevance_token_t prt =
1019             pp2_relevance_tokenize(service->mergekey_pct);
1020
1021         pp2_relevance_first(prt, (const char *) mergekey, 0);
1022         while ((norm_str = pp2_relevance_token_next(prt)))
1023         {
1024             if (*norm_str)
1025             {
1026                 if (wrbuf_len(norm_wr))
1027                     wrbuf_puts(norm_wr, " ");
1028                 wrbuf_puts(norm_wr, norm_str);
1029             }
1030         }
1031         pp2_relevance_token_destroy(prt);
1032         xmlFree(mergekey);
1033     }
1034     else
1035     {
1036         /* no mergekey defined in XSL. Look for mergekey metadata instead */
1037         int field_id;
1038         for (field_id = 0; field_id < service->num_metadata; field_id++)
1039         {
1040             struct conf_metadata *ser_md = &service->metadata[field_id];
1041             if (ser_md->mergekey != Metadata_mergekey_no)
1042             {
1043                 int r = get_mergekey_from_doc(doc, root, ser_md->name,
1044                                               service, norm_wr);
1045                 if (r == 0 && ser_md->mergekey == Metadata_mergekey_required)
1046                 {
1047                     /* no mergekey on this one and it is required.. 
1048                        Generate unique key instead */
1049                     wrbuf_rewind(norm_wr);
1050                     break;
1051                 }
1052             }
1053         }
1054     }
1055
1056     /* generate unique key if none is not generated already or is empty */
1057     if (wrbuf_len(norm_wr) == 0)
1058     {
1059         wrbuf_printf(norm_wr, "%s-%d",
1060                      client_get_database(cl)->database->url, record_no);
1061     }
1062     if (wrbuf_len(norm_wr) > 0)
1063         mergekey_norm = nmem_strdup(nmem, wrbuf_cstr(norm_wr));
1064     wrbuf_destroy(norm_wr);
1065     return mergekey_norm;
1066 }
1067
1068 /** \brief see if metadata for pz:recordfilter exists 
1069     \param root xml root element of normalized record
1070     \param sdb session database for client
1071     \retval 0 if there is no metadata for pz:recordfilter
1072     \retval 1 if there is metadata for pz:recordfilter
1073
1074     If there is no pz:recordfilter defined, this function returns 1
1075     as well.
1076 */
1077     
1078 static int check_record_filter(xmlNode *root, struct session_database *sdb)
1079 {
1080     int match = 0;
1081     xmlNode *n;
1082     const char *s;
1083     s = session_setting_oneval(sdb, PZ_RECORDFILTER);
1084
1085     if (!s || !*s)
1086         return 1;
1087
1088     for (n = root->children; n; n = n->next)
1089     {
1090         if (n->type != XML_ELEMENT_NODE)
1091             continue;
1092         if (!strcmp((const char *) n->name, "metadata"))
1093         {
1094             xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
1095             if (type)
1096             {
1097                 size_t len;
1098                 const char *eq = strchr(s, '~');
1099                 if (eq)
1100                     len = eq - s;
1101                 else
1102                     len = strlen(s);
1103                 if (len == strlen((const char *)type) &&
1104                     !memcmp((const char *) type, s, len))
1105                 {
1106                     xmlChar *value = xmlNodeGetContent(n);
1107                     if (value && *value)
1108                     {
1109                         if (!eq || strstr((const char *) value, eq+1))
1110                             match = 1;
1111                     }
1112                     xmlFree(value);
1113                 }
1114                 xmlFree(type);
1115             }
1116         }
1117     }
1118     return match;
1119 }
1120
1121
1122 static int ingest_to_cluster(struct client *cl,
1123                              xmlDoc *xdoc,
1124                              xmlNode *root,
1125                              int record_no,
1126                              const char *mergekey_norm);
1127
1128 /** \brief ingest XML record
1129     \param cl client holds the result set for record
1130     \param rec record buffer (0 terminated)
1131     \param record_no record position (1, 2, ..)
1132     \retval 0 OK
1133     \retval -1 failure
1134 */
1135 int ingest_record(struct client *cl, const char *rec,
1136                   int record_no, NMEM nmem)
1137 {
1138     struct session *se = client_get_session(cl);
1139     int ret = 0;
1140     struct session_database *sdb = client_get_database(cl);
1141     struct conf_service *service = se->service;
1142     xmlDoc *xdoc = normalize_record(sdb, service, rec, nmem);
1143     xmlNode *root;
1144     const char *mergekey_norm;
1145     
1146     if (!xdoc)
1147         return -1;
1148     
1149     root = xmlDocGetRootElement(xdoc);
1150     
1151     if (!check_record_filter(root, sdb))
1152     {
1153         yaz_log(YLOG_WARN, "Filtered out record no %d from %s", record_no,
1154                 sdb->database->url);
1155         xmlFreeDoc(xdoc);
1156         return -1;
1157     }
1158     
1159     mergekey_norm = get_mergekey(xdoc, cl, record_no, service, nmem);
1160     if (!mergekey_norm)
1161     {
1162         yaz_log(YLOG_WARN, "Got no mergekey");
1163         xmlFreeDoc(xdoc);
1164         return -1;
1165     }
1166     client_unlock(cl);
1167     pazpar2_sleep(0.01);
1168     session_enter(se);
1169     client_lock(cl);
1170     if (client_get_session(cl) == se)
1171         ret = ingest_to_cluster(cl, xdoc, root, record_no, mergekey_norm);
1172     client_unlock(cl);
1173     session_leave(se);
1174     client_lock(cl);
1175     
1176     xmlFreeDoc(xdoc);
1177     return ret;
1178 }
1179
1180 static int ingest_to_cluster(struct client *cl,
1181                              xmlDoc *xdoc,
1182                              xmlNode *root,
1183                              int record_no,
1184                              const char *mergekey_norm)
1185 {
1186     xmlNode *n;
1187     xmlChar *type = 0;
1188     xmlChar *value = 0;
1189     struct session_database *sdb = client_get_database(cl);
1190     struct session *se = client_get_session(cl);
1191     struct conf_service *service = se->service;
1192     struct record *record = record_create(se->nmem, 
1193                                           service->num_metadata,
1194                                           service->num_sortkeys, cl,
1195                                           record_no);
1196     struct record_cluster *cluster = reclist_insert(se->reclist,
1197                                                     service, 
1198                                                     record,
1199                                                     mergekey_norm,
1200                                                     &se->total_merged);
1201     if (!cluster)
1202         return -1;
1203     if (global_parameters.dump_records)
1204         yaz_log(YLOG_LOG, "Cluster id %s from %s (#%d)", cluster->recid,
1205                 sdb->database->url, record_no);
1206     relevance_newrec(se->relevance, cluster);
1207     
1208     // now parsing XML record and adding data to cluster or record metadata
1209     for (n = root->children; n; n = n->next)
1210     {
1211         pp2_relevance_token_t prt;
1212         if (type)
1213             xmlFree(type);
1214         if (value)
1215             xmlFree(value);
1216         type = value = 0;
1217         
1218         if (n->type != XML_ELEMENT_NODE)
1219             continue;
1220         if (!strcmp((const char *) n->name, "metadata"))
1221         {
1222             struct conf_metadata *ser_md = 0;
1223             struct conf_sortkey *ser_sk = 0;
1224             struct record_metadata **wheretoput = 0;
1225             struct record_metadata *rec_md = 0;
1226             int md_field_id = -1;
1227             int sk_field_id = -1;
1228             
1229             type = xmlGetProp(n, (xmlChar *) "type");
1230             value = xmlNodeListGetString(xdoc, n->children, 1);
1231             
1232             if (!type || !value || !*value)
1233                 continue;
1234             
1235             md_field_id 
1236                 = conf_service_metadata_field_id(service, (const char *) type);
1237             if (md_field_id < 0)
1238             {
1239                 if (se->number_of_warnings_unknown_metadata == 0)
1240                 {
1241                     yaz_log(YLOG_WARN, 
1242                             "Ignoring unknown metadata element: %s", type);
1243                 }
1244                 se->number_of_warnings_unknown_metadata++;
1245                 continue;
1246             }
1247             
1248             ser_md = &service->metadata[md_field_id];
1249             
1250             if (ser_md->sortkey_offset >= 0){
1251                 sk_field_id = ser_md->sortkey_offset;
1252                 ser_sk = &service->sortkeys[sk_field_id];
1253             }
1254
1255             // non-merged metadata
1256             rec_md = record_metadata_init(se->nmem, (const char *) value,
1257                                           ser_md->type, n->properties);
1258             if (!rec_md)
1259             {
1260                 yaz_log(YLOG_WARN, "bad metadata data '%s' for element '%s'",
1261                         value, type);
1262                 continue;
1263             }
1264             wheretoput = &record->metadata[md_field_id];
1265             while (*wheretoput)
1266                 wheretoput = &(*wheretoput)->next;
1267             *wheretoput = rec_md;
1268
1269             // merged metadata
1270             rec_md = record_metadata_init(se->nmem, (const char *) value,
1271                                           ser_md->type, 0);
1272             wheretoput = &cluster->metadata[md_field_id];
1273
1274             // and polulate with data:
1275             // assign cluster or record based on merge action
1276             if (ser_md->merge == Metadata_merge_unique)
1277             {
1278                 struct record_metadata *mnode;
1279                 for (mnode = *wheretoput; mnode; mnode = mnode->next)
1280                     if (!strcmp((const char *) mnode->data.text.disp, 
1281                                 rec_md->data.text.disp))
1282                         break;
1283                 if (!mnode)
1284                 {
1285                     rec_md->next = *wheretoput;
1286                     *wheretoput = rec_md;
1287                 }
1288             }
1289             else if (ser_md->merge == Metadata_merge_longest)
1290             {
1291                 if (!*wheretoput 
1292                     || strlen(rec_md->data.text.disp) 
1293                     > strlen((*wheretoput)->data.text.disp))
1294                 {
1295                     *wheretoput = rec_md;
1296                     if (ser_sk)
1297                     {
1298                         const char *sort_str = 0;
1299                         int skip_article = 
1300                             ser_sk->type == Metadata_sortkey_skiparticle;
1301
1302                         if (!cluster->sortkeys[sk_field_id])
1303                             cluster->sortkeys[sk_field_id] = 
1304                                 nmem_malloc(se->nmem, 
1305                                             sizeof(union data_types));
1306                          
1307                         prt = pp2_relevance_tokenize(service->sort_pct);
1308
1309                         pp2_relevance_first(prt, rec_md->data.text.disp,
1310                                             skip_article);
1311
1312                         pp2_relevance_token_next(prt);
1313                          
1314                         sort_str = pp2_get_sort(prt);
1315                          
1316                         cluster->sortkeys[sk_field_id]->text.disp = 
1317                             rec_md->data.text.disp;
1318                         if (!sort_str)
1319                         {
1320                             sort_str = rec_md->data.text.disp;
1321                             yaz_log(YLOG_WARN, 
1322                                     "Could not make sortkey. Bug #1858");
1323                         }
1324                         cluster->sortkeys[sk_field_id]->text.sort = 
1325                             nmem_strdup(se->nmem, sort_str);
1326 #if 0
1327                         yaz_log(YLOG_LOG, "text disp=%s",
1328                                 cluster->sortkeys[sk_field_id]->text.disp);
1329                         yaz_log(YLOG_LOG, "text sort=%s",
1330                                 cluster->sortkeys[sk_field_id]->text.sort);
1331 #endif
1332                         pp2_relevance_token_destroy(prt);
1333                     }
1334                 }
1335             }
1336             else if (ser_md->merge == Metadata_merge_all)
1337             {
1338                 rec_md->next = *wheretoput;
1339                 *wheretoput = rec_md;
1340             }
1341             else if (ser_md->merge == Metadata_merge_range)
1342             {
1343                 if (!*wheretoput)
1344                 {
1345                     *wheretoput = rec_md;
1346                     if (ser_sk)
1347                         cluster->sortkeys[sk_field_id] 
1348                             = &rec_md->data;
1349                 }
1350                 else
1351                 {
1352                     int this_min = rec_md->data.number.min;
1353                     int this_max = rec_md->data.number.max;
1354                     if (this_min < (*wheretoput)->data.number.min)
1355                         (*wheretoput)->data.number.min = this_min;
1356                     if (this_max > (*wheretoput)->data.number.max)
1357                         (*wheretoput)->data.number.max = this_max;
1358                 }
1359             }
1360
1361
1362             // ranking of _all_ fields enabled ... 
1363             if (ser_md->rank)
1364                 relevance_countwords(se->relevance, cluster, 
1365                                      (char *) value, ser_md->rank,
1366                                      ser_md->name);
1367
1368             // construct facets ... 
1369             if (ser_md->termlist)
1370             {
1371                 if (ser_md->type == Metadata_type_year)
1372                 {
1373                     char year[64];
1374                     sprintf(year, "%d", rec_md->data.number.max);
1375                     add_facet(se, (char *) type, year);
1376                     if (rec_md->data.number.max != rec_md->data.number.min)
1377                     {
1378                         sprintf(year, "%d", rec_md->data.number.min);
1379                         add_facet(se, (char *) type, year);
1380                     }
1381                 }
1382                 else
1383                     add_facet(se, (char *) type, (char *) value);
1384             }
1385
1386             // cleaning up
1387             xmlFree(type);
1388             xmlFree(value);
1389             type = value = 0;
1390         }
1391         else
1392         {
1393             if (se->number_of_warnings_unknown_elements == 0)
1394                 yaz_log(YLOG_WARN,
1395                         "Unexpected element in internal record: %s", n->name);
1396             se->number_of_warnings_unknown_elements++;
1397         }
1398     }
1399     if (type)
1400         xmlFree(type);
1401     if (value)
1402         xmlFree(value);
1403
1404     relevance_donerecord(se->relevance, cluster);
1405     se->total_records++;
1406
1407     return 0;
1408 }
1409
1410 /*
1411  * Local variables:
1412  * c-basic-offset: 4
1413  * c-file-style: "Stroustrup"
1414  * indent-tabs-mode: nil
1415  * End:
1416  * vim: shiftwidth=4 tabstop=8 expandtab
1417  */
1418