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