src/logic.c

   1 /* $Id: logic.c,v 1.27 2007-05-10 11:46:09 adam Exp $
   2    Copyright (c) 2006-2007, Index Data.
   3
   4 This file is part of Pazpar2.
   5
   6 Pazpar2 is free software; you can redistribute it and/or modify it under
   7 the terms of the GNU General Public License as published by the Free
   8 Software Foundation; either version 2, or (at your option) any later
   9 version.
  10
  11 Pazpar2 is distributed in the hope that it will be useful, but WITHOUT ANY
  12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14 for more details.
  15
  16 You should have received a copy of the GNU General Public License
  17 along with Pazpar2; see the file LICENSE.  If not, write to the
  18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
  19 02111-1307, USA.
  20  */
  21
  22 /** \file logic.c
  23     \brief high-level logic; mostly user sessions and settings
  24 */
  25
  26 #include <stdlib.h>
  27 #include <stdio.h>
  28 #include <string.h>
  29 #include <sys/time.h>
  30 #include <unistd.h>
  31 #include <sys/socket.h>
  32 #include <netdb.h>
  33 #include <signal.h>
  34 #include <ctype.h>
  35 #include <assert.h>
  36
  37 #include <yaz/marcdisp.h>
  38 #include <yaz/comstack.h>
  39 #include <yaz/tcpip.h>
  40 #include <yaz/proto.h>
  41 #include <yaz/readconf.h>
  42 #include <yaz/pquery.h>
  43 #include <yaz/otherinfo.h>
  44 #include <yaz/yaz-util.h>
  45 #include <yaz/nmem.h>
  46 #include <yaz/query-charset.h>
  47 #include <yaz/querytowrbuf.h>
  48 #if YAZ_VERSIONL >= 0x020163
  49 #include <yaz/oid_db.h>
  50 #endif
  51
  52 #if HAVE_CONFIG_H
  53 #include "cconfig.h"
  54 #endif
  55
  56 #define USE_TIMING 0
  57 #if USE_TIMING
  58 #include <yaz/timing.h>
  59 #endif
  60
  61 #include <netinet/in.h>
  62
  63 #include "pazpar2.h"
  64 #include "eventl.h"
  65 #include "http.h"
  66 #include "termlists.h"
  67 #include "reclists.h"
  68 #include "relevance.h"
  69 #include "config.h"
  70 #include "database.h"
  71 #include "client.h"
  72 #include "settings.h"
  73 #include "normalize7bit.h"
  74
  75 #define MAX_CHUNK 15
  76
  77 // Note: Some things in this structure will eventually move to configuration
  78 struct parameters global_parameters =
  79 {
  80     "",
  81     "",
  82     "",
  83     "",
  84     0,
  85     0,
  86     30,
  87     "81",
  88     "Index Data PazPar2",
  89     VERSION,
  90     600, // 10 minutes
  91     60,
  92     100,
  93     MAX_CHUNK,
  94     0,
  95     0
  96 };
  97
  98
  99 // Recursively traverse query structure to extract terms.
 100 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 101 {
 102     char **words;
 103     int numwords;
 104     int i;
 105
 106     switch (n->kind)
 107     {
 108         case CCL_RPN_AND:
 109         case CCL_RPN_OR:
 110         case CCL_RPN_NOT:
 111         case CCL_RPN_PROX:
 112             pull_terms(nmem, n->u.p[0], termlist, num);
 113             pull_terms(nmem, n->u.p[1], termlist, num);
 114             break;
 115         case CCL_RPN_TERM:
 116             nmem_strsplit(nmem, " ", n->u.t.term, &words, &numwords);
 117             for (i = 0; i < numwords; i++)
 118                 termlist[(*num)++] = words[i];
 119             break;
 120         default: // NOOP
 121             break;
 122     }
 123 }
 124
 125
 126
 127 static void add_facet(struct session *s, const char *type, const char *value)
 128 {
 129     int i;
 130
 131     if (!*value)
 132         return;
 133     for (i = 0; i < s->num_termlists; i++)
 134         if (!strcmp(s->termlists[i].name, type))
 135             break;
 136     if (i == s->num_termlists)
 137     {
 138         if (i == SESSION_MAX_TERMLISTS)
 139         {
 140             yaz_log(YLOG_FATAL, "Too many termlists");
 141             return;
 142         }
 143
 144         s->termlists[i].name = nmem_strdup(s->nmem, type);
 145         s->termlists[i].termlist
 146             = termlist_create(s->nmem, s->expected_maxrecs, 15);
 147         s->num_termlists = i + 1;
 148     }
 149     termlist_insert(s->termlists[i].termlist, value);
 150 }
 151
 152 xmlDoc *normalize_record(struct session_database *sdb, Z_External *rec)
 153 {
 154     struct database_retrievalmap *m;
 155     struct database *db = sdb->database;
 156     xmlNode *res;
 157     xmlDoc *rdoc;
 158
 159     // First normalize to XML
 160     if (sdb->yaz_marc)
 161     {
 162         char *buf;
 163         int len;
 164         if (rec->which != Z_External_octet)
 165         {
 166             yaz_log(YLOG_WARN, "Unexpected external branch, probably BER %s",
 167                     db->url);
 168             return 0;
 169         }
 170         buf = (char*) rec->u.octet_aligned->buf;
 171         len = rec->u.octet_aligned->len;
 172         if (yaz_marc_read_iso2709(sdb->yaz_marc, buf, len) < 0)
 173         {
 174             yaz_log(YLOG_WARN, "Failed to decode MARC %s", db->url);
 175             return 0;
 176         }
 177
 178         yaz_marc_write_using_libxml2(sdb->yaz_marc, 1);
 179         if (yaz_marc_write_xml(sdb->yaz_marc, &res,
 180                     "http://www.loc.gov/MARC21/slim", 0, 0) < 0)
 181         {
 182             yaz_log(YLOG_WARN, "Failed to encode as XML %s",
 183                     db->url);
 184             return 0;
 185         }
 186         rdoc = xmlNewDoc((xmlChar *) "1.0");
 187         xmlDocSetRootElement(rdoc, res);
 188
 189     }
 190     else
 191     {
 192         yaz_log(YLOG_FATAL,
 193                 "Unknown native_syntax in normalize_record from %s",
 194                 db->url);
 195         return 0;
 196     }
 197
 198     if (global_parameters.dump_records){
 199         fprintf(stderr,
 200                 "Input Record (normalized) from %s\n----------------\n",
 201                 db->url);
 202 #if LIBXML_VERSION >= 20600
 203         xmlDocFormatDump(stderr, rdoc, 1);
 204 #else
 205         xmlDocDump(stderr, rdoc);
 206 #endif
 207     }
 208
 209     for (m = sdb->map; m; m = m->next){
 210         xmlDoc *new = 0;
 211
 212 #if 1
 213         {
 214             xmlNodePtr root = 0;
 215             new = xsltApplyStylesheet(m->stylesheet, rdoc, 0);
 216             root= xmlDocGetRootElement(new);
 217         if (!new || !root || !(root->children))
 218         {
 219             yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
 220                     sdb->database->url);
 221             xmlFreeDoc(new);
 222             xmlFreeDoc(rdoc);
 223             return 0;
 224         }
 225         }
 226 #else
 227         // do it another way to detect transformation errors right now
 228         // but does not seem to work either!
 229         {
 230             xsltTransformContextPtr ctxt;
 231             ctxt = xsltNewTransformContext(m->stylesheet, rdoc);
 232             new = xsltApplyStylesheetUser(m->stylesheet, rdoc, 0, 0, 0, ctxt);
 233             if ((ctxt->state == XSLT_STATE_ERROR) ||
 234                 (ctxt->state == XSLT_STATE_STOPPED)){
 235                 yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
 236                         cl->database->database->url);
 237                 xmlFreeDoc(new);
 238                 xmlFreeDoc(rdoc);
 239                 return 0;
 240             }
 241         }
 242 #endif
 243
 244         xmlFreeDoc(rdoc);
 245         rdoc = new;
 246     }
 247     if (global_parameters.dump_records)
 248     {
 249         fprintf(stderr, "Record from %s\n----------------\n",
 250                 sdb->database->url);
 251 #if LIBXML_VERSION >= 20600
 252         xmlDocFormatDump(stderr, rdoc, 1);
 253 #else
 254         xmlDocDump(stderr, rdoc);
 255 #endif
 256     }
 257     return rdoc;
 258 }
 259
 260 // Retrieve first defined value for 'name' for given database.
 261 // Will be extended to take into account user associated with session
 262 char *session_setting_oneval(struct session_database *db, int offset)
 263 {
 264     if (!db->settings[offset])
 265         return "";
 266     return db->settings[offset]->value;
 267 }
 268
 269
 270
 271 // Initialize YAZ Map structures for MARC-based targets
 272 static int prepare_yazmarc(struct session_database *sdb)
 273 {
 274     char *s;
 275
 276     if (!sdb->settings)
 277     {
 278         yaz_log(YLOG_WARN, "No settings for %s", sdb->database->url);
 279         return -1;
 280     }
 281     if ((s = session_setting_oneval(sdb, PZ_NATIVESYNTAX))
 282         && !strncmp(s, "iso2709", 7))
 283     {
 284         char *encoding = "marc-8s", *e;
 285         yaz_iconv_t cm;
 286
 287         // See if a native encoding is specified
 288         if ((e = strchr(s, ';')))
 289             encoding = e + 1;
 290
 291         sdb->yaz_marc = yaz_marc_create();
 292         yaz_marc_subfield_str(sdb->yaz_marc, "\t");
 293
 294         cm = yaz_iconv_open("utf-8", encoding);
 295         if (!cm)
 296         {
 297             yaz_log(YLOG_FATAL,
 298                     "Unable to map from %s to UTF-8 for target %s",
 299                     encoding, sdb->database->url);
 300             return -1;
 301         }
 302         yaz_marc_iconv(sdb->yaz_marc, cm);
 303     }
 304     return 0;
 305 }
 306
 307 // Prepare XSLT stylesheets for record normalization
 308 // Structures are allocated on the session_wide nmem to avoid having
 309 // to recompute this for every search. This would lead
 310 // to leaking if a single session was to repeatedly change the PZ_XSLT
 311 // setting. However, this is not a realistic use scenario.
 312 static int prepare_map(struct session *se, struct session_database *sdb)
 313 {
 314    char *s;
 315
 316     if (!sdb->settings)
 317     {
 318         yaz_log(YLOG_WARN, "No settings on %s", sdb->database->url);
 319         return -1;
 320     }
 321     if ((s = session_setting_oneval(sdb, PZ_XSLT)))
 322     {
 323         char **stylesheets;
 324         struct database_retrievalmap **m = &sdb->map;
 325         int num, i;
 326
 327         nmem_strsplit(se->session_nmem, ",", s, &stylesheets, &num);
 328         for (i = 0; i < num; i++)
 329         {
 330             (*m) = nmem_malloc(se->session_nmem, sizeof(**m));
 331             (*m)->next = 0;
 332             if (!((*m)->stylesheet = conf_load_stylesheet(stylesheets[i])))
 333             {
 334                 yaz_log(YLOG_FATAL, "Unable to load stylesheet: %s",
 335                         stylesheets[i]);
 336                 return -1;
 337             }
 338             m = &(*m)->next;
 339         }
 340     }
 341     if (!sdb->map)
 342         yaz_log(YLOG_WARN, "No Normalization stylesheet for target %s",
 343                 sdb->database->url);
 344     return 0;
 345 }
 346
 347 // This analyzes settings and recomputes any supporting data structures
 348 // if necessary.
 349 static int prepare_session_database(struct session *se,
 350                                     struct session_database *sdb)
 351 {
 352     if (!sdb->settings)
 353     {
 354         yaz_log(YLOG_WARN,
 355                 "No settings associated with %s", sdb->database->url);
 356         return -1;
 357     }
 358     if (sdb->settings[PZ_NATIVESYNTAX] && !sdb->yaz_marc)
 359     {
 360         if (prepare_yazmarc(sdb) < 0)
 361             return -1;
 362     }
 363     if (sdb->settings[PZ_XSLT] && !sdb->map)
 364     {
 365         if (prepare_map(se, sdb) < 0)
 366             return -1;
 367     }
 368     return 0;
 369 }
 370
 371
 372 void session_set_watch(struct session *s, int what,
 373                        session_watchfun fun, void *data)
 374 {
 375     s->watchlist[what].fun = fun;
 376     s->watchlist[what].data = data;
 377 }
 378
 379 void session_alert_watch(struct session *s, int what)
 380 {
 381     if (!s->watchlist[what].fun)
 382         return;
 383     (*s->watchlist[what].fun)(s->watchlist[what].data);
 384     s->watchlist[what].fun = 0;
 385     s->watchlist[what].data = 0;
 386 }
 387
 388 //callback for grep_databases
 389 static void select_targets_callback(void *context, struct session_database *db)
 390 {
 391     struct session *se = (struct session*) context;
 392     struct client *cl = client_create();
 393     client_set_database(cl, db);
 394     client_set_session(cl, se);
 395 }
 396
 397 // Associates a set of clients with a session;
 398 // Note: Session-databases represent databases with per-session
 399 // setting overrides
 400 int select_targets(struct session *se, struct database_criterion *crit)
 401 {
 402     while (se->clients)
 403         client_destroy(se->clients);
 404
 405     return session_grep_databases(se, crit, select_targets_callback);
 406 }
 407
 408 int session_active_clients(struct session *s)
 409 {
 410     struct client *c;
 411     int res = 0;
 412
 413     for (c = s->clients; c; c = client_next_in_session(c))
 414         if (client_is_active(c))
 415             res++;
 416
 417     return res;
 418 }
 419
 420 // parses crit1=val1,crit2=val2|val3,...
 421 static struct database_criterion *parse_filter(NMEM m, const char *buf)
 422 {
 423     struct database_criterion *res = 0;
 424     char **values;
 425     int num;
 426     int i;
 427
 428     if (!buf || !*buf)
 429         return 0;
 430     nmem_strsplit(m, ",", buf,  &values, &num);
 431     for (i = 0; i < num; i++)
 432     {
 433         char **subvalues;
 434         int subnum;
 435         int subi;
 436         struct database_criterion *new = nmem_malloc(m, sizeof(*new));
 437         char *eq = strchr(values[i], '=');
 438         if (!eq)
 439         {
 440             yaz_log(YLOG_WARN, "Missing equal-sign in filter");
 441             return 0;
 442         }
 443         *(eq++) = '\0';
 444         new->name = values[i];
 445         nmem_strsplit(m, "|", eq, &subvalues, &subnum);
 446         new->values = 0;
 447         for (subi = 0; subi < subnum; subi++)
 448         {
 449             struct database_criterion_value *newv
 450                 = nmem_malloc(m, sizeof(*newv));
 451             newv->value = subvalues[subi];
 452             newv->next = new->values;
 453             new->values = newv;
 454         }
 455         new->next = res;
 456         res = new;
 457     }
 458     return res;
 459 }
 460
 461 char *search(struct session *se, char *query, char *filter)
 462 {
 463     int live_channels = 0;
 464     struct client *cl;
 465     struct database_criterion *criteria;
 466
 467     yaz_log(YLOG_DEBUG, "Search");
 468
 469     nmem_reset(se->nmem);
 470     criteria = parse_filter(se->nmem, filter);
 471     se->requestid++;
 472     live_channels = select_targets(se, criteria);
 473     if (live_channels)
 474     {
 475         int maxrecs = live_channels * global_parameters.toget;
 476         se->num_termlists = 0;
 477         se->reclist = reclist_create(se->nmem, maxrecs);
 478         // This will be initialized in send_search()
 479         se->total_records = se->total_hits = se->total_merged = 0;
 480         se->expected_maxrecs = maxrecs;
 481     }
 482     else
 483         return "NOTARGETS";
 484
 485     se->relevance = 0;
 486
 487     for (cl = se->clients; cl; cl = client_next_in_session(cl))
 488     {
 489         if (prepare_session_database(se, client_get_database(cl)) < 0)
 490             return "CONFIG_ERROR";
 491         // Query must parse for all targets
 492         if (client_parse_query(cl, query) < 0)
 493             return "QUERY";
 494     }
 495
 496     for (cl = se->clients; cl; cl = client_next_in_session(cl))
 497     {
 498         client_prep_connection(cl);
 499     }
 500
 501     return 0;
 502 }
 503
 504 // Creates a new session_database object for a database
 505 static void session_init_databases_fun(void *context, struct database *db)
 506 {
 507     struct session *se = (struct session *) context;
 508     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
 509     int num = settings_num();
 510     int i;
 511
 512     new->database = db;
 513     new->yaz_marc = 0;
 514     new->pct = pp2_charset_create();
 515     new->map = 0;
 516     new->settings
 517         = nmem_malloc(se->session_nmem, sizeof(struct settings *) * num);
 518     memset(new->settings, 0, sizeof(struct settings*) * num);
 519
 520     if (db->settings)
 521     {
 522         for (i = 0; i < num; i++)
 523             new->settings[i] = db->settings[i];
 524     }
 525     new->next = se->databases;
 526     se->databases = new;
 527 }
 528
 529 // Doesn't free memory associated with sdb -- nmem takes care of that
 530 static void session_database_destroy(struct session_database *sdb)
 531 {
 532     struct database_retrievalmap *m;
 533
 534     for (m = sdb->map; m; m = m->next)
 535         xsltFreeStylesheet(m->stylesheet);
 536     if (sdb->yaz_marc)
 537         yaz_marc_destroy(sdb->yaz_marc);
 538     if (sdb->pct)
 539         pp2_charset_destroy(sdb->pct);
 540 }
 541
 542 // Initialize session_database list -- this represents this session's view
 543 // of the database list -- subject to modification by the settings ws command
 544 void session_init_databases(struct session *se)
 545 {
 546     se->databases = 0;
 547     grep_databases(se, 0, session_init_databases_fun);
 548 }
 549
 550 // Probably session_init_databases_fun should be refactored instead of
 551 // called here.
 552 static struct session_database *load_session_database(struct session *se,
 553                                                       char *id)
 554 {
 555     struct database *db = find_database(id, 0);
 556
 557     session_init_databases_fun((void*) se, db);
 558     // New sdb is head of se->databases list
 559     return se->databases;
 560 }
 561
 562 // Find an existing session database. If not found, load it
 563 static struct session_database *find_session_database(struct session *se,
 564                                                       char *id)
 565 {
 566     struct session_database *sdb;
 567
 568     for (sdb = se->databases; sdb; sdb = sdb->next)
 569         if (!strcmp(sdb->database->url, id))
 570             return sdb;
 571     return load_session_database(se, id);
 572 }
 573
 574 // Apply a session override to a database
 575 void session_apply_setting(struct session *se, char *dbname, char *setting,
 576                            char *value)
 577 {
 578     struct session_database *sdb = find_session_database(se, dbname);
 579     struct setting *new = nmem_malloc(se->session_nmem, sizeof(*new));
 580     int offset = settings_offset(setting);
 581
 582     if (offset < 0)
 583     {
 584         yaz_log(YLOG_WARN, "Unknown setting %s", setting);
 585         return;
 586     }
 587     new->precedence = 0;
 588     new->target = dbname;
 589     new->name = setting;
 590     new->value = value;
 591     new->next = sdb->settings[offset];
 592     sdb->settings[offset] = new;
 593
 594     // Force later recompute of settings-driven data structures
 595     // (happens when a search starts and client connections are prepared)
 596     switch (offset)
 597     {
 598         case PZ_NATIVESYNTAX:
 599             if (sdb->yaz_marc)
 600             {
 601                 yaz_marc_destroy(sdb->yaz_marc);
 602                 sdb->yaz_marc = 0;
 603             }
 604             break;
 605         case PZ_XSLT:
 606             if (sdb->map)
 607             {
 608                 struct database_retrievalmap *m;
 609                 // We don't worry about the map structure -- it's in nmem
 610                 for (m = sdb->map; m; m = m->next)
 611                     xsltFreeStylesheet(m->stylesheet);
 612                 sdb->map = 0;
 613             }
 614             break;
 615     }
 616 }
 617
 618 void destroy_session(struct session *s)
 619 {
 620     struct session_database *sdb;
 621
 622     yaz_log(YLOG_LOG, "Destroying session");
 623     while (s->clients)
 624         client_destroy(s->clients);
 625     for (sdb = s->databases; sdb; sdb = sdb->next)
 626         session_database_destroy(sdb);
 627     nmem_destroy(s->nmem);
 628     wrbuf_destroy(s->wrbuf);
 629 }
 630
 631 struct session *new_session(NMEM nmem)
 632 {
 633     int i;
 634     struct session *session = nmem_malloc(nmem, sizeof(*session));
 635
 636     yaz_log(YLOG_DEBUG, "New Pazpar2 session");
 637
 638     session->total_hits = 0;
 639     session->total_records = 0;
 640     session->num_termlists = 0;
 641     session->reclist = 0;
 642     session->requestid = -1;
 643     session->clients = 0;
 644     session->expected_maxrecs = 0;
 645     session->session_nmem = nmem;
 646     session->nmem = nmem_create();
 647     session->wrbuf = wrbuf_alloc();
 648     session_init_databases(session);
 649     for (i = 0; i <= SESSION_WATCH_MAX; i++)
 650     {
 651         session->watchlist[i].data = 0;
 652         session->watchlist[i].fun = 0;
 653     }
 654
 655     return session;
 656 }
 657
 658 struct hitsbytarget *hitsbytarget(struct session *se, int *count)
 659 {
 660     static struct hitsbytarget res[1000]; // FIXME MM
 661     struct client *cl;
 662
 663     *count = 0;
 664     for (cl = se->clients; cl; cl = client_next_in_session(cl))
 665     {
 666         char *name = session_setting_oneval(client_get_database(cl), PZ_NAME);
 667
 668         res[*count].id = client_get_database(cl)->database->url;
 669         res[*count].name = *name ? name : "Unknown";
 670         res[*count].hits = client_get_hits(cl);
 671         res[*count].records = client_get_num_records(cl);
 672         res[*count].diagnostic = client_get_diagnostic(cl);
 673         res[*count].state = client_get_state_str(cl);
 674         res[*count].connected  = client_get_connection(cl) ? 1 : 0;
 675         (*count)++;
 676     }
 677
 678     return res;
 679 }
 680
 681 struct termlist_score **termlist(struct session *s, const char *name, int *num)
 682 {
 683     int i;
 684
 685     for (i = 0; i < s->num_termlists; i++)
 686         if (!strcmp((const char *) s->termlists[i].name, name))
 687             return termlist_highscore(s->termlists[i].termlist, num);
 688     return 0;
 689 }
 690
 691 #ifdef MISSING_HEADERS
 692 void report_nmem_stats(void)
 693 {
 694     size_t in_use, is_free;
 695
 696     nmem_get_memory_in_use(&in_use);
 697     nmem_get_memory_free(&is_free);
 698
 699     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld",
 700             (long) in_use, (long) is_free);
 701 }
 702 #endif
 703
 704 struct record_cluster *show_single(struct session *s, int id)
 705 {
 706     struct record_cluster *r;
 707
 708     reclist_rewind(s->reclist);
 709     while ((r = reclist_read_record(s->reclist)))
 710         if (r->recid == id)
 711             return r;
 712     return 0;
 713 }
 714
 715 struct record_cluster **show(struct session *s, struct reclist_sortparms *sp,
 716                              int start, int *num, int *total, int *sumhits,
 717                              NMEM nmem_show)
 718 {
 719     struct record_cluster **recs = nmem_malloc(nmem_show, *num
 720                                        * sizeof(struct record_cluster *));
 721     struct reclist_sortparms *spp;
 722     int i;
 723 #if USE_TIMING
 724     yaz_timing_t t = yaz_timing_create();
 725 #endif
 726
 727     for (spp = sp; spp; spp = spp->next)
 728         if (spp->type == Metadata_sortkey_relevance)
 729         {
 730             relevance_prepare_read(s->relevance, s->reclist);
 731             break;
 732         }
 733     reclist_sort(s->reclist, sp);
 734
 735     *total = s->reclist->num_records;
 736     *sumhits = s->total_hits;
 737
 738     for (i = 0; i < start; i++)
 739         if (!reclist_read_record(s->reclist))
 740         {
 741             *num = 0;
 742             recs = 0;
 743             break;
 744         }
 745
 746     for (i = 0; i < *num; i++)
 747     {
 748         struct record_cluster *r = reclist_read_record(s->reclist);
 749         if (!r)
 750         {
 751             *num = i;
 752             break;
 753         }
 754         recs[i] = r;
 755     }
 756 #if USE_TIMING
 757     yaz_timing_stop(t);
 758     yaz_log(YLOG_LOG, "show %6.5f %3.2f %3.2f",
 759             yaz_timing_get_real(t), yaz_timing_get_user(t),
 760             yaz_timing_get_sys(t));
 761     yaz_timing_destroy(&t);
 762 #endif
 763     return recs;
 764 }
 765
 766 void statistics(struct session *se, struct statistics *stat)
 767 {
 768     struct client *cl;
 769     int count = 0;
 770
 771     memset(stat, 0, sizeof(*stat));
 772     for (cl = se->clients; cl; cl = client_next_in_session(cl))
 773     {
 774         if (!client_get_connection(cl))
 775             stat->num_no_connection++;
 776         switch (client_get_state(cl))
 777         {
 778             case Client_Connecting: stat->num_connecting++; break;
 779             case Client_Initializing: stat->num_initializing++; break;
 780             case Client_Searching: stat->num_searching++; break;
 781             case Client_Presenting: stat->num_presenting++; break;
 782             case Client_Idle: stat->num_idle++; break;
 783             case Client_Failed: stat->num_failed++; break;
 784             case Client_Error: stat->num_error++; break;
 785             default: break;
 786         }
 787         count++;
 788     }
 789     stat->num_hits = se->total_hits;
 790     stat->num_records = se->total_records;
 791
 792     stat->num_clients = count;
 793 }
 794
 795 void start_http_listener(void)
 796 {
 797     char hp[128] = "";
 798     struct conf_server *ser = global_parameters.server;
 799
 800     if (*global_parameters.listener_override)
 801         strcpy(hp, global_parameters.listener_override);
 802     else
 803     {
 804         strcpy(hp, ser->host ? ser->host : "");
 805         if (ser->port)
 806         {
 807             if (*hp)
 808                 strcat(hp, ":");
 809             sprintf(hp + strlen(hp), "%d", ser->port);
 810         }
 811     }
 812     http_init(hp);
 813 }
 814
 815 void start_proxy(void)
 816 {
 817     char hp[128] = "";
 818     struct conf_server *ser = global_parameters.server;
 819
 820     if (*global_parameters.proxy_override)
 821         strcpy(hp, global_parameters.proxy_override);
 822     else if (ser->proxy_host || ser->proxy_port)
 823     {
 824         strcpy(hp, ser->proxy_host ? ser->proxy_host : "");
 825         if (ser->proxy_port)
 826         {
 827             if (*hp)
 828                 strcat(hp, ":");
 829             sprintf(hp + strlen(hp), "%d", ser->proxy_port);
 830         }
 831     }
 832     else
 833         return;
 834
 835     http_set_proxyaddr(hp, ser->myurl ? ser->myurl : "");
 836 }
 837
 838 void start_zproxy(void)
 839 {
 840     struct conf_server *ser = global_parameters.server;
 841
 842     if (*global_parameters.zproxy_override){
 843         yaz_log(YLOG_LOG, "Z39.50 proxy  %s",
 844                 global_parameters.zproxy_override);
 845         return;
 846     }
 847
 848     else if (ser->zproxy_host || ser->zproxy_port)
 849     {
 850         char hp[128] = "";
 851
 852         strcpy(hp, ser->zproxy_host ? ser->zproxy_host : "");
 853         if (ser->zproxy_port)
 854         {
 855             if (*hp)
 856                 strcat(hp, ":");
 857             else
 858                 strcat(hp, "@:");
 859
 860             sprintf(hp + strlen(hp), "%d", ser->zproxy_port);
 861         }
 862         strcpy(global_parameters.zproxy_override, hp);
 863         yaz_log(YLOG_LOG, "Z39.50 proxy  %s",
 864                 global_parameters.zproxy_override);
 865
 866     }
 867     else
 868         return;
 869 }
 870
 871 // Master list of connections we're handling events to
 872 static IOCHAN channel_list = 0;
 873 void pazpar2_add_channel(IOCHAN chan)
 874 {
 875     chan->next = channel_list;
 876     channel_list = chan;
 877 }
 878
 879 void pazpar2_event_loop()
 880 {
 881     event_loop(&channel_list);
 882 }
 883
 884 struct record *ingest_record(struct client *cl, Z_External *rec,
 885                              int record_no)
 886 {
 887     xmlDoc *xdoc = normalize_record(client_get_database(cl), rec);
 888     xmlNode *root, *n;
 889     struct record *record;
 890     struct record_cluster *cluster;
 891     struct session *se = client_get_session(cl);
 892     xmlChar *mergekey, *mergekey_norm;
 893     xmlChar *type = 0;
 894     xmlChar *value = 0;
 895     struct conf_service *service = global_parameters.server->service;
 896
 897     if (!xdoc)
 898         return 0;
 899
 900     root = xmlDocGetRootElement(xdoc);
 901     if (!(mergekey = xmlGetProp(root, (xmlChar *) "mergekey")))
 902     {
 903         yaz_log(YLOG_WARN, "No mergekey found in record");
 904         xmlFreeDoc(xdoc);
 905         return 0;
 906     }
 907
 908 #if 0
 909     record = nmem_malloc(se->nmem, sizeof(struct record));
 910     record->next = 0;
 911     record->client = cl;
 912     record->metadata = nmem_malloc(se->nmem,
 913             sizeof(struct record_metadata*) * service->num_metadata);
 914     memset(record->metadata, 0,
 915            sizeof(struct record_metadata*) * service->num_metadata);
 916
 917 #else
 918     record = record_create(se->nmem,
 919                            service->num_metadata, service->num_sortkeys);
 920     record_assign_client(record, cl);
 921 #endif
 922
 923     mergekey_norm = (xmlChar *) nmem_strdup(se->nmem, (char*) mergekey);
 924     xmlFree(mergekey);
 925     normalize7bit_mergekey((char *) mergekey_norm, 0);
 926
 927     cluster = reclist_insert(se->reclist,
 928                              global_parameters.server->service,
 929                              record, (char *) mergekey_norm,
 930                              &se->total_merged);
 931     if (global_parameters.dump_records)
 932         yaz_log(YLOG_LOG, "Cluster id %d from %s (#%d)", cluster->recid,
 933                 client_get_database(cl)->database->url, record_no);
 934     if (!cluster)
 935     {
 936         /* no room for record */
 937         xmlFreeDoc(xdoc);
 938         return 0;
 939     }
 940     relevance_newrec(se->relevance, cluster);
 941
 942
 943     // now parsing XML record and adding data to cluster or record metadata
 944     for (n = root->children; n; n = n->next)
 945     {
 946         if (type)
 947             xmlFree(type);
 948         if (value)
 949             xmlFree(value);
 950         type = value = 0;
 951
 952         if (n->type != XML_ELEMENT_NODE)
 953             continue;
 954         if (!strcmp((const char *) n->name, "metadata"))
 955         {
 956             struct conf_metadata *ser_md = 0;
 957             struct conf_sortkey *ser_sk = 0;
 958             struct record_metadata **wheretoput = 0;
 959             struct record_metadata *rec_md = 0;
 960             int md_field_id = -1;
 961             int sk_field_id = -1;
 962             int first, last;
 963
 964             type = xmlGetProp(n, (xmlChar *) "type");
 965             value = xmlNodeListGetString(xdoc, n->children, 0);
 966
 967             if (!type || !value)
 968                 continue;
 969
 970 #if 0
 971             // First, find out what field we're looking at
 972             for (md_field_id = 0; md_field_id < service->num_metadata;
 973                  md_field_id++)
 974                 if (!strcmp((const char *) type,
 975                             service->metadata[md_field_id].name))
 976                 {
 977                     ser_md = &service->metadata[md_field_id];
 978                     if (ser_md->sortkey_offset >= 0){
 979                         sk_field_id = ser_md->sortkey_offset;
 980                         ser_sk = &service->sortkeys[sk_field_id];
 981                     }
 982                     break;
 983                 }
 984
 985             if (!ser_md)
 986             {
 987                 yaz_log(YLOG_WARN,
 988                         "Ignoring unknown metadata element: %s", type);
 989                 continue;
 990             }
 991
 992 #else
 993             md_field_id
 994                 = conf_service_metadata_field_id(service, (const char *) type);
 995             if (md_field_id < 0)
 996             {
 997                 yaz_log(YLOG_WARN,
 998                         "Ignoring unknown metadata element: %s", type);
 999                 continue;
1000             }
1001
1002             ser_md = &service->metadata[md_field_id];
1003
1004             if (ser_md->sortkey_offset >= 0){
1005                 sk_field_id = ser_md->sortkey_offset;
1006                 ser_sk = &service->sortkeys[sk_field_id];
1007             }
1008 #endif
1009
1010             // Find out where we are putting it - based on merge or not
1011             if (ser_md->merge == Metadata_merge_no)
1012                 wheretoput = &record->metadata[md_field_id];
1013             else
1014                 wheretoput = &cluster->metadata[md_field_id];
1015
1016             // create new record_metadata
1017 #if 0
1018             rec_md = nmem_malloc(se->nmem, sizeof(struct record_metadata));
1019             rec_md->next = 0;
1020 #else
1021             rec_md = record_metadata_create(se->nmem);
1022 #endif
1023
1024             // and polulate with data:
1025             // type based charmapping decisions follow here
1026             if (ser_md->type == Metadata_type_generic)
1027             {
1028
1029 #if 0
1030                 char *p, *pe;
1031                 for (p = (char *) value; *p && isspace(*p); p++)
1032                     ;
1033                 for (pe = p + strlen(p) - 1;
1034                         pe > p && strchr(" ,/.:([", *pe); pe--)
1035                     *pe = '\0';
1036 #else
1037                 char * p = (char *) value;
1038                 p = normalize7bit_generic(p, " ,/.:([");
1039 #endif
1040
1041                 rec_md->data.text = nmem_strdup(se->nmem, p);
1042
1043             }
1044             else if (ser_md->type == Metadata_type_year)
1045             {
1046                 if (extract7bit_years((char *) value, &first, &last) < 0)
1047                     continue;
1048             }
1049             else
1050             {
1051                 yaz_log(YLOG_WARN,
1052                         "Unknown type in metadata element %s", type);
1053                 continue;
1054             }
1055
1056 #if 0  // this test does not belong here.
1057        // the condition is enforced in the constructor
1058        // inside conf_metadata_assign()
1059        // and will _never_ occur
1060             if (ser_md->type == Metadata_type_year
1061                 && ser_md->merge != Metadata_merge_range)
1062             {
1063                 yaz_log(YLOG_WARN, "Only range merging supported for years");
1064                 continue;
1065             }
1066 # endif
1067
1068
1069             // and polulate with data:
1070             // assign cluster or record based on merge action
1071             if (ser_md->merge == Metadata_merge_unique)
1072             {
1073                 struct record_metadata *mnode;
1074                 for (mnode = *wheretoput; mnode; mnode = mnode->next)
1075                     if (!strcmp((const char *) mnode->data.text,
1076                                 rec_md->data.text))
1077                         break;
1078                 if (!mnode)
1079                 {
1080                     rec_md->next = *wheretoput;
1081                     *wheretoput = rec_md;
1082                 }
1083             }
1084             else if (ser_md->merge == Metadata_merge_longest)
1085             {
1086                 if (!*wheretoput
1087                     || strlen(rec_md->data.text)
1088                        > strlen((*wheretoput)->data.text))
1089                 {
1090                     *wheretoput = rec_md;
1091                     if (ser_sk)
1092                     {
1093                         char *s = nmem_strdup(se->nmem, rec_md->data.text);
1094                         if (!cluster->sortkeys[sk_field_id])
1095                             cluster->sortkeys[sk_field_id] =
1096                                 nmem_malloc(se->nmem,
1097                                             sizeof(union data_types));
1098                         normalize7bit_mergekey(s,
1099                              (ser_sk->type == Metadata_sortkey_skiparticle));
1100                         cluster->sortkeys[sk_field_id]->text = s;
1101                     }
1102                 }
1103             }
1104             else if (ser_md->merge == Metadata_merge_all
1105                      || ser_md->merge == Metadata_merge_no)
1106             {
1107                 rec_md->next = *wheretoput;
1108                 *wheretoput = rec_md;
1109             }
1110             else if (ser_md->merge == Metadata_merge_range)
1111             {
1112
1113 #if 0           // this assert does not belong here.
1114                 // the condition is enforced in the constructor
1115                 // inside conf_metadata_assign()
1116                 // and will _never_ occur
1117                 assert(ser_md->type == Metadata_type_year);
1118 #endif
1119
1120                 if (!*wheretoput)
1121                 {
1122                     *wheretoput = rec_md;
1123                     (*wheretoput)->data.number.min = first;
1124                     (*wheretoput)->data.number.max = last;
1125                     if (ser_sk)
1126                         cluster->sortkeys[sk_field_id]
1127                             = &rec_md->data;
1128                 }
1129                 else
1130                 {
1131                     if (first < (*wheretoput)->data.number.min)
1132                         (*wheretoput)->data.number.min = first;
1133                     if (last > (*wheretoput)->data.number.max)
1134                         (*wheretoput)->data.number.max = last;
1135                 }
1136 #ifdef GAGA
1137                 if (ser_sk)
1138                 {
1139                     union data_types *sdata
1140                         = cluster->sortkeys[sk_field_id];
1141                     yaz_log(YLOG_LOG, "SK range: %d-%d",
1142                             sdata->number.min, sdata->number.max);
1143                 }
1144 #endif
1145             }
1146
1147 #if 0      // this else is only entered when Metadata_merge_no
1148            // but I believe then we should _not_ pollute with log messages
1149             else
1150
1151                 yaz_log(YLOG_WARN,
1152                         "Don't know how to merge on element name %s",
1153                         ser_md->name);
1154 #endif
1155
1156
1157             // ranking of _all_ fields enabled ...
1158             if (ser_md->rank)
1159                 relevance_countwords(se->relevance, cluster,
1160                                      (char *) value, ser_md->rank);
1161
1162             // construct facets ...
1163             if (ser_md->termlist)
1164             {
1165                 if (ser_md->type == Metadata_type_year)
1166                 {
1167                     char year[64];
1168                     sprintf(year, "%d", last);
1169                     add_facet(se, (char *) type, year);
1170                     if (first != last)
1171                     {
1172                         sprintf(year, "%d", first);
1173                         add_facet(se, (char *) type, year);
1174                     }
1175                 }
1176                 else
1177                     add_facet(se, (char *) type, (char *) value);
1178             }
1179
1180             // cleaning up
1181             xmlFree(type);
1182             xmlFree(value);
1183             type = value = 0;
1184         }
1185         else
1186             yaz_log(YLOG_WARN,
1187                     "Unexpected element %s in internal record", n->name);
1188     }
1189     if (type)
1190         xmlFree(type);
1191     if (value)
1192         xmlFree(value);
1193
1194     xmlFreeDoc(xdoc);
1195
1196     relevance_donerecord(se->relevance, cluster);
1197     se->total_records++;
1198
1199     return record;
1200 }
1201
1202
1203
1204 /*
1205  * Local variables:
1206  * c-basic-offset: 4
1207  * indent-tabs-mode: nil
1208  * End:
1209  * vim: shiftwidth=4 tabstop=8 expandtab
1210  */