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