src/session.c

   1 /* This file is part of Pazpar2.
   2    Copyright (C) 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
  60 #define USE_TIMING 0
  61 #if USE_TIMING
  62 #include <yaz/timing.h>
  63 #endif
  64
  65 #include "ppmutex.h"
  66 #include "parameters.h"
  67 #include "session.h"
  68 #include "eventl.h"
  69 #include "http.h"
  70 #include "termlists.h"
  71 #include "reclists.h"
  72 #include "relevance.h"
  73 #include "database.h"
  74 #include "client.h"
  75 #include "settings.h"
  76 #include "normalize7bit.h"
  77
  78 #include <libxml/tree.h>
  79
  80 #define MAX_CHUNK 15
  81
  82 #define MAX(a,b) ((a)>(b)?(a):(b))
  83
  84 // Note: Some things in this structure will eventually move to configuration
  85 struct parameters global_parameters =
  86 {
  87     0,   // dump_records
  88     0,   // debug_mode
  89     0,   // predictable sessions
  90 };
  91
  92 struct client_list {
  93     struct client *client;
  94     struct client_list *next;
  95 };
  96
  97 /* session counting (1) , disable client counting (0) */
  98 static YAZ_MUTEX g_session_mutex = 0;
  99 static int no_sessions = 0;
 100
 101 static int session_use(int delta)
 102 {
 103     int sessions;
 104     if (!g_session_mutex)
 105         yaz_mutex_create(&g_session_mutex);
 106     yaz_mutex_enter(g_session_mutex);
 107     no_sessions += delta;
 108     sessions = no_sessions;
 109     yaz_mutex_leave(g_session_mutex);
 110     yaz_log(YLOG_DEBUG, "%s sessions=%d", delta == 0 ? "" :
 111             (delta > 0 ? "INC" : "DEC"), no_sessions);
 112     return sessions;
 113 }
 114
 115 int sessions_count(void)
 116 {
 117     return session_use(0);
 118 }
 119
 120 static void log_xml_doc(xmlDoc *doc)
 121 {
 122     FILE *lf = yaz_log_file();
 123     xmlChar *result = 0;
 124     int len = 0;
 125 #if LIBXML_VERSION >= 20600
 126     xmlDocDumpFormatMemory(doc, &result, &len, 1);
 127 #else
 128     xmlDocDumpMemory(doc, &result, &len);
 129 #endif
 130     if (lf && len)
 131     {
 132         (void) fwrite(result, 1, len, lf);
 133         fprintf(lf, "\n");
 134     }
 135     xmlFree(result);
 136 }
 137
 138 static void session_enter(struct session *s, const char *caller)
 139 {
 140     if (caller)
 141         session_log(s, YLOG_DEBUG, "Session lock by %s", caller);
 142     yaz_mutex_enter(s->session_mutex);
 143 }
 144
 145 static void session_leave(struct session *s, const char *caller)
 146 {
 147     yaz_mutex_leave(s->session_mutex);
 148     if (caller)
 149         session_log(s, YLOG_DEBUG, "Session unlock by %s", caller);
 150 }
 151
 152 static int run_icu(struct session *s, const char *icu_chain_id,
 153                    const char *value,
 154                    WRBUF norm_wr, WRBUF disp_wr)
 155 {
 156     const char *facet_component;
 157     struct conf_service *service = s->service;
 158     pp2_charset_token_t prt =
 159         pp2_charset_token_create(service->charsets, icu_chain_id);
 160     if (!prt)
 161     {
 162         session_log(s, YLOG_FATAL,
 163                     "Unknown ICU chain '%s'", icu_chain_id);
 164         return 0;
 165     }
 166     pp2_charset_token_first(prt, value, 0);
 167     while ((facet_component = pp2_charset_token_next(prt)))
 168     {
 169         const char *display_component;
 170         if (*facet_component)
 171         {
 172             if (wrbuf_len(norm_wr))
 173                 wrbuf_puts(norm_wr, " ");
 174             wrbuf_puts(norm_wr, facet_component);
 175         }
 176         display_component = pp2_get_display(prt);
 177         if (display_component)
 178         {
 179             if (wrbuf_len(disp_wr))
 180                 wrbuf_puts(disp_wr, " ");
 181             wrbuf_puts(disp_wr, display_component);
 182         }
 183     }
 184     pp2_charset_token_destroy(prt);
 185     return 1;
 186 }
 187
 188 static void session_normalize_facet(struct session *s,
 189                                     const char *type, const char *value,
 190                                     WRBUF display_wrbuf, WRBUF facet_wrbuf)
 191 {
 192     struct conf_service *service = s->service;
 193     int i;
 194     const char *icu_chain_id = 0;
 195
 196     for (i = 0; i < service->num_metadata; i++)
 197         if (!strcmp((service->metadata + i)->name, type))
 198             icu_chain_id = (service->metadata + i)->facetrule;
 199     if (!icu_chain_id)
 200         icu_chain_id = "facet";
 201
 202     run_icu(s, icu_chain_id, value, facet_wrbuf, display_wrbuf);
 203 }
 204
 205 void add_facet(struct session *s, const char *type, const char *value, int count)
 206 {
 207     WRBUF facet_wrbuf = wrbuf_alloc();
 208     WRBUF display_wrbuf = wrbuf_alloc();
 209
 210     session_normalize_facet(s, type, value, display_wrbuf, facet_wrbuf);
 211
 212     if (wrbuf_len(facet_wrbuf))
 213     {
 214         struct named_termlist **tp = &s->termlists;
 215         for (; (*tp); tp = &(*tp)->next)
 216             if (!strcmp((*tp)->name, type))
 217                 break;
 218         if (!*tp)
 219         {
 220             *tp = nmem_malloc(s->nmem, sizeof(**tp));
 221             (*tp)->name = nmem_strdup(s->nmem, type);
 222             (*tp)->termlist = termlist_create(s->nmem);
 223             (*tp)->next = 0;
 224         }
 225         termlist_insert((*tp)->termlist, wrbuf_cstr(display_wrbuf),
 226                         wrbuf_cstr(facet_wrbuf), count);
 227     }
 228     wrbuf_destroy(facet_wrbuf);
 229     wrbuf_destroy(display_wrbuf);
 230 }
 231
 232 static xmlDoc *record_to_xml(struct session *se,
 233                              struct session_database *sdb, const char *rec)
 234 {
 235     struct database *db = sdb->database;
 236     xmlDoc *rdoc = 0;
 237
 238     rdoc = xmlParseMemory(rec, strlen(rec));
 239
 240     if (!rdoc)
 241     {
 242         session_log(se, YLOG_WARN, "Non-wellformed XML");
 243         return 0;
 244     }
 245
 246     if (global_parameters.dump_records)
 247     {
 248         session_log(se, YLOG_LOG, "Un-normalized record from %s", db->id);
 249         log_xml_doc(rdoc);
 250     }
 251
 252     return rdoc;
 253 }
 254
 255 #define MAX_XSLT_ARGS 16
 256
 257 // Add static values from session database settings if applicable
 258 static void insert_settings_parameters(struct session_database *sdb,
 259                                        struct conf_service *service,
 260                                        char **parms,
 261                                        NMEM nmem)
 262 {
 263     int i;
 264     int nparms = 0;
 265     int offset = 0;
 266
 267     for (i = 0; i < service->num_metadata; i++)
 268     {
 269         struct conf_metadata *md = &service->metadata[i];
 270         int setting;
 271
 272         if (md->setting == Metadata_setting_parameter &&
 273             (setting = settings_lookup_offset(service, md->name)) >= 0)
 274         {
 275             const char *val = session_setting_oneval(sdb, setting);
 276             if (val && nparms < MAX_XSLT_ARGS)
 277             {
 278                 char *buf;
 279                 int len = strlen(val);
 280                 buf = nmem_malloc(nmem, len + 3);
 281                 buf[0] = '\'';
 282                 strcpy(buf + 1, val);
 283                 buf[len+1] = '\'';
 284                 buf[len+2] = '\0';
 285                 parms[offset++] = md->name;
 286                 parms[offset++] = buf;
 287                 nparms++;
 288             }
 289         }
 290     }
 291     parms[offset] = 0;
 292 }
 293
 294 // Add static values from session database settings if applicable
 295 static void insert_settings_values(struct session_database *sdb, xmlDoc *doc,
 296                                    xmlNode *root,
 297                                    struct conf_service *service)
 298 {
 299     int i;
 300
 301     for (i = 0; i < service->num_metadata; i++)
 302     {
 303         struct conf_metadata *md = &service->metadata[i];
 304         int offset;
 305
 306         if (md->setting == Metadata_setting_postproc &&
 307             (offset = settings_lookup_offset(service, md->name)) >= 0)
 308         {
 309             const char *val = session_setting_oneval(sdb, offset);
 310             if (val)
 311             {
 312                 xmlNode *n = xmlNewTextChild(root, 0, (xmlChar *) "metadata",
 313                                              (xmlChar *) val);
 314                 xmlSetProp(n, (xmlChar *) "type", (xmlChar *) md->name);
 315             }
 316         }
 317     }
 318 }
 319
 320 static xmlDoc *normalize_record(struct session *se,
 321                                 struct session_database *sdb,
 322                                 struct conf_service *service,
 323                                 const char *rec, NMEM nmem)
 324 {
 325     xmlDoc *rdoc = record_to_xml(se, sdb, rec);
 326
 327     if (rdoc)
 328     {
 329         char *parms[MAX_XSLT_ARGS*2+1];
 330
 331         insert_settings_parameters(sdb, service, parms, nmem);
 332
 333         if (normalize_record_transform(sdb->map, &rdoc, (const char **)parms))
 334         {
 335             session_log(se, YLOG_WARN, "Normalize failed");
 336         }
 337     }
 338     return rdoc;
 339 }
 340
 341 void session_settings_dump(struct session *se,
 342                            struct session_database *db,
 343                            WRBUF w)
 344 {
 345     if (db->settings)
 346     {
 347         int i, num = db->num_settings;
 348         for (i = 0; i < num; i++)
 349         {
 350             struct setting *s = db->settings[i];
 351             for (;s ; s = s->next)
 352             {
 353                 wrbuf_puts(w, "<set name=\"");
 354                 wrbuf_xmlputs(w, s->name);
 355                 wrbuf_puts(w, "\" value=\"");
 356                 wrbuf_xmlputs(w, s->value);
 357                 wrbuf_puts(w, "\"/>");
 358             }
 359             if (db->settings[i])
 360                 wrbuf_puts(w, "\n");
 361         }
 362     }
 363 }
 364
 365 // Retrieve first defined value for 'name' for given database.
 366 // Will be extended to take into account user associated with session
 367 const char *session_setting_oneval(struct session_database *db, int offset)
 368 {
 369     if (offset >= db->num_settings || !db->settings[offset])
 370         return "";
 371     return db->settings[offset]->value;
 372 }
 373
 374 // Prepare XSLT stylesheets for record normalization
 375 // Structures are allocated on the session_wide nmem to avoid having
 376 // to recompute this for every search. This would lead
 377 // to leaking if a single session was to repeatedly change the PZ_XSLT
 378 // setting. However, this is not a realistic use scenario.
 379 static int prepare_map(struct session *se, struct session_database *sdb)
 380 {
 381     if (sdb->settings && !sdb->map)
 382     {
 383         const char *s;
 384
 385         if (sdb->settings[PZ_XSLT] &&
 386             (s = session_setting_oneval(sdb, PZ_XSLT)))
 387         {
 388             char auto_stylesheet[256];
 389
 390             if (!strcmp(s, "auto"))
 391             {
 392                 const char *request_syntax = session_setting_oneval(
 393                     sdb, PZ_REQUESTSYNTAX);
 394                 if (request_syntax)
 395                 {
 396                     char *cp;
 397                     yaz_snprintf(auto_stylesheet, sizeof(auto_stylesheet),
 398                                  "%s.xsl", request_syntax);
 399                     for (cp = auto_stylesheet; *cp; cp++)
 400                     {
 401                         /* deliberately only consider ASCII */
 402                         if (*cp > 32 && *cp < 127)
 403                             *cp = tolower(*cp);
 404                     }
 405                     s = auto_stylesheet;
 406                 }
 407                 else
 408                 {
 409                     session_log(se, YLOG_WARN,
 410                                 "No pz:requestsyntax for auto stylesheet");
 411                 }
 412             }
 413             sdb->map = normalize_cache_get(se->normalize_cache,
 414                                            se->service, s);
 415             if (!sdb->map)
 416                 return -1;
 417         }
 418     }
 419     return 0;
 420 }
 421
 422 // called if watch should be removed because http_channel is to be destroyed
 423 static void session_watch_cancel(void *data, struct http_channel *c,
 424                                  void *data2)
 425 {
 426     struct session_watchentry *ent = data;
 427
 428     ent->fun = 0;
 429     ent->data = 0;
 430     ent->obs = 0;
 431 }
 432
 433 // set watch. Returns 0=OK, -1 if watch is already set
 434 int session_set_watch(struct session *s, int what,
 435                       session_watchfun fun, void *data,
 436                       struct http_channel *chan)
 437 {
 438     int ret;
 439     session_enter(s, "session_set_watch");
 440     if (s->watchlist[what].fun)
 441         ret = -1;
 442     else
 443     {
 444
 445         s->watchlist[what].fun = fun;
 446         s->watchlist[what].data = data;
 447         s->watchlist[what].obs = http_add_observer(chan, &s->watchlist[what],
 448                                                    session_watch_cancel);
 449         ret = 0;
 450     }
 451     session_leave(s, "session_set_watch");
 452     return ret;
 453 }
 454
 455 void session_alert_watch(struct session *s, int what)
 456 {
 457     assert(s);
 458     session_enter(s, "session_alert_watch");
 459     if (s->watchlist[what].fun)
 460     {
 461         /* our watch is no longer associated with http_channel */
 462         void *data;
 463         session_watchfun fun;
 464
 465         http_remove_observer(s->watchlist[what].obs);
 466         fun  = s->watchlist[what].fun;
 467         data = s->watchlist[what].data;
 468
 469         /* reset watch before fun is invoked - in case fun wants to set
 470            it again */
 471         s->watchlist[what].fun = 0;
 472         s->watchlist[what].data = 0;
 473         s->watchlist[what].obs = 0;
 474
 475         session_leave(s, "session_alert_watch");
 476         session_log(s, YLOG_DEBUG,
 477                     "Alert Watch: %d calling function: %p", what, fun);
 478         fun(data);
 479     }
 480     else
 481         session_leave(s,"session_alert_watch");
 482 }
 483
 484 //callback for grep_databases
 485 static void select_targets_callback(struct session *se,
 486                                     struct session_database *db)
 487 {
 488     struct client *cl;
 489     struct client_list *l;
 490
 491     for (l = se->clients_cached; l; l = l->next)
 492         if (client_get_database(l->client) == db)
 493             break;
 494
 495     if (l)
 496         cl = l->client;
 497     else
 498     {
 499         cl = client_create(db->database->id);
 500         client_set_database(cl, db);
 501
 502         l = xmalloc(sizeof(*l));
 503         l->client = cl;
 504         l->next = se->clients_cached;
 505         se->clients_cached = l;
 506     }
 507     /* set session always. If may be 0 if client is not active */
 508     client_set_session(cl, se);
 509
 510     l = xmalloc(sizeof(*l));
 511     l->client = cl;
 512     l->next = se->clients_active;
 513     se->clients_active = l;
 514 }
 515
 516 static void session_reset_active_clients(struct session *se,
 517                                          struct client_list *new_list)
 518 {
 519     struct client_list *l;
 520
 521     session_enter(se, "session_reset_active_clients");
 522     l = se->clients_active;
 523     se->clients_active = new_list;
 524     session_leave(se, "session_reset_active_clients");
 525
 526     while (l)
 527     {
 528         struct client_list *l_next = l->next;
 529
 530         client_lock(l->client);
 531         client_set_session(l->client, 0); /* mark client inactive */
 532         client_unlock(l->client);
 533
 534         xfree(l);
 535         l = l_next;
 536     }
 537 }
 538
 539 static void session_remove_cached_clients(struct session *se)
 540 {
 541     struct client_list *l;
 542
 543     session_reset_active_clients(se, 0);
 544
 545     session_enter(se, "session_remove_cached_clients");
 546     l = se->clients_cached;
 547     se->clients_cached = 0;
 548     session_leave(se, "session_remove_cached_clients");
 549
 550     while (l)
 551     {
 552         struct client_list *l_next = l->next;
 553         client_lock(l->client);
 554         client_set_session(l->client, 0);
 555         client_set_database(l->client, 0);
 556         client_unlock(l->client);
 557         client_destroy(l->client);
 558         xfree(l);
 559         l = l_next;
 560     }
 561 }
 562
 563 // Associates a set of clients with a session;
 564 // Note: Session-databases represent databases with per-session
 565 // setting overrides
 566 static int select_targets(struct session *se, const char *filter)
 567 {
 568     return session_grep_databases(se, filter, select_targets_callback);
 569 }
 570
 571 int session_active_clients(struct session *s)
 572 {
 573     struct client_list *l;
 574     int res = 0;
 575
 576     for (l = s->clients_active; l; l = l->next)
 577         if (client_is_active(l->client))
 578             res++;
 579
 580     return res;
 581 }
 582
 583 int session_is_preferred_clients_ready(struct session *s)
 584 {
 585     struct client_list *l;
 586     int res = 0;
 587
 588     for (l = s->clients_active; l; l = l->next)
 589         if (client_is_active_preferred(l->client))
 590             res++;
 591     session_log(s, YLOG_DEBUG, "Has %d active preferred clients.", res);
 592     return res == 0;
 593 }
 594
 595 static void session_clear_set(struct session *se, struct reclist_sortparms *sp)
 596 {
 597     reclist_destroy(se->reclist);
 598     if (nmem_total(se->nmem))
 599         session_log(se, YLOG_DEBUG, "NMEN operation usage %zd",
 600                     nmem_total(se->nmem));
 601     nmem_reset(se->nmem);
 602     se->total_records = se->total_merged = 0;
 603     se->termlists = 0;
 604     relevance_clear(se->relevance);
 605
 606     /* reset list of sorted results and clear to relevance search */
 607     se->sorted_results = nmem_malloc(se->nmem, sizeof(*se->sorted_results));
 608     se->sorted_results->name = nmem_strdup(se->nmem, sp->name);
 609     se->sorted_results->increasing = sp->increasing;
 610     se->sorted_results->type = sp->type;
 611     se->sorted_results->next = 0;
 612
 613     session_log(se, YLOG_DEBUG, "clear_set session_sort: field=%s increasing=%d type=%d configured",
 614             sp->name, sp->increasing, sp->type);
 615
 616     se->reclist = reclist_create(se->nmem);
 617 }
 618
 619 void session_sort(struct session *se, struct reclist_sortparms *sp,
 620                   const char *mergekey, const char *rank)
 621 {
 622     struct client_list *l;
 623     const char *field = sp->name;
 624     int increasing = sp->increasing;
 625     int type  = sp->type;
 626     int clients_research = 0;
 627
 628     session_enter(se, "session_sort");
 629     session_log(se, YLOG_DEBUG, "session_sort field=%s increasing=%d type=%d",
 630                 field, increasing, type);
 631
 632     if (rank && (!se->rank || strcmp(se->rank, rank)))
 633     {
 634         /* new rank must research/reingest anyway */
 635         assert(rank);
 636         xfree(se->rank);
 637         se->rank = *rank ? xstrdup(rank) : 0;
 638         clients_research = 1;
 639         session_log(se, YLOG_DEBUG, "session_sort: new rank = %s",
 640                     rank);
 641     }
 642     if (mergekey && (!se->mergekey || strcmp(se->mergekey, mergekey)))
 643     {
 644         /* new mergekey must research/reingest anyway */
 645         assert(mergekey);
 646         xfree(se->mergekey);
 647         se->mergekey = *mergekey ? xstrdup(mergekey) : 0;
 648         clients_research = 1;
 649         session_log(se, YLOG_DEBUG, "session_sort: new mergekey = %s",
 650                     mergekey);
 651     }
 652     if (clients_research == 0)
 653     {
 654         struct reclist_sortparms *sr;
 655         for (sr = se->sorted_results; sr; sr = sr->next)
 656             if (!reclist_sortparms_cmp(sr, sp))
 657                 break;
 658         if (sr)
 659         {
 660             session_log(se, YLOG_DEBUG, "session_sort: field=%s increasing=%d type=%d already fetched",
 661                         field, increasing, type);
 662             session_leave(se, "session_sort");
 663             return;
 664         }
 665     }
 666     session_log(se, YLOG_DEBUG, "session_sort: field=%s increasing=%d type=%d must fetch",
 667                 field, increasing, type);
 668
 669     // We need to reset reclist on every sort that changes the records, not just for position
 670     // So if just one client requires new searching, we need to clear set.
 671     // Ask each of the client if sorting requires re-search due to native sort
 672     // If it does it will require us to
 673     for (l = se->clients_active; l; l = l->next)
 674     {
 675         struct client *cl = l->client;
 676         // Assume no re-search is required.
 677         client_parse_init(cl, 1);
 678         clients_research += client_parse_sort(cl, sp);
 679     }
 680     if (!clients_research || se->clients_starting)
 681     {
 682         // A new sorting based on same record set
 683         struct reclist_sortparms *sr = nmem_malloc(se->nmem, sizeof(*sr));
 684         sr->name = nmem_strdup(se->nmem, field);
 685         sr->increasing = increasing;
 686         sr->type = type;
 687         sr->next = se->sorted_results;
 688         se->sorted_results = sr;
 689         session_log(se, YLOG_DEBUG, "session_sort: no research/ingesting done");
 690         session_leave(se, "session_sort");
 691     }
 692     else
 693     {
 694         se->clients_starting = 1;
 695         session_log(se, YLOG_DEBUG,
 696                     "session_sort: reset results due to %d clients researching",
 697                     clients_research);
 698         session_clear_set(se, sp);
 699         session_log(se, YLOG_DEBUG, "Re- search/ingesting for clients due to change in sort order");
 700
 701         session_leave(se, "session_sort");
 702         for (l = se->clients_active; l; l = l->next)
 703         {
 704             struct client *cl = l->client;
 705             if (client_get_state(cl) == Client_Connecting ||
 706                 client_get_state(cl) == Client_Idle ||
 707                 client_get_state(cl) == Client_Working) {
 708                 client_start_search(cl);
 709             }
 710             else
 711             {
 712                 session_log(se, YLOG_DEBUG,
 713                             "session_sort: %s: No re-start/ingest in show. "
 714                             "Wrong client state: %d",
 715                             client_get_id(cl), client_get_state(cl));
 716             }
 717         }
 718         session_enter(se, "session_sort");
 719         se->clients_starting = 0;
 720         session_leave(se, "session_sort");
 721     }
 722 }
 723
 724 void session_stop(struct session *se)
 725 {
 726     struct client_list *l;
 727     session_enter(se, "session_stop1");
 728     if (se->clients_starting)
 729     {
 730         session_leave(se, "session_stop1");
 731         return;
 732     }
 733     se->clients_starting = 1;
 734     session_leave(se, "session_stop1");
 735
 736     session_alert_watch(se, SESSION_WATCH_SHOW);
 737     session_alert_watch(se, SESSION_WATCH_BYTARGET);
 738     session_alert_watch(se, SESSION_WATCH_TERMLIST);
 739     session_alert_watch(se, SESSION_WATCH_SHOW_PREF);
 740
 741     for (l = se->clients_active; l; l = l->next)
 742     {
 743         struct client *cl = l->client;
 744         client_stop(cl);
 745     }
 746     session_enter(se, "session_stop2");
 747     se->clients_starting = 0;
 748     session_leave(se, "session_stop2");
 749 }
 750
 751 enum pazpar2_error_code session_search(struct session *se,
 752                                        const char *query,
 753                                        const char *startrecs,
 754                                        const char *maxrecs,
 755                                        const char *filter,
 756                                        const char *limit,
 757                                        const char **addinfo,
 758                                        const char **addinfo2,
 759                                        struct reclist_sortparms *sp,
 760                                        const char *mergekey,
 761                                        const char *rank)
 762 {
 763     int live_channels = 0;
 764     int no_working = 0;
 765     int no_failed_query = 0;
 766     int no_failed_limit = 0;
 767     struct client_list *l;
 768
 769     session_log(se, YLOG_DEBUG, "Search");
 770
 771     *addinfo = 0;
 772
 773     session_enter(se, "session_search0");
 774     if (se->clients_starting)
 775     {
 776         session_leave(se, "session_search0");
 777         return PAZPAR2_NO_ERROR;
 778     }
 779     se->clients_starting = 1;
 780     session_leave(se, "session_search0");
 781
 782     if (se->settings_modified) {
 783         session_remove_cached_clients(se);
 784     }
 785     else
 786         session_reset_active_clients(se, 0);
 787
 788     session_enter(se, "session_search");
 789     se->settings_modified = 0;
 790
 791     if (mergekey)
 792     {
 793         xfree(se->mergekey);
 794         se->mergekey = *mergekey ? xstrdup(mergekey) : 0;
 795     }
 796     if (rank)
 797     {
 798         xfree(se->rank);
 799         se->rank = *rank ? xstrdup(rank) : 0;
 800     }
 801
 802     session_clear_set(se, sp);
 803     relevance_destroy(&se->relevance);
 804
 805     live_channels = select_targets(se, filter);
 806     if (!live_channels)
 807     {
 808         session_leave(se, "session_search");
 809         se->clients_starting = 0;
 810         return PAZPAR2_NO_TARGETS;
 811     }
 812
 813     facet_limits_destroy(se->facet_limits);
 814     se->facet_limits = facet_limits_create(limit);
 815     if (!se->facet_limits)
 816     {
 817         *addinfo = "limit";
 818         session_leave(se, "session_search");
 819         se->clients_starting = 0;
 820         return PAZPAR2_MALFORMED_PARAMETER_VALUE;
 821     }
 822
 823     session_leave(se, "session_search");
 824
 825     session_alert_watch(se, SESSION_WATCH_SHOW);
 826     session_alert_watch(se, SESSION_WATCH_BYTARGET);
 827     session_alert_watch(se, SESSION_WATCH_TERMLIST);
 828     session_alert_watch(se, SESSION_WATCH_SHOW_PREF);
 829
 830     for (l = se->clients_active; l; l = l->next)
 831     {
 832         int parse_ret;
 833         struct client *cl = l->client;
 834         client_parse_init(cl, 1);
 835         if (prepare_map(se, client_get_database(cl)) < 0)
 836             continue;
 837
 838         parse_ret = client_parse_query(cl, query, se->facet_limits, addinfo2);
 839         if (parse_ret == -1)
 840             no_failed_query++;
 841         else if (parse_ret == -2)
 842             no_failed_limit++;
 843         else if (parse_ret < 0)
 844             no_working++; /* other error, such as bad CCL map */
 845         else
 846         {
 847             client_parse_range(cl, startrecs, maxrecs);
 848             client_parse_sort(cl, sp);
 849             client_start_search(cl);
 850             no_working++;
 851         }
 852     }
 853     session_enter(se, "session_search2");
 854     se->clients_starting = 0;
 855     session_leave(se, "session_search2");
 856     if (no_working == 0)
 857     {
 858         if (no_failed_query > 0)
 859         {
 860             *addinfo = "query";
 861             return PAZPAR2_MALFORMED_PARAMETER_VALUE;
 862         }
 863         else if (no_failed_limit > 0)
 864         {
 865             *addinfo = "limit";
 866             return PAZPAR2_MALFORMED_PARAMETER_VALUE;
 867         }
 868         else
 869             return PAZPAR2_NO_TARGETS;
 870     }
 871     return PAZPAR2_NO_ERROR;
 872 }
 873
 874 // Creates a new session_database object for a database
 875 static void session_init_databases_fun(void *context, struct database *db)
 876 {
 877     struct session *se = (struct session *) context;
 878     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
 879     int i;
 880
 881     new->database = db;
 882
 883     new->map = 0;
 884     assert(db->settings);
 885     new->settings = nmem_malloc(se->session_nmem,
 886                                 sizeof(struct settings *) * db->num_settings);
 887     new->num_settings = db->num_settings;
 888     for (i = 0; i < db->num_settings; i++)
 889     {
 890         struct setting *setting = db->settings[i];
 891         new->settings[i] = setting;
 892     }
 893     new->next = se->databases;
 894     se->databases = new;
 895 }
 896
 897 // Doesn't free memory associated with sdb -- nmem takes care of that
 898 static void session_database_destroy(struct session_database *sdb)
 899 {
 900     sdb->map = 0;
 901 }
 902
 903 // Initialize session_database list -- this represents this session's view
 904 // of the database list -- subject to modification by the settings ws command
 905 void session_init_databases(struct session *se)
 906 {
 907     se->databases = 0;
 908     predef_grep_databases(se, se->service, session_init_databases_fun);
 909 }
 910
 911 // Probably session_init_databases_fun should be refactored instead of
 912 // called here.
 913 static struct session_database *load_session_database(struct session *se,
 914                                                       const char *id)
 915 {
 916     struct database *db = new_database_inherit_settings(id, se->session_nmem, se->service->settings);
 917     session_init_databases_fun((void*) se, db);
 918
 919     // New sdb is head of se->databases list
 920     return se->databases;
 921 }
 922
 923 // Find an existing session database. If not found, load it
 924 static struct session_database *find_session_database(struct session *se,
 925                                                       const char *id)
 926 {
 927     struct session_database *sdb;
 928
 929     for (sdb = se->databases; sdb; sdb = sdb->next)
 930         if (!strcmp(sdb->database->id, id))
 931             return sdb;
 932     return load_session_database(se, id);
 933 }
 934
 935 // Apply a session override to a database
 936 void session_apply_setting(struct session *se, const char *dbname,
 937                            const char *name, const char *value)
 938 {
 939     session_enter(se, "session_apply_setting");
 940     {
 941         struct session_database *sdb = find_session_database(se, dbname);
 942         struct conf_service *service = se->service;
 943         struct setting *s;
 944         int offset = settings_create_offset(service, name);
 945
 946         expand_settings_array(&sdb->settings, &sdb->num_settings, offset,
 947                               se->session_nmem);
 948         // Force later recompute of settings-driven data structures
 949         // (happens when a search starts and client connections are prepared)
 950         if (offset == PZ_XSLT)
 951             sdb->map = 0;
 952         se->settings_modified = 1;
 953         for (s = sdb->settings[offset]; s; s = s->next)
 954             if (!strcmp(s->name, name) &&
 955                 dbname && s->target && !strcmp(dbname, s->target))
 956                 break;
 957         if (!s)
 958         {
 959             s = nmem_malloc(se->session_nmem, sizeof(*s));
 960             s->precedence = 0;
 961             s->target = nmem_strdup(se->session_nmem, dbname);
 962             s->name = nmem_strdup(se->session_nmem, name);
 963             s->next = sdb->settings[offset];
 964             sdb->settings[offset] = s;
 965         }
 966         s->value = nmem_strdup(se->session_nmem, value);
 967     }
 968     session_leave(se, "session_apply_setting");
 969 }
 970
 971 void session_destroy(struct session *se)
 972 {
 973     struct session_database *sdb;
 974     session_log(se, YLOG_LOG, "destroy");
 975     session_use(-1);
 976     session_remove_cached_clients(se);
 977
 978     for (sdb = se->databases; sdb; sdb = sdb->next)
 979         session_database_destroy(sdb);
 980     normalize_cache_destroy(se->normalize_cache);
 981     relevance_destroy(&se->relevance);
 982     reclist_destroy(se->reclist);
 983     xfree(se->mergekey);
 984     xfree(se->rank);
 985     if (nmem_total(se->nmem))
 986         session_log(se, YLOG_DEBUG, "NMEN operation usage %zd", nmem_total(se->nmem));
 987     if (nmem_total(se->session_nmem))
 988         session_log(se, YLOG_DEBUG, "NMEN session usage %zd", nmem_total(se->session_nmem));
 989     facet_limits_destroy(se->facet_limits);
 990     nmem_destroy(se->nmem);
 991     service_destroy(se->service);
 992     yaz_mutex_destroy(&se->session_mutex);
 993 }
 994
 995 size_t session_get_memory_status(struct session *session) {
 996     size_t session_nmem;
 997     if (session == 0)
 998         return 0;
 999     session_enter(session, "session_get_memory_status");
1000     session_nmem = nmem_total(session->nmem);
1001     session_leave(session, "session_get_memory_status");
1002     return session_nmem;
1003 }
1004
1005
1006 struct session *new_session(NMEM nmem, struct conf_service *service,
1007                             unsigned session_id)
1008 {
1009     int i;
1010     struct session *session = nmem_malloc(nmem, sizeof(*session));
1011
1012     char tmp_str[50];
1013
1014     sprintf(tmp_str, "session#%u", session_id);
1015
1016     session->session_id = session_id;
1017     session_log(session, YLOG_DEBUG, "New");
1018     session->service = service;
1019     session->relevance = 0;
1020     session->total_records = 0;
1021     session->number_of_warnings_unknown_elements = 0;
1022     session->number_of_warnings_unknown_metadata = 0;
1023     session->termlists = 0;
1024     session->reclist = reclist_create(nmem);
1025     session->clients_active = 0;
1026     session->clients_cached = 0;
1027     session->settings_modified = 0;
1028     session->session_nmem = nmem;
1029     session->nmem = nmem_create();
1030     session->databases = 0;
1031     session->sorted_results = 0;
1032     session->facet_limits = 0;
1033     session->mergekey = 0;
1034     session->rank = 0;
1035     session->clients_starting = 0;
1036
1037     for (i = 0; i <= SESSION_WATCH_MAX; i++)
1038     {
1039         session->watchlist[i].data = 0;
1040         session->watchlist[i].fun = 0;
1041     }
1042     session->normalize_cache = normalize_cache_create();
1043     session->session_mutex = 0;
1044     pazpar2_mutex_create(&session->session_mutex, tmp_str);
1045     session_log(session, YLOG_LOG, "create");
1046
1047     session_use(1);
1048     return session;
1049 }
1050
1051 const char * client_get_suggestions_xml(struct client *cl, WRBUF wrbuf);
1052
1053 static struct hitsbytarget *hitsbytarget_nb(struct session *se,
1054                                             int *count, NMEM nmem)
1055 {
1056     struct hitsbytarget *res = 0;
1057     struct client_list *l;
1058     size_t sz = 0;
1059
1060     for (l = se->clients_active; l; l = l->next)
1061         sz++;
1062
1063     res = nmem_malloc(nmem, sizeof(*res) * sz);
1064     *count = 0;
1065     for (l = se->clients_active; l; l = l->next)
1066     {
1067         struct client *cl = l->client;
1068         WRBUF w = wrbuf_alloc();
1069         const char *name = session_setting_oneval(client_get_database(cl),
1070                                                   PZ_NAME);
1071         res[*count].id = client_get_id(cl);
1072         res[*count].name = *name ? name : "Unknown";
1073         res[*count].hits = client_get_hits(cl);
1074         res[*count].approximation = client_get_approximation(cl);
1075         res[*count].records = client_get_num_records(cl,
1076                                                      &res[*count].filtered,
1077                                                      0, 0);
1078         res[*count].diagnostic =
1079             client_get_diagnostic(cl, &res[*count].message,
1080                                   &res[*count].addinfo);
1081         res[*count].state = client_get_state_str(cl);
1082         res[*count].connected  = client_get_connection(cl) ? 1 : 0;
1083         session_settings_dump(se, client_get_database(cl), w);
1084         res[*count].settings_xml = nmem_strdup(nmem, wrbuf_cstr(w));
1085         wrbuf_rewind(w);
1086         wrbuf_puts(w, "");
1087         res[*count].suggestions_xml = nmem_strdup(nmem, client_get_suggestions_xml(cl, w));
1088         wrbuf_destroy(w);
1089         (*count)++;
1090     }
1091     return res;
1092 }
1093
1094 struct hitsbytarget *get_hitsbytarget(struct session *se, int *count, NMEM nmem)
1095 {
1096     struct hitsbytarget *p;
1097     session_enter(se, "get_hitsbytarget");
1098     p = hitsbytarget_nb(se, count, nmem);
1099     session_leave(se, "get_hitsbytarget");
1100     return p;
1101 }
1102
1103 // Compares two hitsbytarget nodes by hitcount
1104 static int cmp_ht(const void *p1, const void *p2)
1105 {
1106     const struct hitsbytarget *h1 = p1;
1107     const struct hitsbytarget *h2 = p2;
1108     return h2->hits - h1->hits;
1109 }
1110
1111 // Compares two hitsbytarget nodes by hitcount
1112 static int cmp_ht_approx(const void *p1, const void *p2)
1113 {
1114     const struct hitsbytarget *h1 = p1;
1115     const struct hitsbytarget *h2 = p2;
1116     return h2->approximation - h1->approximation;
1117 }
1118
1119 static int targets_termlist_nb(WRBUF wrbuf, struct session *se, int num,
1120                                NMEM nmem, int version)
1121 {
1122     struct hitsbytarget *ht;
1123     int count, i;
1124
1125     ht = hitsbytarget_nb(se, &count, nmem);
1126     if (version >= 2)
1127         qsort(ht, count, sizeof(struct hitsbytarget), cmp_ht_approx);
1128     else
1129         qsort(ht, count, sizeof(struct hitsbytarget), cmp_ht);
1130     for (i = 0; i < count && i < num && ht[i].hits > 0; i++)
1131     {
1132
1133         // do only print terms which have display names
1134
1135         wrbuf_puts(wrbuf, "<term>\n");
1136
1137         wrbuf_puts(wrbuf, "<id>");
1138         wrbuf_xmlputs(wrbuf, ht[i].id);
1139         wrbuf_puts(wrbuf, "</id>\n");
1140
1141         wrbuf_puts(wrbuf, "<name>");
1142         if (!ht[i].name || !ht[i].name[0])
1143             wrbuf_xmlputs(wrbuf, "NO TARGET NAME");
1144         else
1145             wrbuf_xmlputs(wrbuf, ht[i].name);
1146         wrbuf_puts(wrbuf, "</name>\n");
1147
1148         wrbuf_printf(wrbuf, "<frequency>" ODR_INT_PRINTF "</frequency>\n",
1149                      ht[i].hits);
1150
1151         if (version >= 2) {
1152             // Should not print if we know it isn't a approximation.
1153             wrbuf_printf(wrbuf, "<approximation>" ODR_INT_PRINTF "</approximation>\n", ht[i].approximation);
1154             wrbuf_printf(wrbuf, "<records>%d</records>\n", ht[i].records - ht[i].filtered);
1155             wrbuf_printf(wrbuf, "<filtered>%d</filtered>\n", ht[i].filtered);
1156         }
1157
1158         wrbuf_puts(wrbuf, "<state>");
1159         wrbuf_xmlputs(wrbuf, ht[i].state);
1160         wrbuf_puts(wrbuf, "</state>\n");
1161
1162         wrbuf_printf(wrbuf, "<diagnostic>%d</diagnostic>\n",
1163                      ht[i].diagnostic);
1164         wrbuf_puts(wrbuf, "</term>\n");
1165     }
1166     return count;
1167 }
1168
1169 void perform_termlist(struct http_channel *c, struct session *se,
1170                       const char *name, int num, int version)
1171 {
1172     int j;
1173     NMEM nmem_tmp = nmem_create();
1174     char **names;
1175     int num_names = 0;
1176
1177     if (!name)
1178         name = "*";
1179
1180     nmem_strsplit(nmem_tmp, ",", name, &names, &num_names);
1181
1182     session_enter(se, "perform_termlist");
1183
1184     for (j = 0; j < num_names; j++)
1185     {
1186         const char *tname;
1187         int must_generate_empty = 1; /* bug 5350 */
1188
1189         struct named_termlist *t = se->termlists;
1190         for (; t; t = t->next)
1191         {
1192             tname = t->name;
1193             if (!strcmp(names[j], tname) || !strcmp(names[j], "*"))
1194             {
1195                 struct termlist_score **p = 0;
1196                 int len;
1197
1198                 wrbuf_puts(c->wrbuf, "<list name=\"");
1199                 wrbuf_xmlputs(c->wrbuf, tname);
1200                 wrbuf_puts(c->wrbuf, "\">\n");
1201                 must_generate_empty = 0;
1202
1203                 p = termlist_highscore(t->termlist, &len, nmem_tmp);
1204                 if (p)
1205                 {
1206                     int i;
1207                     for (i = 0; i < len && i < num; i++)
1208                     {
1209                         // prevent sending empty term elements
1210                         if (!p[i]->display_term || !p[i]->display_term[0])
1211                             continue;
1212
1213                         wrbuf_puts(c->wrbuf, "<term>");
1214                         wrbuf_puts(c->wrbuf, "<name>");
1215                         wrbuf_xmlputs(c->wrbuf, p[i]->display_term);
1216                         wrbuf_puts(c->wrbuf, "</name>");
1217
1218                         wrbuf_printf(c->wrbuf,
1219                                      "<frequency>%d</frequency>",
1220                                      p[i]->frequency);
1221                         wrbuf_puts(c->wrbuf, "</term>\n");
1222                     }
1223                 }
1224                 wrbuf_puts(c->wrbuf, "</list>\n");
1225             }
1226         }
1227         tname = "xtargets";
1228         if (!strcmp(names[j], tname) || !strcmp(names[j], "*"))
1229         {
1230             wrbuf_puts(c->wrbuf, "<list name=\"");
1231             wrbuf_xmlputs(c->wrbuf, tname);
1232             wrbuf_puts(c->wrbuf, "\">\n");
1233
1234             targets_termlist_nb(c->wrbuf, se, num, c->nmem, version);
1235             wrbuf_puts(c->wrbuf, "</list>\n");
1236             must_generate_empty = 0;
1237         }
1238         if (must_generate_empty)
1239         {
1240             wrbuf_puts(c->wrbuf, "<list name=\"");
1241             wrbuf_xmlputs(c->wrbuf, names[j]);
1242             wrbuf_puts(c->wrbuf, "\"/>\n");
1243         }
1244     }
1245     session_leave(se, "perform_termlist");
1246     nmem_destroy(nmem_tmp);
1247 }
1248
1249 #ifdef MISSING_HEADERS
1250 void report_nmem_stats(void)
1251 {
1252     size_t in_use, is_free;
1253
1254     nmem_get_memory_in_use(&in_use);
1255     nmem_get_memory_free(&is_free);
1256
1257     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld",
1258             (long) in_use, (long) is_free);
1259 }
1260 #endif
1261
1262 struct record_cluster *show_single_start(struct session *se, const char *id,
1263                                          struct record_cluster **prev_r,
1264                                          struct record_cluster **next_r)
1265 {
1266     struct record_cluster *r = 0;
1267
1268     session_enter(se, "show_single_start");
1269     *prev_r = 0;
1270     *next_r = 0;
1271     reclist_limit(se->reclist, se, 1);
1272
1273     reclist_enter(se->reclist);
1274     while ((r = reclist_read_record(se->reclist)))
1275     {
1276         if (!strcmp(r->recid, id))
1277         {
1278             *next_r = reclist_read_record(se->reclist);
1279             break;
1280         }
1281         *prev_r = r;
1282     }
1283     reclist_leave(se->reclist);
1284     if (!r)
1285         session_leave(se, "show_single_start");
1286     return r;
1287 }
1288
1289 void show_single_stop(struct session *se, struct record_cluster *rec)
1290 {
1291     session_leave(se, "show_single_stop");
1292 }
1293
1294
1295 int session_fetch_more(struct session *se)
1296 {
1297     struct client_list *l;
1298     int ret = 0;
1299
1300     for (l = se->clients_active; l; l = l->next)
1301     {
1302         struct client *cl = l->client;
1303         if (client_get_state(cl) == Client_Idle)
1304         {
1305             if (client_fetch_more(cl))
1306             {
1307                 session_log(se, YLOG_LOG, "%s: more to fetch",
1308                             client_get_id(cl));
1309                 ret = 1;
1310             }
1311             else
1312             {
1313                 int filtered;
1314                 int ingest_failures;
1315                 int record_failures;
1316                 int num = client_get_num_records(
1317                     cl, &filtered, &ingest_failures, &record_failures);
1318
1319                 session_log(se, YLOG_LOG, "%s: hits=" ODR_INT_PRINTF
1320                             " fetched=%d filtered=%d",
1321                             client_get_id(cl),
1322                             client_get_hits(cl),
1323                             num, filtered);
1324                 if (ingest_failures || record_failures)
1325                 {
1326                     session_log(se, YLOG_WARN, "%s:"
1327                                 " ingest failures=%d record failures=%d",
1328                                 client_get_id(cl),
1329                                 ingest_failures, record_failures);
1330                 }
1331             }
1332         }
1333         else
1334         {
1335             session_log(se, YLOG_LOG, "%s: no fetch due to state=%s",
1336                         client_get_id(cl), client_get_state_str(cl));
1337         }
1338
1339     }
1340     return ret;
1341 }
1342
1343 struct record_cluster **show_range_start(struct session *se,
1344                                          struct reclist_sortparms *sp,
1345                                          int start, int *num, int *total,
1346                                          Odr_int *sumhits, Odr_int *approx_hits,
1347                                          void (*show_records_ready)(void *data),
1348                                          struct http_channel *chan)
1349 {
1350     struct record_cluster **recs = 0;
1351     struct reclist_sortparms *spp;
1352     struct client_list *l;
1353     int i;
1354 #if USE_TIMING
1355     yaz_timing_t t = yaz_timing_create();
1356 #endif
1357     session_enter(se, "show_range_start");
1358     *sumhits = 0;
1359     *approx_hits = 0;
1360     *total = 0;
1361     reclist_limit(se->reclist, se, 0);
1362     if (se->relevance)
1363     {
1364         for (spp = sp; spp; spp = spp->next)
1365             if (spp->type == Metadata_type_relevance)
1366             {
1367                 relevance_prepare_read(se->relevance, se->reclist);
1368                 break;
1369             }
1370         for (l = se->clients_active; l; l = l->next) {
1371             *sumhits += client_get_hits(l->client);
1372             *approx_hits += client_get_approximation(l->client);
1373         }
1374     }
1375     reclist_sort(se->reclist, sp);
1376
1377     reclist_enter(se->reclist);
1378     *total = reclist_get_num_records(se->reclist);
1379
1380     for (l = se->clients_active; l; l = l->next)
1381         client_update_show_stat(l->client, 0);
1382
1383     for (i = 0; i < start; i++)
1384     {
1385         struct record_cluster *r = reclist_read_record(se->reclist);
1386         if (!r)
1387         {
1388             *num = 0;
1389             break;
1390         }
1391         else
1392         {
1393             struct record *rec = r->records;
1394             for (;rec; rec = rec->next)
1395                 client_update_show_stat(rec->client, 1);
1396         }
1397     }
1398     recs = nmem_malloc(se->nmem, (*num > 0 ? *num : 1) * sizeof(*recs));
1399     for (i = 0; i < *num; i++)
1400     {
1401         struct record_cluster *r = reclist_read_record(se->reclist);
1402         if (!r)
1403         {
1404             *num = i;
1405             break;
1406         }
1407         else
1408         {
1409             struct record *rec = r->records;
1410             for (;rec; rec = rec->next)
1411                 client_update_show_stat(rec->client, 1);
1412             recs[i] = r;
1413         }
1414     }
1415     reclist_leave(se->reclist);
1416 #if USE_TIMING
1417     yaz_timing_stop(t);
1418     session_log(se, YLOG_LOG, "show %6.5f %3.2f %3.2f",
1419             yaz_timing_get_real(t), yaz_timing_get_user(t),
1420             yaz_timing_get_sys(t));
1421     yaz_timing_destroy(&t);
1422 #endif
1423
1424     if (!session_fetch_more(se))
1425         session_log(se, YLOG_LOG, "can not fetch more");
1426     else
1427     {
1428         show_range_stop(se, recs);
1429         session_log(se, YLOG_LOG, "fetching more in progress");
1430         if (session_set_watch(se, SESSION_WATCH_SHOW,
1431                               show_records_ready, chan, chan))
1432         {
1433             session_log(se, YLOG_WARN, "Ignoring show block");
1434             session_enter(se, "show_range_start");
1435         }
1436         else
1437         {
1438             session_log(se, YLOG_LOG, "session watch OK");
1439             return 0;
1440         }
1441     }
1442     return recs;
1443 }
1444
1445 void show_range_stop(struct session *se, struct record_cluster **recs)
1446 {
1447     session_leave(se, "show_range_stop");
1448 }
1449
1450 void statistics(struct session *se, struct statistics *stat)
1451 {
1452     struct client_list *l;
1453     int count = 0;
1454
1455     memset(stat, 0, sizeof(*stat));
1456     stat->num_hits = 0;
1457     for (l = se->clients_active; l; l = l->next)
1458     {
1459         struct client *cl = l->client;
1460         if (!client_get_connection(cl))
1461             stat->num_no_connection++;
1462         stat->num_hits += client_get_hits(cl);
1463         switch (client_get_state(cl))
1464         {
1465         case Client_Connecting: stat->num_connecting++; break;
1466         case Client_Working: stat->num_working++; break;
1467         case Client_Idle: stat->num_idle++; break;
1468         case Client_Failed: stat->num_failed++; break;
1469         case Client_Error: stat->num_error++; break;
1470         default: break;
1471         }
1472         count++;
1473     }
1474     stat->num_records = se->total_records;
1475
1476     stat->num_clients = count;
1477 }
1478
1479 static struct record_metadata *record_metadata_init(
1480     NMEM nmem, const char *value, enum conf_metadata_type type,
1481     struct _xmlAttr *attr)
1482 {
1483     struct record_metadata *rec_md = record_metadata_create(nmem);
1484     struct record_metadata_attr **attrp = &rec_md->attributes;
1485
1486     for (; attr; attr = attr->next)
1487     {
1488         if (attr->children && attr->children->content)
1489         {
1490             if (strcmp((const char *) attr->name, "type")
1491                 && strcmp((const char *) attr->name, "empty"))
1492             {  /* skip the "type" + "empty" attribute..
1493                   The "Type" is already part of the element in output
1494                   (md-%s) and so repeating it here is redundant */
1495                 *attrp = nmem_malloc(nmem, sizeof(**attrp));
1496                 (*attrp)->name =
1497                     nmem_strdup(nmem, (const char *) attr->name);
1498                 (*attrp)->value =
1499                     nmem_strdup(nmem, (const char *) attr->children->content);
1500                 attrp = &(*attrp)->next;
1501             }
1502         }
1503     }
1504     *attrp = 0;
1505
1506     switch (type)
1507     {
1508     case Metadata_type_generic:
1509     case Metadata_type_skiparticle:
1510         if (strstr(value, "://")) /* looks like a URL */
1511             rec_md->data.text.disp = nmem_strdup(nmem, value);
1512         else
1513             rec_md->data.text.disp =
1514                 normalize7bit_generic(nmem_strdup(nmem, value), " ,/.:([");
1515         rec_md->data.text.sort = 0;
1516         rec_md->data.text.snippet = 0;
1517         break;
1518     case Metadata_type_year:
1519     case Metadata_type_date:
1520     {
1521         int first, last;
1522         int longdate = 0;
1523
1524         if (type == Metadata_type_date)
1525             longdate = 1;
1526         if (extract7bit_dates((char *) value, &first, &last, longdate) < 0)
1527             return 0;
1528
1529         rec_md->data.number.min = first;
1530         rec_md->data.number.max = last;
1531     }
1532     break;
1533     case Metadata_type_float:
1534         rec_md->data.fnumber = atof(value);
1535         break;
1536     case Metadata_type_relevance:
1537     case Metadata_type_position:
1538         return 0;
1539     }
1540     return rec_md;
1541 }
1542
1543 static void mergekey_norm_wr(pp2_charset_fact_t charsets,
1544                              WRBUF norm_wr, const char *value)
1545 {
1546     const char *norm_str;
1547     pp2_charset_token_t prt =
1548         pp2_charset_token_create(charsets, "mergekey");
1549
1550     pp2_charset_token_first(prt, value, 0);
1551     while ((norm_str = pp2_charset_token_next(prt)))
1552     {
1553         if (*norm_str)
1554         {
1555             if (wrbuf_len(norm_wr))
1556                 wrbuf_puts(norm_wr, " ");
1557             wrbuf_puts(norm_wr, norm_str);
1558         }
1559     }
1560     pp2_charset_token_destroy(prt);
1561 }
1562
1563 static int get_mergekey_from_doc(xmlDoc *doc, xmlNode *root, const char *name,
1564                                  struct conf_service *service, WRBUF norm_wr)
1565 {
1566     xmlNode *n;
1567     int no_found = 0;
1568     for (n = root->children; n; n = n->next)
1569     {
1570         if (n->type != XML_ELEMENT_NODE)
1571             continue;
1572         if (!strcmp((const char *) n->name, "metadata"))
1573         {
1574             xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
1575             if (type == NULL) {
1576                 yaz_log(YLOG_FATAL, "Missing type attribute on metadata element. Skipping!");
1577             }
1578             else if (!strcmp(name, (const char *) type))
1579             {
1580                 xmlChar *value = xmlNodeListGetString(doc, n->children, 1);
1581                 if (value && *value)
1582                 {
1583                     if (wrbuf_len(norm_wr) > 0)
1584                         wrbuf_puts(norm_wr, " ");
1585                     wrbuf_puts(norm_wr, name);
1586                     mergekey_norm_wr(service->charsets, norm_wr,
1587                                      (const char *) value);
1588                     no_found++;
1589                 }
1590                 if (value)
1591                     xmlFree(value);
1592             }
1593             xmlFree(type);
1594         }
1595     }
1596     return no_found;
1597 }
1598
1599 static const char *get_mergekey(xmlDoc *doc, xmlNode *root,
1600                                 struct client *cl, int record_no,
1601                                 struct conf_service *service, NMEM nmem,
1602                                 const char *session_mergekey)
1603 {
1604     char *mergekey_norm = 0;
1605     WRBUF norm_wr = wrbuf_alloc();
1606     xmlChar *mergekey;
1607
1608     if (session_mergekey)
1609     {
1610         int i, num = 0;
1611         char **values = 0;
1612         nmem_strsplit_escape2(nmem, ",", session_mergekey, &values,
1613                               &num, 1, '\\', 1);
1614
1615         for (i = 0; i < num; i++)
1616             get_mergekey_from_doc(doc, root, values[i], service, norm_wr);
1617     }
1618     else if ((mergekey = xmlGetProp(root, (xmlChar *) "mergekey")))
1619     {
1620         mergekey_norm_wr(service->charsets, norm_wr, (const char *) mergekey);
1621         xmlFree(mergekey);
1622     }
1623     else
1624     {
1625         /* no mergekey defined in XSL. Look for mergekey metadata instead */
1626         int field_id;
1627         for (field_id = 0; field_id < service->num_metadata; field_id++)
1628         {
1629             struct conf_metadata *ser_md = &service->metadata[field_id];
1630             if (ser_md->mergekey != Metadata_mergekey_no)
1631             {
1632                 int r = get_mergekey_from_doc(doc, root, ser_md->name,
1633                                               service, norm_wr);
1634                 if (r == 0 && ser_md->mergekey == Metadata_mergekey_required)
1635                 {
1636                     /* no mergekey on this one and it is required..
1637                        Generate unique key instead */
1638                     wrbuf_rewind(norm_wr);
1639                     break;
1640                 }
1641             }
1642         }
1643     }
1644
1645     /* generate unique key if none is not generated already or is empty */
1646     if (wrbuf_len(norm_wr) == 0)
1647     {
1648         wrbuf_printf(norm_wr, "position: %s-%d",
1649                      client_get_id(cl), record_no);
1650     }
1651     else
1652     {
1653         const char *lead = "content: ";
1654         wrbuf_insert(norm_wr, 0, lead, strlen(lead));
1655     }
1656     if (wrbuf_len(norm_wr) > 0)
1657         mergekey_norm = nmem_strdup(nmem, wrbuf_cstr(norm_wr));
1658     wrbuf_destroy(norm_wr);
1659     return mergekey_norm;
1660 }
1661
1662 /** \brief see if metadata for pz:recordfilter exists
1663     \param root xml root element of normalized record
1664     \param sdb session database for client
1665     \retval 0 if there is no metadata for pz:recordfilter
1666     \retval 1 if there is metadata for pz:recordfilter
1667
1668     If there is no pz:recordfilter defined, this function returns 1
1669     as well.
1670 */
1671
1672 static int check_record_filter(xmlNode *root, struct session_database *sdb)
1673 {
1674     int match = 0;
1675     xmlNode *n;
1676     const char *s;
1677     s = session_setting_oneval(sdb, PZ_RECORDFILTER);
1678
1679     if (!s || !*s)
1680         return 1;
1681
1682     for (n = root->children; n; n = n->next)
1683     {
1684         if (n->type != XML_ELEMENT_NODE)
1685             continue;
1686         if (!strcmp((const char *) n->name, "metadata"))
1687         {
1688             xmlChar *type = xmlGetProp(n, (xmlChar *) "type");
1689             if (type)
1690             {
1691                 size_t len;
1692                 int substring;
1693                 const char *eq;
1694
1695                 if ((eq = strchr(s, '=')))
1696                     substring = 0;
1697                 else if ((eq = strchr(s, '~')))
1698                     substring = 1;
1699                 if (eq)
1700                     len = eq - s;
1701                 else
1702                     len = strlen(s);
1703                 if (len == strlen((const char *)type) &&
1704                     !memcmp((const char *) type, s, len))
1705                 {
1706                     xmlChar *value = xmlNodeGetContent(n);
1707                     if (value && *value)
1708                     {
1709                         if (!eq ||
1710                             (substring && strstr((const char *) value, eq+1)) ||
1711                             (!substring && !strcmp((const char *) value, eq + 1)))
1712                             match = 1;
1713                     }
1714                     xmlFree(value);
1715                 }
1716                 xmlFree(type);
1717             }
1718         }
1719     }
1720     return match;
1721 }
1722
1723 static int ingest_to_cluster(struct client *cl,
1724                              xmlDoc *xdoc,
1725                              xmlNode *root,
1726                              int record_no,
1727                              struct record_metadata_attr *mergekey);
1728
1729 static int ingest_sub_record(struct client *cl, xmlDoc *xdoc, xmlNode *root,
1730                              int record_no, NMEM nmem,
1731                              struct session_database *sdb,
1732                              struct record_metadata_attr *mergekeys)
1733 {
1734     int ret = 0;
1735     struct session *se = client_get_session(cl);
1736
1737     if (!check_record_filter(root, sdb))
1738     {
1739         session_log(se, YLOG_LOG,
1740                     "Filtered out record no %d from %s",
1741                     record_no, sdb->database->id);
1742         return 0;
1743     }
1744     session_enter(se, "ingest_sub_record");
1745     if (client_get_session(cl) == se && se->relevance)
1746         ret = ingest_to_cluster(cl, xdoc, root, record_no, mergekeys);
1747     session_leave(se, "ingest_sub_record");
1748
1749     return ret;
1750 }
1751
1752 /** \brief ingest XML record
1753     \param cl client holds the result set for record
1754     \param rec record buffer (0 terminated)
1755     \param record_no record position (1, 2, ..)
1756     \param nmem working NMEM
1757     \retval 0 OK
1758     \retval -1 failure
1759     \retval -2 Filtered
1760 */
1761 int ingest_record(struct client *cl, const char *rec,
1762                   int record_no, NMEM nmem)
1763 {
1764     struct session *se = client_get_session(cl);
1765     struct session_database *sdb = client_get_database(cl);
1766     struct conf_service *service = se->service;
1767     xmlDoc *xdoc = normalize_record(se, sdb, service, rec, nmem);
1768     int r = ingest_xml_record(cl, xdoc, record_no, nmem, 0);
1769     client_store_xdoc(cl, record_no, xdoc);
1770     return r;
1771 }
1772
1773 int ingest_xml_record(struct client *cl, xmlDoc *xdoc,
1774                       int record_no, NMEM nmem, int cached_copy)
1775 {
1776     struct session *se = client_get_session(cl);
1777     struct session_database *sdb = client_get_database(cl);
1778     struct conf_service *service = se->service;
1779     xmlNode *root;
1780     int r = 0;
1781     if (!xdoc)
1782         return -1;
1783
1784     if (global_parameters.dump_records)
1785     {
1786         session_log(se, YLOG_LOG, "Normalized record from %s",
1787                     sdb->database->id);
1788         log_xml_doc(xdoc);
1789     }
1790
1791     root = xmlDocGetRootElement(xdoc);
1792
1793     if (!strcmp((const char *) root->name, "cluster"))
1794     {
1795         int no_merge_keys = 0;
1796         int no_merge_dups = 0;
1797         xmlNode *sroot;
1798         struct record_metadata_attr *mk = 0;
1799
1800         for (sroot = root->children; sroot; sroot = sroot->next)
1801             if (sroot->type == XML_ELEMENT_NODE &&
1802                 !strcmp((const char *) sroot->name, "record"))
1803             {
1804                 struct record_metadata_attr **mkp;
1805                 const char *mergekey_norm =
1806                     get_mergekey(xdoc, sroot, cl, record_no, service, nmem,
1807                                  se->mergekey);
1808                 if (!mergekey_norm)
1809                 {
1810                     r = -1;
1811                     break;
1812                 }
1813                 for (mkp = &mk; *mkp; mkp = &(*mkp)->next)
1814                     if (!strcmp((*mkp)->value, mergekey_norm))
1815                         break;
1816                 if (!*mkp)
1817                 {
1818                     *mkp = (struct record_metadata_attr*)
1819                         nmem_malloc(nmem, sizeof(**mkp));
1820                     (*mkp)->name = 0;
1821                     (*mkp)->value = nmem_strdup(nmem, mergekey_norm);
1822                     (*mkp)->next = 0;
1823                     no_merge_keys++;
1824                 }
1825                 else
1826                     no_merge_dups++;
1827             }
1828         if (no_merge_keys > 1 || no_merge_dups > 0)
1829         {
1830             yaz_log(YLOG_LOG, "Got %d mergekeys, %d dups for position %d",
1831                     no_merge_keys, no_merge_dups, record_no);
1832         }
1833         for (sroot = root->children; !r && sroot; sroot = sroot->next)
1834             if (sroot->type == XML_ELEMENT_NODE &&
1835                 !strcmp((const char *) sroot->name, "record"))
1836             {
1837                 if (!cached_copy)
1838                     insert_settings_values(sdb, xdoc, root, service);
1839                 r = ingest_sub_record(cl, xdoc, sroot, record_no, nmem, sdb,
1840                                       mk);
1841             }
1842     }
1843     else if (!strcmp((const char *) root->name, "record"))
1844     {
1845         const char *mergekey_norm =
1846             get_mergekey(xdoc, root, cl, record_no, service, nmem,
1847                          se->mergekey);
1848         if (mergekey_norm)
1849         {
1850             struct record_metadata_attr *mk = (struct record_metadata_attr*)
1851                 nmem_malloc(nmem, sizeof(*mk));
1852             mk->name = 0;
1853             mk->value = nmem_strdup(nmem, mergekey_norm);
1854             mk->next = 0;
1855
1856             if (!cached_copy)
1857                 insert_settings_values(sdb, xdoc, root, service);
1858             r = ingest_sub_record(cl, xdoc, root, record_no, nmem, sdb, mk);
1859         }
1860     }
1861     else
1862     {
1863         session_log(se, YLOG_WARN, "Bad pz root element: %s",
1864                     (const char *) root->name);
1865         r = -1;
1866     }
1867     return r;
1868 }
1869
1870
1871 //    struct conf_metadata *ser_md = &service->metadata[md_field_id];
1872 //    struct record_metadata *rec_md = record->metadata[md_field_id];
1873 static int match_metadata_local(struct conf_service *service,
1874                                 struct conf_metadata *ser_md,
1875                                 struct record_metadata *rec_md0,
1876                                 char **values, int num_v)
1877 {
1878     int i;
1879     struct record_metadata *rec_md = rec_md0;
1880     WRBUF val_wr = 0;
1881     WRBUF text_wr = wrbuf_alloc();
1882     for (i = 0; i < num_v; )
1883     {
1884         if (rec_md)
1885         {
1886             if (ser_md->type == Metadata_type_year
1887                 || ser_md->type == Metadata_type_date)
1888             {
1889                 int y = atoi(values[i]);
1890                 if (y >= rec_md->data.number.min
1891                     && y <= rec_md->data.number.max)
1892                     break;
1893             }
1894             else
1895             {
1896                 if (!val_wr)
1897                 {
1898                     val_wr = wrbuf_alloc();
1899                     mergekey_norm_wr(service->charsets, val_wr, values[i]);
1900                 }
1901                 wrbuf_rewind(text_wr);
1902                 mergekey_norm_wr(service->charsets, text_wr,
1903                                  rec_md->data.text.disp);
1904                 if (!strcmp(wrbuf_cstr(val_wr), wrbuf_cstr(text_wr)))
1905                     break;
1906             }
1907             rec_md = rec_md->next;
1908         }
1909         else
1910         {
1911             rec_md = rec_md0;
1912             wrbuf_destroy(val_wr);
1913             val_wr = 0;
1914             i++;
1915         }
1916     }
1917     wrbuf_destroy(val_wr);
1918     wrbuf_destroy(text_wr);
1919     return i < num_v ? 1 : 0;
1920 }
1921
1922 int session_check_cluster_limit(struct session *se, struct record_cluster *rec)
1923 {
1924     int i;
1925     struct conf_service *service = se->service;
1926     int ret = 1;
1927     const char *name;
1928     const char *value;
1929     NMEM nmem_tmp = nmem_create();
1930
1931     for (i = 0; (name = facet_limits_get(se->facet_limits, i, &value)); i++)
1932     {
1933         int j;
1934         for (j = 0; j < service->num_metadata; j++)
1935         {
1936             struct conf_metadata *md = service->metadata + j;
1937             if (!strcmp(md->name, name) && md->limitcluster)
1938             {
1939                 char **values = 0;
1940                 int num = 0;
1941                 int md_field_id =
1942                     conf_service_metadata_field_id(service,
1943                                                    md->limitcluster);
1944
1945                 if (md_field_id < 0)
1946                 {
1947                     ret = 0;
1948                     break;
1949                 }
1950
1951                 nmem_strsplit_escape2(nmem_tmp, "|", value, &values,
1952                                       &num, 1, '\\', 1);
1953
1954                 if (!match_metadata_local(service,
1955                                           &service->metadata[md_field_id],
1956                                           rec->metadata[md_field_id],
1957                                           values, num))
1958                 {
1959                     ret = 0;
1960                     break;
1961                 }
1962             }
1963         }
1964     }
1965     nmem_destroy(nmem_tmp);
1966     return ret;
1967 }
1968
1969 // Skip record on non-zero
1970 static int check_limit_local(struct client *cl,
1971                              struct record *record,
1972                              int record_no)
1973 {
1974     int skip_record = 0;
1975     struct session *se = client_get_session(cl);
1976     struct conf_service *service = se->service;
1977     NMEM nmem_tmp = nmem_create();
1978     struct session_database *sdb = client_get_database(cl);
1979     int l = 0;
1980     while (!skip_record)
1981     {
1982         int md_field_id;
1983         char **values = 0;
1984         int num_v = 0;
1985         const char *name =
1986             client_get_facet_limit_local(cl, sdb, &l, nmem_tmp,
1987                                          &num_v, &values);
1988         if (!name)
1989             break;
1990
1991         if (!strcmp(name, "*"))
1992         {
1993             for (md_field_id = 0; md_field_id < service->num_metadata;
1994                  md_field_id++)
1995             {
1996                 if (match_metadata_local(
1997                         service,
1998                         &service->metadata[md_field_id],
1999                         record->metadata[md_field_id],
2000                         values, num_v))
2001                     break;
2002             }
2003             if (md_field_id == service->num_metadata)
2004                 skip_record = 1;
2005         }
2006         else
2007         {
2008             md_field_id = conf_service_metadata_field_id(service, name);
2009             if (md_field_id < 0)
2010             {
2011                 skip_record = 1;
2012                 break;
2013             }
2014             if (!match_metadata_local(
2015                     service,
2016                     &service->metadata[md_field_id],
2017                     record->metadata[md_field_id],
2018                     values, num_v))
2019             {
2020                 skip_record = 1;
2021             }
2022         }
2023     }
2024     nmem_destroy(nmem_tmp);
2025     return skip_record;
2026 }
2027
2028 static int ingest_to_cluster(struct client *cl,
2029                              xmlDoc *xdoc,
2030                              xmlNode *root,
2031                              int record_no,
2032                              struct record_metadata_attr *merge_keys)
2033 {
2034     xmlNode *n;
2035     xmlChar *type = 0;
2036     xmlChar *value = 0;
2037     struct session *se = client_get_session(cl);
2038     struct conf_service *service = se->service;
2039     int term_factor = 1;
2040     struct record_cluster *cluster;
2041     struct record_metadata **metadata0;
2042     struct session_database *sdb = client_get_database(cl);
2043     NMEM ingest_nmem = 0;
2044     char **rank_values = 0;
2045     int rank_num = 0;
2046     struct record *record = record_create(se->nmem,
2047                                           service->num_metadata,
2048                                           service->num_sortkeys, cl,
2049                                           record_no);
2050
2051     for (n = root->children; n; n = n->next)
2052     {
2053         if (type)
2054             xmlFree(type);
2055         if (value)
2056             xmlFree(value);
2057         type = value = 0;
2058
2059         if (n->type != XML_ELEMENT_NODE)
2060             continue;
2061         if (!strcmp((const char *) n->name, "metadata"))
2062         {
2063             struct conf_metadata *ser_md = 0;
2064             struct record_metadata **wheretoput = 0;
2065             struct record_metadata *rec_md = 0;
2066             int md_field_id = -1;
2067
2068             type = xmlGetProp(n, (xmlChar *) "type");
2069             value = xmlNodeListGetString(xdoc, n->children, 1);
2070             if (!type)
2071                 continue;
2072             if (!value || !*value)
2073             {
2074                 xmlChar *empty = xmlGetProp(n, (xmlChar *) "empty");
2075                 if (!empty)
2076                     continue;
2077                 if (value)
2078                     xmlFree(value);
2079                 value = empty;
2080             }
2081             md_field_id
2082                 = conf_service_metadata_field_id(service, (const char *) type);
2083             if (md_field_id < 0)
2084             {
2085                 if (se->number_of_warnings_unknown_metadata == 0)
2086                 {
2087                     session_log(se, YLOG_WARN,
2088                             "Ignoring unknown metadata element: %s", type);
2089                 }
2090                 se->number_of_warnings_unknown_metadata++;
2091                 continue;
2092             }
2093
2094             ser_md = &service->metadata[md_field_id];
2095
2096             // non-merged metadata
2097             rec_md = record_metadata_init(se->nmem, (const char *) value,
2098                                           ser_md->type, n->properties);
2099             if (!rec_md)
2100             {
2101                 session_log(se, YLOG_WARN, "bad metadata data '%s' "
2102                             "for element '%s'", value, type);
2103                 continue;
2104             }
2105
2106             if (ser_md->type == Metadata_type_generic)
2107             {
2108                 WRBUF w = wrbuf_alloc();
2109                 if (relevance_snippet(se->relevance,
2110                                       (char*) value, ser_md->name, w))
2111                     rec_md->data.text.snippet = nmem_strdup(se->nmem,
2112                                                             wrbuf_cstr(w));
2113                 wrbuf_destroy(w);
2114             }
2115
2116
2117             wheretoput = &record->metadata[md_field_id];
2118             while (*wheretoput)
2119                 wheretoput = &(*wheretoput)->next;
2120             *wheretoput = rec_md;
2121         }
2122     }
2123
2124     if (check_limit_local(cl, record, record_no))
2125     {
2126         if (type)
2127             xmlFree(type);
2128         if (value)
2129             xmlFree(value);
2130         return -2;
2131     }
2132     cluster = reclist_insert(se->reclist, se->relevance, service, record,
2133                              merge_keys, &se->total_merged);
2134     if (!cluster)
2135     {
2136         if (type)
2137             xmlFree(type);
2138         if (value)
2139             xmlFree(value);
2140         return 0; // complete match with existing record
2141     }
2142
2143     {
2144         const char *use_term_factor_str =
2145             session_setting_oneval(sdb, PZ_TERMLIST_TERM_FACTOR);
2146         if (use_term_factor_str && use_term_factor_str[0] == '1')
2147         {
2148             int maxrecs = client_get_maxrecs(cl);
2149             int hits = (int) client_get_hits(cl);
2150             term_factor = MAX(hits, maxrecs) /  MAX(1, maxrecs);
2151             assert(term_factor >= 1);
2152             session_log(se, YLOG_DEBUG, "Using term factor: %d (%d / %d)",
2153                         term_factor, MAX(hits, maxrecs), MAX(1, maxrecs));
2154         }
2155     }
2156
2157     if (global_parameters.dump_records)
2158         session_log(se, YLOG_LOG, "Cluster id %s from %s (#%d)", cluster->recid,
2159                     sdb->database->id, record_no);
2160
2161     // original metadata, to check if first existence of a field
2162     metadata0 = xmalloc(sizeof(*metadata0) * service->num_metadata);
2163     memcpy(metadata0, cluster->metadata,
2164            sizeof(*metadata0) * service->num_metadata);
2165
2166     ingest_nmem = nmem_create();
2167     if (se->rank)
2168     {
2169         yaz_log(YLOG_LOG, "local in sort : %s", se->rank);
2170         nmem_strsplit_escape2(ingest_nmem, ",", se->rank, &rank_values,
2171                               &rank_num, 1, '\\', 1);
2172     }
2173
2174     // now parsing XML record and adding data to cluster or record metadata
2175     for (n = root->children; n; n = n->next)
2176     {
2177         if (type)
2178             xmlFree(type);
2179         if (value)
2180             xmlFree(value);
2181         type = value = 0;
2182
2183         if (n->type != XML_ELEMENT_NODE)
2184             continue;
2185         if (!strcmp((const char *) n->name, "metadata"))
2186         {
2187             struct conf_metadata *ser_md = 0;
2188             struct conf_sortkey *ser_sk = 0;
2189             struct record_metadata **wheretoput = 0;
2190             struct record_metadata *rec_md = 0;
2191             int md_field_id = -1;
2192             int sk_field_id = -1;
2193             const char *rank = 0;
2194             xmlChar *xml_rank = 0;
2195
2196             type = xmlGetProp(n, (xmlChar *) "type");
2197             value = xmlNodeListGetString(xdoc, n->children, 1);
2198
2199             if (!type || !value || !*value)
2200                 continue;
2201
2202             md_field_id
2203                 = conf_service_metadata_field_id(service, (const char *) type);
2204             if (md_field_id < 0)
2205                 continue;
2206
2207             ser_md = &service->metadata[md_field_id];
2208
2209             if (ser_md->sortkey_offset >= 0)
2210             {
2211                 sk_field_id = ser_md->sortkey_offset;
2212                 ser_sk = &service->sortkeys[sk_field_id];
2213             }
2214
2215             // merged metadata
2216             rec_md = record_metadata_init(se->nmem, (const char *) value,
2217                                           ser_md->type, 0);
2218
2219             // see if the field was not in cluster already (from beginning)
2220
2221             if (!rec_md)
2222                 continue;
2223
2224             if (rank_num)
2225             {
2226                 int i;
2227                 for (i = 0; i < rank_num; i++)
2228                 {
2229                     const char *val = rank_values[i];
2230                     const char *cp = strchr(val, '=');
2231                     if (!cp)
2232                         continue;
2233                     if ((cp - val) == strlen((const char *) type)
2234                         && !memcmp(val, type, cp - val))
2235                     {
2236                         rank = cp + 1;
2237                         break;
2238                     }
2239                 }
2240             }
2241             else
2242             {
2243                 xml_rank = xmlGetProp(n, (xmlChar *) "rank");
2244                 rank = xml_rank ? (const char *) xml_rank : ser_md->rank;
2245             }
2246
2247             wheretoput = &cluster->metadata[md_field_id];
2248
2249             if (ser_md->merge == Metadata_merge_first)
2250             {
2251                 if (!metadata0[md_field_id])
2252                 {
2253                     while (*wheretoput)
2254                         wheretoput = &(*wheretoput)->next;
2255                     *wheretoput = rec_md;
2256                 }
2257             }
2258             else if (ser_md->merge == Metadata_merge_unique)
2259             {
2260                 while (*wheretoput)
2261                 {
2262                     if (!strcmp((const char *) (*wheretoput)->data.text.disp,
2263                                 rec_md->data.text.disp))
2264                         break;
2265                     wheretoput = &(*wheretoput)->next;
2266                 }
2267                 if (!*wheretoput)
2268                     *wheretoput = rec_md;
2269             }
2270             else if (ser_md->merge == Metadata_merge_longest)
2271             {
2272                 if (!*wheretoput
2273                     || strlen(rec_md->data.text.disp)
2274                     > strlen((*wheretoput)->data.text.disp))
2275                 {
2276                     *wheretoput = rec_md;
2277                     if (ser_sk)
2278                     {
2279                         pp2_charset_token_t prt;
2280                         const char *sort_str = 0;
2281                         int skip_article =
2282                             ser_sk->type == Metadata_type_skiparticle;
2283
2284                         if (!cluster->sortkeys[sk_field_id])
2285                             cluster->sortkeys[sk_field_id] =
2286                                 nmem_malloc(se->nmem,
2287                                             sizeof(union data_types));
2288
2289                         prt =
2290                             pp2_charset_token_create(service->charsets, "sort");
2291
2292                         pp2_charset_token_first(prt, rec_md->data.text.disp,
2293                                                 skip_article);
2294
2295                         pp2_charset_token_next(prt);
2296
2297                         sort_str = pp2_get_sort(prt);
2298
2299                         cluster->sortkeys[sk_field_id]->text.disp =
2300                             rec_md->data.text.disp;
2301                         if (!sort_str)
2302                         {
2303                             sort_str = rec_md->data.text.disp;
2304                             session_log(se, YLOG_WARN,
2305                                     "Could not make sortkey. Bug #1858");
2306                         }
2307                         cluster->sortkeys[sk_field_id]->text.sort =
2308                             nmem_strdup(se->nmem, sort_str);
2309                         pp2_charset_token_destroy(prt);
2310                     }
2311                 }
2312             }
2313             else if (ser_md->merge == Metadata_merge_all)
2314             {
2315                 while (*wheretoput)
2316                     wheretoput = &(*wheretoput)->next;
2317                 *wheretoput = rec_md;
2318             }
2319             else if (ser_md->merge == Metadata_merge_range)
2320             {
2321                 if (!*wheretoput)
2322                 {
2323                     *wheretoput = rec_md;
2324                     if (ser_sk)
2325                         cluster->sortkeys[sk_field_id]
2326                             = &rec_md->data;
2327                 }
2328                 else
2329                 {
2330                     int this_min = rec_md->data.number.min;
2331                     int this_max = rec_md->data.number.max;
2332                     if (this_min < (*wheretoput)->data.number.min)
2333                         (*wheretoput)->data.number.min = this_min;
2334                     if (this_max > (*wheretoput)->data.number.max)
2335                         (*wheretoput)->data.number.max = this_max;
2336                 }
2337             }
2338
2339             // ranking of _all_ fields enabled ...
2340             if (rank)
2341             {
2342                 relevance_countwords(se->relevance, cluster,
2343                                      (char *) value, rank, ser_md->name);
2344             }
2345             // construct facets ... unless the client already has reported them
2346             if (ser_md->termlist && !client_has_facet(cl, (char *) type))
2347             {
2348                 if (ser_md->type == Metadata_type_year)
2349                 {
2350                     char year[64];
2351                     sprintf(year, "%d", rec_md->data.number.max);
2352
2353                     add_facet(se, (char *) type, year, term_factor);
2354                     if (rec_md->data.number.max != rec_md->data.number.min)
2355                     {
2356                         sprintf(year, "%d", rec_md->data.number.min);
2357                         add_facet(se, (char *) type, year, term_factor);
2358                     }
2359                 }
2360                 else
2361                     add_facet(se, (char *) type, (char *) value, term_factor);
2362             }
2363
2364             // cleaning up
2365             if (xml_rank)
2366                 xmlFree(xml_rank);
2367             xmlFree(type);
2368             xmlFree(value);
2369             type = value = 0;
2370         }
2371         else
2372         {
2373             if (se->number_of_warnings_unknown_elements == 0)
2374                 session_log(se, YLOG_WARN,
2375                         "Unexpected element in internal record: %s", n->name);
2376             se->number_of_warnings_unknown_elements++;
2377         }
2378     }
2379     if (type)
2380         xmlFree(type);
2381     if (value)
2382         xmlFree(value);
2383
2384     nmem_destroy(ingest_nmem);
2385     xfree(metadata0);
2386     relevance_donerecord(se->relevance, cluster);
2387     se->total_records++;
2388
2389     return 0;
2390 }
2391
2392 void session_log(struct session *s, int level, const char *fmt, ...)
2393 {
2394     char buf[1024];
2395     va_list ap;
2396     va_start(ap, fmt);
2397
2398     yaz_vsnprintf(buf, sizeof(buf)-30, fmt, ap);
2399     yaz_log(level, "Session %u: %s", s ? s->session_id : 0, buf);
2400
2401     va_end(ap);
2402 }
2403
2404 /*
2405  * Local variables:
2406  * c-basic-offset: 4
2407  * c-file-style: "Stroustrup"
2408  * indent-tabs-mode: nil
2409  * End:
2410  * vim: shiftwidth=4 tabstop=8 expandtab
2411  */
2412