src/session.c

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