src/pazpar2.c

   1 /* $Id: pazpar2.c,v 1.78 2007-04-13 11:13:08 marc 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 #include <stdlib.h>
  23 #include <stdio.h>
  24 #include <string.h>
  25 #include <sys/time.h>
  26 #include <unistd.h>
  27 #include <sys/socket.h>
  28 #include <netdb.h>
  29 #include <signal.h>
  30 #include <ctype.h>
  31 #include <assert.h>
  32
  33 #include <yaz/marcdisp.h>
  34 #include <yaz/comstack.h>
  35 #include <yaz/tcpip.h>
  36 #include <yaz/proto.h>
  37 #include <yaz/readconf.h>
  38 #include <yaz/pquery.h>
  39 #include <yaz/otherinfo.h>
  40 #include <yaz/yaz-util.h>
  41 #include <yaz/nmem.h>
  42 #include <yaz/query-charset.h>
  43 #include <yaz/querytowrbuf.h>
  44 #if YAZ_VERSIONL >= 0x020163
  45 #include <yaz/oid_db.h>
  46 #endif
  47
  48 #if HAVE_CONFIG_H
  49 #include "cconfig.h"
  50 #endif
  51
  52 #define USE_TIMING 0
  53 #if USE_TIMING
  54 #include <yaz/timing.h>
  55 #endif
  56
  57 #include <netinet/in.h>
  58
  59 #include "pazpar2.h"
  60 #include "eventl.h"
  61 #include "http.h"
  62 #include "termlists.h"
  63 #include "reclists.h"
  64 #include "relevance.h"
  65 #include "config.h"
  66 #include "database.h"
  67 #include "settings.h"
  68
  69 #define MAX_CHUNK 15
  70
  71 static void client_fatal(struct client *cl);
  72 static void connection_destroy(struct connection *co);
  73 static int client_prep_connection(struct client *cl);
  74 static void ingest_records(struct client *cl, Z_Records *r);
  75 void session_alert_watch(struct session *s, int what);
  76
  77 IOCHAN channel_list = 0;  // Master list of connections we're handling events to
  78
  79 static struct connection *connection_freelist = 0;
  80 static struct client *client_freelist = 0;
  81
  82 static char *client_states[] = {
  83     "Client_Connecting",
  84     "Client_Connected",
  85     "Client_Idle",
  86     "Client_Initializing",
  87     "Client_Searching",
  88     "Client_Presenting",
  89     "Client_Error",
  90     "Client_Failed",
  91     "Client_Disconnected",
  92     "Client_Stopped"
  93 };
  94
  95 // Note: Some things in this structure will eventually move to configuration
  96 struct parameters global_parameters =
  97 {
  98     "",
  99     "",
 100     "",
 101     "",
 102     0,
 103     0,
 104     30,
 105     "81",
 106     "Index Data PazPar2 (MasterKey)",
 107     VERSION,
 108     600, // 10 minutes
 109     60,
 110     100,
 111     MAX_CHUNK,
 112     0,
 113     0
 114 };
 115
 116 static int send_apdu(struct client *c, Z_APDU *a)
 117 {
 118     struct connection *co = c->connection;
 119     char *buf;
 120     int len, r;
 121
 122     if (!z_APDU(global_parameters.odr_out, &a, 0, 0))
 123     {
 124         odr_perror(global_parameters.odr_out, "Encoding APDU");
 125         abort();
 126     }
 127     buf = odr_getbuf(global_parameters.odr_out, &len, 0);
 128     r = cs_put(co->link, buf, len);
 129     if (r < 0)
 130     {
 131         yaz_log(YLOG_WARN, "cs_put: %s", cs_errmsg(cs_errno(co->link)));
 132         return -1;
 133     }
 134     else if (r == 1)
 135     {
 136         fprintf(stderr, "cs_put incomplete (ParaZ does not handle that)\n");
 137         exit(1);
 138     }
 139     odr_reset(global_parameters.odr_out); /* release the APDU structure  */
 140     co->state = Conn_Waiting;
 141     return 0;
 142 }
 143
 144 // Set authentication token in init if one is set for the client
 145 // TODO: Extend this to handle other schemes than open (should be simple)
 146 static void init_authentication(struct client *cl, Z_InitRequest *req)
 147 {
 148     struct session_database *sdb = cl->database;
 149     char *auth = session_setting_oneval(sdb, PZ_AUTHENTICATION);
 150
 151     if (auth)
 152     {
 153         Z_IdAuthentication *idAuth = odr_malloc(global_parameters.odr_out,
 154                 sizeof(*idAuth));
 155         idAuth->which = Z_IdAuthentication_open;
 156         idAuth->u.open = auth;
 157         req->idAuthentication = idAuth;
 158     }
 159 }
 160
 161 static void send_init(IOCHAN i)
 162 {
 163
 164     struct connection *co = iochan_getdata(i);
 165     struct client *cl = co->client;
 166     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_initRequest);
 167
 168     a->u.initRequest->implementationId = global_parameters.implementationId;
 169     a->u.initRequest->implementationName = global_parameters.implementationName;
 170     a->u.initRequest->implementationVersion =
 171         global_parameters.implementationVersion;
 172     ODR_MASK_SET(a->u.initRequest->options, Z_Options_search);
 173     ODR_MASK_SET(a->u.initRequest->options, Z_Options_present);
 174     ODR_MASK_SET(a->u.initRequest->options, Z_Options_namedResultSets);
 175
 176     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_1);
 177     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_2);
 178     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_3);
 179
 180     init_authentication(cl, a->u.initRequest);
 181
 182     /* add virtual host if tunneling through Z39.50 proxy */
 183
 184     if (0 < strlen(global_parameters.zproxy_override)
 185         && 0 < strlen(cl->database->database->url))
 186     {
 187 #if YAZ_VERSIONL >= 0x020163
 188         const int *oid_proxy = yaz_string_to_oid(yaz_oid_std(),
 189                                                  CLASS_USERINFO, OID_STR_PROXY);
 190         yaz_oi_set_string_oid(&a->u.initRequest->otherInfo,
 191                               global_parameters.odr_out, oid_proxy,
 192                               1, cl->database->database->url);
 193 #else
 194         yaz_oi_set_string_oidval(&a->u.initRequest->otherInfo,
 195                                  global_parameters.odr_out, VAL_PROXY,
 196                                  1, cl->database->database->url);
 197 #endif
 198     }
 199
 200     if (send_apdu(cl, a) >= 0)
 201     {
 202         iochan_setflags(i, EVENT_INPUT);
 203         cl->state = Client_Initializing;
 204     }
 205     else
 206         cl->state = Client_Error;
 207     odr_reset(global_parameters.odr_out);
 208 }
 209
 210 static void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 211 {
 212     switch (n->kind)
 213     {
 214         case CCL_RPN_AND:
 215         case CCL_RPN_OR:
 216         case CCL_RPN_NOT:
 217         case CCL_RPN_PROX:
 218             pull_terms(nmem, n->u.p[0], termlist, num);
 219             pull_terms(nmem, n->u.p[1], termlist, num);
 220             break;
 221         case CCL_RPN_TERM:
 222             termlist[(*num)++] = nmem_strdup(nmem, n->u.t.term);
 223             break;
 224         default: // NOOP
 225             break;
 226     }
 227 }
 228
 229 // Extract terms from query into null-terminated termlist
 230 static void extract_terms(NMEM nmem, struct ccl_rpn_node *query, char **termlist)
 231 {
 232     int num = 0;
 233
 234     pull_terms(nmem, query, termlist, &num);
 235     termlist[num] = 0;
 236 }
 237
 238 static void send_search(IOCHAN i)
 239 {
 240     struct connection *co = iochan_getdata(i);
 241     struct client *cl = co->client;
 242     struct session *se = cl->session;
 243     struct session_database *sdb = cl->database;
 244     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_searchRequest);
 245     int ndb, cerror, cpos;
 246     char **databaselist;
 247     Z_Query *zquery;
 248     struct ccl_rpn_node *cn;
 249     int ssub = 0, lslb = 100000, mspn = 10;
 250     char *recsyn = 0;
 251     char *piggyback = 0;
 252     char *queryenc = 0;
 253     yaz_iconv_t iconv = 0;
 254
 255     yaz_log(YLOG_DEBUG, "Sending search to %s", cl->database->database->url);
 256
 257     cn = ccl_find_str(sdb->database->ccl_map, se->query, &cerror, &cpos);
 258     if (!cn)
 259         return;
 260
 261     if (!se->relevance)
 262     {
 263         // Initialize relevance structure with query terms
 264         char *p[512];
 265         extract_terms(se->nmem, cn, p);
 266         se->relevance = relevance_create(se->nmem, (const char **) p,
 267                 se->expected_maxrecs);
 268     }
 269
 270     // constructing RPN query
 271     a->u.searchRequest->query = zquery = odr_malloc(global_parameters.odr_out,
 272             sizeof(Z_Query));
 273     zquery->which = Z_Query_type_1;
 274     zquery->u.type_1 = ccl_rpn_query(global_parameters.odr_out, cn);
 275     ccl_rpn_delete(cn);
 276
 277     // converting to target encoding
 278     if ((queryenc = session_setting_oneval(sdb, PZ_QUERYENCODING))){
 279         iconv = yaz_iconv_open(queryenc, "UTF-8");
 280         if (iconv){
 281             yaz_query_charset_convert_rpnquery(zquery->u.type_1,
 282                                                global_parameters.odr_out,
 283                                                iconv);
 284             yaz_iconv_close(iconv);
 285         } else
 286             yaz_log(YLOG_WARN, "Query encoding failed %s %s",
 287                     cl->database->database->url, queryenc);
 288     }
 289
 290     for (ndb = 0; sdb->database->databases[ndb]; ndb++)
 291         ;
 292     databaselist = odr_malloc(global_parameters.odr_out, sizeof(char*) * ndb);
 293     for (ndb = 0; sdb->database->databases[ndb]; ndb++)
 294         databaselist[ndb] = sdb->database->databases[ndb];
 295
 296     if (!(piggyback = session_setting_oneval(sdb, PZ_PIGGYBACK)) || *piggyback == '1')
 297     {
 298         if ((recsyn = session_setting_oneval(sdb, PZ_REQUESTSYNTAX)))
 299         {
 300 #if YAZ_VERSIONL >= 0x020163
 301             a->u.searchRequest->preferredRecordSyntax =
 302                 yaz_string_to_oid_odr(yaz_oid_std(),
 303                                       CLASS_RECSYN, recsyn,
 304                                       global_parameters.odr_out);
 305 #else
 306             a->u.searchRequest->preferredRecordSyntax =
 307                 yaz_str_to_z3950oid(global_parameters.odr_out,
 308                                     CLASS_RECSYN, recsyn);
 309 #endif
 310         }
 311         a->u.searchRequest->smallSetUpperBound = &ssub;
 312         a->u.searchRequest->largeSetLowerBound = &lslb;
 313         a->u.searchRequest->mediumSetPresentNumber = &mspn;
 314     }
 315     a->u.searchRequest->resultSetName = "Default";
 316     a->u.searchRequest->databaseNames = databaselist;
 317     a->u.searchRequest->num_databaseNames = ndb;
 318
 319
 320     {  //scope for sending and logging queries
 321         WRBUF wbquery = wrbuf_alloc();
 322         yaz_query_to_wrbuf(wbquery, zquery);
 323
 324
 325         if (send_apdu(cl, a) >= 0)
 326             {
 327                 iochan_setflags(i, EVENT_INPUT);
 328                 cl->state = Client_Searching;
 329                 cl->requestid = se->requestid;
 330                 yaz_log(YLOG_LOG, "SearchRequest %s %s %s",
 331                          cl->database->database->url,
 332                         queryenc ? queryenc : "UTF-8",
 333                         wrbuf_cstr(wbquery));
 334             }
 335         else {
 336             cl->state = Client_Error;
 337                 yaz_log(YLOG_WARN, "Failed SearchRequest %s  %s %s",
 338                          cl->database->database->url,
 339                         queryenc ? queryenc : "UTF-8",
 340                         wrbuf_cstr(wbquery));
 341         }
 342
 343         wrbuf_destroy(wbquery);
 344     }
 345
 346     odr_reset(global_parameters.odr_out);
 347 }
 348
 349 static void send_present(IOCHAN i)
 350 {
 351     struct connection *co = iochan_getdata(i);
 352     struct client *cl = co->client;
 353     struct session_database *sdb = cl->database;
 354     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_presentRequest);
 355     int toget;
 356     int start = cl->records + 1;
 357     char *recsyn;
 358
 359     toget = global_parameters.chunk;
 360     if (toget > global_parameters.toget - cl->records)
 361         toget = global_parameters.toget - cl->records;
 362     if (toget > cl->hits - cl->records)
 363         toget = cl->hits - cl->records;
 364
 365     yaz_log(YLOG_DEBUG, "Trying to present %d records\n", toget);
 366
 367     a->u.presentRequest->resultSetStartPoint = &start;
 368     a->u.presentRequest->numberOfRecordsRequested = &toget;
 369
 370     a->u.presentRequest->resultSetId = "Default";
 371
 372     if ((recsyn = session_setting_oneval(sdb, PZ_REQUESTSYNTAX)))
 373     {
 374 #if YAZ_VERSIONL >= 0x020163
 375         a->u.presentRequest->preferredRecordSyntax =
 376             yaz_string_to_oid_odr(yaz_oid_std(),
 377                                   CLASS_RECSYN, recsyn,
 378                                   global_parameters.odr_out);
 379 #else
 380         a->u.presentRequest->preferredRecordSyntax =
 381             yaz_str_to_z3950oid(global_parameters.odr_out,
 382                                 CLASS_RECSYN, recsyn);
 383 #endif
 384     }
 385
 386     if (send_apdu(cl, a) >= 0)
 387     {
 388         iochan_setflags(i, EVENT_INPUT);
 389         cl->state = Client_Presenting;
 390     }
 391     else
 392         cl->state = Client_Error;
 393     odr_reset(global_parameters.odr_out);
 394 }
 395
 396 static void do_initResponse(IOCHAN i, Z_APDU *a)
 397 {
 398     struct connection *co = iochan_getdata(i);
 399     struct client *cl = co->client;
 400     Z_InitResponse *r = a->u.initResponse;
 401
 402     yaz_log(YLOG_DEBUG, "Init response %s", cl->database->database->url);
 403
 404     if (*r->result)
 405     {
 406         cl->state = Client_Idle;
 407     }
 408     else
 409         cl->state = Client_Failed; // FIXME need to do something to the connection
 410 }
 411
 412 static void do_searchResponse(IOCHAN i, Z_APDU *a)
 413 {
 414     struct connection *co = iochan_getdata(i);
 415     struct client *cl = co->client;
 416     struct session *se = cl->session;
 417     Z_SearchResponse *r = a->u.searchResponse;
 418
 419     yaz_log(YLOG_DEBUG, "Search response %s (status=%d)",
 420             cl->database->database->url, *r->searchStatus);
 421
 422     if (*r->searchStatus)
 423     {
 424         cl->hits = *r->resultCount;
 425         se->total_hits += cl->hits;
 426         if (r->presentStatus && !*r->presentStatus && r->records)
 427         {
 428             yaz_log(YLOG_DEBUG, "Records in search response %s",
 429                     cl->database->database->url);
 430             ingest_records(cl, r->records);
 431         }
 432         cl->state = Client_Idle;
 433     }
 434     else
 435     {          /*"FAILED"*/
 436         cl->hits = 0;
 437         cl->state = Client_Error;
 438         if (r->records) {
 439             Z_Records *recs = r->records;
 440             if (recs->which == Z_Records_NSD)
 441             {
 442                 yaz_log(YLOG_WARN,
 443                         "Search response: Non-surrogate diagnostic %s",
 444                         cl->database->database->url);
 445                 cl->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 446                 cl->state = Client_Error;
 447             }
 448         }
 449     }
 450 }
 451
 452 static void do_closeResponse(IOCHAN i, Z_APDU *a)
 453 {
 454     struct connection *co = iochan_getdata(i);
 455     struct client *cl = co->client;
 456     /* Z_Close *r = a->u.close; */
 457
 458     yaz_log(YLOG_WARN, "Close response %s", cl->database->database->url);
 459
 460     cl->state = Client_Failed;
 461     connection_destroy(co);
 462 }
 463
 464
 465 char *normalize_mergekey(char *buf, int skiparticle)
 466 {
 467     char *p = buf, *pout = buf;
 468
 469     if (skiparticle)
 470     {
 471         char firstword[64];
 472         char articles[] = "the den der die des an a "; // must end in space
 473
 474         while (*p && !isalnum(*p))
 475             p++;
 476         pout = firstword;
 477         while (*p && *p != ' ' && pout - firstword < 62)
 478             *(pout++) = tolower(*(p++));
 479         *(pout++) = ' ';
 480         *(pout++) = '\0';
 481         if (!strstr(articles, firstword))
 482             p = buf;
 483         pout = buf;
 484     }
 485
 486     while (*p)
 487     {
 488         while (*p && !isalnum(*p))
 489             p++;
 490         while (isalnum(*p))
 491             *(pout++) = tolower(*(p++));
 492         if (*p)
 493             *(pout++) = ' ';
 494         while (*p && !isalnum(*p))
 495             p++;
 496     }
 497     if (buf != pout)
 498         do {
 499             *(pout--) = '\0';
 500         }
 501         while (pout > buf && *pout == ' ');
 502
 503     return buf;
 504 }
 505
 506 static void add_facet(struct session *s, const char *type, const char *value)
 507 {
 508     int i;
 509
 510     if (!*value)
 511         return;
 512     for (i = 0; i < s->num_termlists; i++)
 513         if (!strcmp(s->termlists[i].name, type))
 514             break;
 515     if (i == s->num_termlists)
 516     {
 517         if (i == SESSION_MAX_TERMLISTS)
 518         {
 519             yaz_log(YLOG_FATAL, "Too many termlists");
 520             exit(1);
 521         }
 522         s->termlists[i].name = nmem_strdup(s->nmem, type);
 523         s->termlists[i].termlist = termlist_create(s->nmem, s->expected_maxrecs, 15);
 524         s->num_termlists = i + 1;
 525     }
 526     termlist_insert(s->termlists[i].termlist, value);
 527 }
 528
 529 static xmlDoc *normalize_record(struct client *cl, Z_External *rec)
 530 {
 531     struct database_retrievalmap *m;
 532     struct database *db = cl->database->database;
 533     xmlNode *res;
 534     xmlDoc *rdoc;
 535
 536     // First normalize to XML
 537     if (db->yaz_marc)
 538     {
 539         char *buf;
 540         int len;
 541         if (rec->which != Z_External_octet)
 542         {
 543             yaz_log(YLOG_WARN, "Unexpected external branch, probably BER %s",
 544                     cl->database->database->url);
 545             return 0;
 546         }
 547         buf = (char*) rec->u.octet_aligned->buf;
 548         len = rec->u.octet_aligned->len;
 549         if (yaz_marc_read_iso2709(db->yaz_marc, buf, len) < 0)
 550         {
 551             yaz_log(YLOG_WARN, "Failed to decode MARC %s",
 552                     cl->database->database->url);
 553             return 0;
 554         }
 555
 556         yaz_marc_write_using_libxml2(db->yaz_marc, 1);
 557         if (yaz_marc_write_xml(db->yaz_marc, &res,
 558                     "http://www.loc.gov/MARC21/slim", 0, 0) < 0)
 559         {
 560             yaz_log(YLOG_WARN, "Failed to encode as XML %s",
 561                     cl->database->database->url);
 562             return 0;
 563         }
 564         rdoc = xmlNewDoc((xmlChar *) "1.0");
 565         xmlDocSetRootElement(rdoc, res);
 566
 567     }
 568     else
 569     {
 570         yaz_log(YLOG_FATAL,
 571                 "Unknown native_syntax in normalize_record from %s",
 572                 cl->database->database->url);
 573         exit(1);
 574     }
 575
 576     if (global_parameters.dump_records){
 577         fprintf(stderr,
 578                 "Input Record (normalized) from %s\n----------------\n",
 579                 cl->database->database->url);
 580 #if LIBXML_VERSION >= 20600
 581         xmlDocFormatDump(stderr, rdoc, 1);
 582 #else
 583         xmlDocDump(stderr, rdoc);
 584 #endif
 585     }
 586
 587     for (m = db->map; m; m = m->next){
 588         xmlDoc *new = 0;
 589
 590 #if 1
 591         {
 592             xmlNodePtr root = 0;
 593             new = xsltApplyStylesheet(m->stylesheet, rdoc, 0);
 594             root= xmlDocGetRootElement(new);
 595         if (!new || !root || !(root->children))
 596         {
 597             yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
 598                     cl->database->database->url);
 599             xmlFreeDoc(new);
 600             xmlFreeDoc(rdoc);
 601             return 0;
 602         }
 603         }
 604 #endif
 605
 606 #if 0
 607         // do it another way to detect transformation errors right now
 608         // but does not seem to work either!
 609         {
 610             xsltTransformContextPtr ctxt;
 611             ctxt = xsltNewTransformContext(m->stylesheet, rdoc);
 612             new = xsltApplyStylesheetUser(m->stylesheet, rdoc, 0, 0, 0, ctxt);
 613             if ((ctxt->state == XSLT_STATE_ERROR) ||
 614                 (ctxt->state == XSLT_STATE_STOPPED)){
 615                 yaz_log(YLOG_WARN, "XSLT transformation failed from %s",
 616                         cl->database->database->url);
 617                 xmlFreeDoc(new);
 618                 xmlFreeDoc(rdoc);
 619                 return 0;
 620             }
 621         }
 622 #endif
 623
 624         xmlFreeDoc(rdoc);
 625         rdoc = new;
 626     }
 627     if (global_parameters.dump_records)
 628     {
 629         fprintf(stderr, "Record from %s\n----------------\n",
 630                 cl->database->database->url);
 631 #if LIBXML_VERSION >= 20600
 632         xmlDocFormatDump(stderr, rdoc, 1);
 633 #else
 634         xmlDocDump(stderr, rdoc);
 635 #endif
 636     }
 637     return rdoc;
 638 }
 639
 640 // Extract what appears to be years from buf, storing highest and
 641 // lowest values.
 642 static int extract_years(const char *buf, int *first, int *last)
 643 {
 644     *first = -1;
 645     *last = -1;
 646     while (*buf)
 647     {
 648         const char *e;
 649         int len;
 650
 651         while (*buf && !isdigit(*buf))
 652             buf++;
 653         len = 0;
 654         for (e = buf; *e && isdigit(*e); e++)
 655             len++;
 656         if (len == 4)
 657         {
 658             int value = atoi(buf);
 659             if (*first < 0 || value < *first)
 660                 *first = value;
 661             if (*last < 0 || value > *last)
 662                 *last = value;
 663         }
 664         buf = e;
 665     }
 666     return *first;
 667 }
 668
 669 static struct record *ingest_record(struct client *cl, Z_External *rec)
 670 {
 671     xmlDoc *xdoc = normalize_record(cl, rec);
 672     xmlNode *root, *n;
 673     struct record *res;
 674     struct record_cluster *cluster;
 675     struct session *se = cl->session;
 676     xmlChar *mergekey, *mergekey_norm;
 677     xmlChar *type = 0;
 678     xmlChar *value = 0;
 679     struct conf_service *service = global_parameters.server->service;
 680
 681     if (!xdoc)
 682         return 0;
 683
 684     root = xmlDocGetRootElement(xdoc);
 685     if (!(mergekey = xmlGetProp(root, (xmlChar *) "mergekey")))
 686     {
 687         yaz_log(YLOG_WARN, "No mergekey found in record");
 688         xmlFreeDoc(xdoc);
 689         return 0;
 690     }
 691
 692     res = nmem_malloc(se->nmem, sizeof(struct record));
 693     res->next = 0;
 694     res->client = cl;
 695     res->metadata = nmem_malloc(se->nmem,
 696             sizeof(struct record_metadata*) * service->num_metadata);
 697     memset(res->metadata, 0, sizeof(struct record_metadata*) * service->num_metadata);
 698
 699     mergekey_norm = (xmlChar *) nmem_strdup(se->nmem, (char*) mergekey);
 700     xmlFree(mergekey);
 701     normalize_mergekey((char *) mergekey_norm, 0);
 702
 703     cluster = reclist_insert(se->reclist, res, (char *) mergekey_norm,
 704                              &se->total_merged);
 705     if (global_parameters.dump_records)
 706         yaz_log(YLOG_LOG, "Cluster id %d from %s (#%d)", cluster->recid,
 707                 cl->database->database->url, cl->records);
 708     if (!cluster)
 709     {
 710         /* no room for record */
 711         xmlFreeDoc(xdoc);
 712         return 0;
 713     }
 714     relevance_newrec(se->relevance, cluster);
 715
 716     for (n = root->children; n; n = n->next)
 717     {
 718         if (type)
 719             xmlFree(type);
 720         if (value)
 721             xmlFree(value);
 722         type = value = 0;
 723
 724         if (n->type != XML_ELEMENT_NODE)
 725             continue;
 726         if (!strcmp((const char *) n->name, "metadata"))
 727         {
 728             struct conf_metadata *md = 0;
 729             struct conf_sortkey *sk = 0;
 730             struct record_metadata **wheretoput, *newm;
 731             int imeta;
 732             int first, last;
 733
 734             type = xmlGetProp(n, (xmlChar *) "type");
 735             value = xmlNodeListGetString(xdoc, n->children, 0);
 736
 737             if (!type || !value)
 738                 continue;
 739
 740             // First, find out what field we're looking at
 741             for (imeta = 0; imeta < service->num_metadata; imeta++)
 742                 if (!strcmp((const char *) type, service->metadata[imeta].name))
 743                 {
 744                     md = &service->metadata[imeta];
 745                     if (md->sortkey_offset >= 0)
 746                         sk = &service->sortkeys[md->sortkey_offset];
 747                     break;
 748                 }
 749             if (!md)
 750             {
 751                 yaz_log(YLOG_WARN, "Ignoring unknown metadata element: %s", type);
 752                 continue;
 753             }
 754
 755             // Find out where we are putting it
 756             if (md->merge == Metadata_merge_no)
 757                 wheretoput = &res->metadata[imeta];
 758             else
 759                 wheretoput = &cluster->metadata[imeta];
 760
 761             // Put it there
 762             newm = nmem_malloc(se->nmem, sizeof(struct record_metadata));
 763             newm->next = 0;
 764             if (md->type == Metadata_type_generic)
 765             {
 766                 char *p, *pe;
 767                 for (p = (char *) value; *p && isspace(*p); p++)
 768                     ;
 769                 for (pe = p + strlen(p) - 1;
 770                         pe > p && strchr(" ,/.:([", *pe); pe--)
 771                     *pe = '\0';
 772                 newm->data.text = nmem_strdup(se->nmem, p);
 773
 774             }
 775             else if (md->type == Metadata_type_year)
 776             {
 777                 if (extract_years((char *) value, &first, &last) < 0)
 778                     continue;
 779             }
 780             else
 781             {
 782                 yaz_log(YLOG_WARN, "Unknown type in metadata element %s", type);
 783                 continue;
 784             }
 785             if (md->type == Metadata_type_year && md->merge != Metadata_merge_range)
 786             {
 787                 yaz_log(YLOG_WARN, "Only range merging supported for years");
 788                 continue;
 789             }
 790             if (md->merge == Metadata_merge_unique)
 791             {
 792                 struct record_metadata *mnode;
 793                 for (mnode = *wheretoput; mnode; mnode = mnode->next)
 794                     if (!strcmp((const char *) mnode->data.text, newm->data.text))
 795                         break;
 796                 if (!mnode)
 797                 {
 798                     newm->next = *wheretoput;
 799                     *wheretoput = newm;
 800                 }
 801             }
 802             else if (md->merge == Metadata_merge_longest)
 803             {
 804                 if (!*wheretoput ||
 805                         strlen(newm->data.text) > strlen((*wheretoput)->data.text))
 806                 {
 807                     *wheretoput = newm;
 808                     if (sk)
 809                     {
 810                         char *s = nmem_strdup(se->nmem, newm->data.text);
 811                         if (!cluster->sortkeys[md->sortkey_offset])
 812                             cluster->sortkeys[md->sortkey_offset] =
 813                                 nmem_malloc(se->nmem, sizeof(union data_types));
 814                         normalize_mergekey(s,
 815                                 (sk->type == Metadata_sortkey_skiparticle));
 816                         cluster->sortkeys[md->sortkey_offset]->text = s;
 817                     }
 818                 }
 819             }
 820             else if (md->merge == Metadata_merge_all || md->merge == Metadata_merge_no)
 821             {
 822                 newm->next = *wheretoput;
 823                 *wheretoput = newm;
 824             }
 825             else if (md->merge == Metadata_merge_range)
 826             {
 827                 assert(md->type == Metadata_type_year);
 828                 if (!*wheretoput)
 829                 {
 830                     *wheretoput = newm;
 831                     (*wheretoput)->data.number.min = first;
 832                     (*wheretoput)->data.number.max = last;
 833                     if (sk)
 834                         cluster->sortkeys[md->sortkey_offset] = &newm->data;
 835                 }
 836                 else
 837                 {
 838                     if (first < (*wheretoput)->data.number.min)
 839                         (*wheretoput)->data.number.min = first;
 840                     if (last > (*wheretoput)->data.number.max)
 841                         (*wheretoput)->data.number.max = last;
 842                 }
 843 #ifdef GAGA
 844                 if (sk)
 845                 {
 846                     union data_types *sdata = cluster->sortkeys[md->sortkey_offset];
 847                     yaz_log(YLOG_LOG, "SK range: %d-%d", sdata->number.min, sdata->number.max);
 848                 }
 849 #endif
 850             }
 851             else
 852                 yaz_log(YLOG_WARN, "Don't know how to merge on element name %s", md->name);
 853
 854             if (md->rank)
 855                 relevance_countwords(se->relevance, cluster,
 856                                      (char *) value, md->rank);
 857             if (md->termlist)
 858             {
 859                 if (md->type == Metadata_type_year)
 860                 {
 861                     char year[64];
 862                     sprintf(year, "%d", last);
 863                     add_facet(se, (char *) type, year);
 864                     if (first != last)
 865                     {
 866                         sprintf(year, "%d", first);
 867                         add_facet(se, (char *) type, year);
 868                     }
 869                 }
 870                 else
 871                     add_facet(se, (char *) type, (char *) value);
 872             }
 873             xmlFree(type);
 874             xmlFree(value);
 875             type = value = 0;
 876         }
 877         else
 878             yaz_log(YLOG_WARN, "Unexpected element %s in internal record", n->name);
 879     }
 880     if (type)
 881         xmlFree(type);
 882     if (value)
 883         xmlFree(value);
 884
 885     xmlFreeDoc(xdoc);
 886
 887     relevance_donerecord(se->relevance, cluster);
 888     se->total_records++;
 889
 890     return res;
 891 }
 892
 893 // Retrieve first defined value for 'name' for given database.
 894 // Will be extended to take into account user associated with session
 895 char *session_setting_oneval(struct session_database *db, int offset)
 896 {
 897     if (!db->settings[offset])
 898         return "";
 899     return db->settings[offset]->value;
 900 }
 901
 902 static void ingest_records(struct client *cl, Z_Records *r)
 903 {
 904 #if USE_TIMING
 905     yaz_timing_t t = yaz_timing_create();
 906 #endif
 907     struct record *rec;
 908     struct session *s = cl->session;
 909     Z_NamePlusRecordList *rlist;
 910     int i;
 911
 912     if (r->which != Z_Records_DBOSD)
 913         return;
 914     rlist = r->u.databaseOrSurDiagnostics;
 915     for (i = 0; i < rlist->num_records; i++)
 916     {
 917         Z_NamePlusRecord *npr = rlist->records[i];
 918
 919         cl->records++;
 920         if (npr->which != Z_NamePlusRecord_databaseRecord)
 921         {
 922             yaz_log(YLOG_WARN,
 923                     "Unexpected record type, probably diagnostic %s",
 924                     cl->database->database->url);
 925             continue;
 926         }
 927
 928         rec = ingest_record(cl, npr->u.databaseRecord);
 929         if (!rec)
 930             continue;
 931     }
 932     if (s->watchlist[SESSION_WATCH_RECORDS].fun && rlist->num_records)
 933         session_alert_watch(s, SESSION_WATCH_RECORDS);
 934
 935 #if USE_TIMING
 936     yaz_timing_stop(t);
 937     yaz_log(YLOG_LOG, "ingest_records %6.5f %3.2f %3.2f",
 938             yaz_timing_get_real(t), yaz_timing_get_user(t),
 939             yaz_timing_get_sys(t));
 940     yaz_timing_destroy(&t);
 941 #endif
 942 }
 943
 944 static void do_presentResponse(IOCHAN i, Z_APDU *a)
 945 {
 946     struct connection *co = iochan_getdata(i);
 947     struct client *cl = co->client;
 948     Z_PresentResponse *r = a->u.presentResponse;
 949
 950     if (r->records) {
 951         Z_Records *recs = r->records;
 952         if (recs->which == Z_Records_NSD)
 953         {
 954             yaz_log(YLOG_WARN, "Non-surrogate diagnostic %s",
 955                     cl->database->database->url);
 956             cl->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 957             cl->state = Client_Error;
 958         }
 959     }
 960
 961     if (!*r->presentStatus && cl->state != Client_Error)
 962     {
 963         yaz_log(YLOG_DEBUG, "Good Present response %s",
 964                 cl->database->database->url);
 965         ingest_records(cl, r->records);
 966         cl->state = Client_Idle;
 967     }
 968     else if (*r->presentStatus)
 969     {
 970         yaz_log(YLOG_WARN, "Bad Present response %s",
 971                 cl->database->database->url);
 972         cl->state = Client_Error;
 973     }
 974 }
 975
 976 static void handler(IOCHAN i, int event)
 977 {
 978     struct connection *co = iochan_getdata(i);
 979     struct client *cl = co->client;
 980     struct session *se = 0;
 981
 982     if (cl)
 983         se = cl->session;
 984     else
 985     {
 986         yaz_log(YLOG_WARN, "Destroying orphan connection");
 987         connection_destroy(co);
 988         return;
 989     }
 990
 991     if (co->state == Conn_Connecting && event & EVENT_OUTPUT)
 992     {
 993         int errcode;
 994         socklen_t errlen = sizeof(errcode);
 995
 996         if (getsockopt(cs_fileno(co->link), SOL_SOCKET, SO_ERROR, &errcode,
 997             &errlen) < 0 || errcode != 0)
 998         {
 999             client_fatal(cl);
1000             return;
1001         }
1002         else
1003         {
1004             yaz_log(YLOG_DEBUG, "Connect OK");
1005             co->state = Conn_Open;
1006             if (cl)
1007                 cl->state = Client_Connected;
1008         }
1009     }
1010
1011     else if (event & EVENT_INPUT)
1012     {
1013         int len = cs_get(co->link, &co->ibuf, &co->ibufsize);
1014
1015         if (len < 0)
1016         {
1017             yaz_log(YLOG_WARN|YLOG_ERRNO, "Error reading from %s",
1018                     cl->database->database->url);
1019             connection_destroy(co);
1020             return;
1021         }
1022         else if (len == 0)
1023         {
1024             yaz_log(YLOG_WARN, "EOF reading from %s", cl->database->database->url);
1025             connection_destroy(co);
1026             return;
1027         }
1028         else if (len > 1) // We discard input if we have no connection
1029         {
1030             co->state = Conn_Open;
1031
1032             if (cl && (cl->requestid == se->requestid || cl->state == Client_Initializing))
1033             {
1034                 Z_APDU *a;
1035
1036                 odr_reset(global_parameters.odr_in);
1037                 odr_setbuf(global_parameters.odr_in, co->ibuf, len, 0);
1038                 if (!z_APDU(global_parameters.odr_in, &a, 0, 0))
1039                 {
1040                     client_fatal(cl);
1041                     return;
1042                 }
1043                 switch (a->which)
1044                 {
1045                     case Z_APDU_initResponse:
1046                         do_initResponse(i, a);
1047                         break;
1048                     case Z_APDU_searchResponse:
1049                         do_searchResponse(i, a);
1050                         break;
1051                     case Z_APDU_presentResponse:
1052                         do_presentResponse(i, a);
1053                         break;
1054                     case Z_APDU_close:
1055                         do_closeResponse(i, a);
1056                         break;
1057                     default:
1058                         yaz_log(YLOG_WARN,
1059                                 "Unexpected Z39.50 response from %s",
1060                                 cl->database->database->url);
1061                         client_fatal(cl);
1062                         return;
1063                 }
1064                 // We aren't expecting staggered output from target
1065                 // if (cs_more(t->link))
1066                 //    iochan_setevent(i, EVENT_INPUT);
1067             }
1068             else  // we throw away response and go to idle mode
1069             {
1070                 yaz_log(YLOG_DEBUG, "Ignoring result of expired operation");
1071                 cl->state = Client_Idle;
1072             }
1073         }
1074         /* if len==1 we do nothing but wait for more input */
1075     }
1076
1077     if (cl->state == Client_Connected) {
1078         send_init(i);
1079     }
1080
1081     if (cl->state == Client_Idle)
1082     {
1083         if (cl->requestid != se->requestid && *se->query) {
1084             send_search(i);
1085         }
1086         else if (cl->hits > 0 && cl->records < global_parameters.toget &&
1087             cl->records < cl->hits) {
1088             send_present(i);
1089         }
1090     }
1091 }
1092
1093 // Disassociate connection from client
1094 static void connection_release(struct connection *co)
1095 {
1096     struct client *cl = co->client;
1097
1098     yaz_log(YLOG_DEBUG, "Connection release %s", co->host->hostport);
1099     if (!cl)
1100         return;
1101     cl->connection = 0;
1102     co->client = 0;
1103 }
1104
1105 // Close connection and recycle structure
1106 static void connection_destroy(struct connection *co)
1107 {
1108     struct host *h = co->host;
1109     cs_close(co->link);
1110     iochan_destroy(co->iochan);
1111
1112     yaz_log(YLOG_DEBUG, "Connection destroy %s", co->host->hostport);
1113     if (h->connections == co)
1114         h->connections = co->next;
1115     else
1116     {
1117         struct connection *pco;
1118         for (pco = h->connections; pco && pco->next != co; pco = pco->next)
1119             ;
1120         if (pco)
1121             pco->next = co->next;
1122         else
1123             abort();
1124     }
1125     if (co->client)
1126     {
1127         if (co->client->state != Client_Idle)
1128             co->client->state = Client_Disconnected;
1129         co->client->connection = 0;
1130     }
1131     co->next = connection_freelist;
1132     connection_freelist = co;
1133 }
1134
1135 // Creates a new connection for client, associated with the host of
1136 // client's database
1137 static struct connection *connection_create(struct client *cl)
1138 {
1139     struct connection *new;
1140     COMSTACK link;
1141     int res;
1142     void *addr;
1143
1144
1145     if (!(link = cs_create(tcpip_type, 0, PROTO_Z3950)))
1146         {
1147             yaz_log(YLOG_FATAL|YLOG_ERRNO, "Failed to create comstack");
1148             exit(1);
1149         }
1150
1151     if (0 == strlen(global_parameters.zproxy_override)){
1152         /* no Z39.50 proxy needed - direct connect */
1153         yaz_log(YLOG_DEBUG, "Connection create %s", cl->database->database->url);
1154
1155         if (!(addr = cs_straddr(link, cl->database->database->host->ipport)))
1156             {
1157                 yaz_log(YLOG_WARN|YLOG_ERRNO,
1158                         "Lookup of IP address %s failed",
1159                         cl->database->database->host->ipport);
1160                 return 0;
1161             }
1162
1163     } else {
1164         /* Z39.50 proxy connect */
1165         yaz_log(YLOG_DEBUG, "Connection create %s proxy %s",
1166                 cl->database->database->url, global_parameters.zproxy_override);
1167
1168         if (!(addr = cs_straddr(link, global_parameters.zproxy_override)))
1169             {
1170                 yaz_log(YLOG_WARN|YLOG_ERRNO,
1171                         "Lookup of IP address %s failed",
1172                         global_parameters.zproxy_override);
1173                 return 0;
1174             }
1175     }
1176
1177     res = cs_connect(link, addr);
1178     if (res < 0)
1179     {
1180         yaz_log(YLOG_WARN|YLOG_ERRNO, "cs_connect %s", cl->database->database->url);
1181         return 0;
1182     }
1183
1184     if ((new = connection_freelist))
1185         connection_freelist = new->next;
1186     else
1187     {
1188         new = xmalloc(sizeof (struct connection));
1189         new->ibuf = 0;
1190         new->ibufsize = 0;
1191     }
1192     new->state = Conn_Connecting;
1193     new->host = cl->database->database->host;
1194     new->next = new->host->connections;
1195     new->host->connections = new;
1196     new->client = cl;
1197     cl->connection = new;
1198     new->link = link;
1199
1200     new->iochan = iochan_create(cs_fileno(link), handler, 0);
1201     iochan_setdata(new->iochan, new);
1202     new->iochan->next = channel_list;
1203     channel_list = new->iochan;
1204     return new;
1205 }
1206
1207 // Close connection and set state to error
1208 static void client_fatal(struct client *cl)
1209 {
1210     yaz_log(YLOG_WARN, "Fatal error from %s", cl->database->database->url);
1211     connection_destroy(cl->connection);
1212     cl->state = Client_Error;
1213 }
1214
1215 // Ensure that client has a connection associated
1216 static int client_prep_connection(struct client *cl)
1217 {
1218     struct connection *co;
1219     struct session *se = cl->session;
1220     struct host *host = cl->database->database->host;
1221
1222     co = cl->connection;
1223
1224     yaz_log(YLOG_DEBUG, "Client prep %s", cl->database->database->url);
1225
1226     if (!co)
1227     {
1228         // See if someone else has an idle connection
1229         // We should look at timestamps here to select the longest-idle connection
1230         for (co = host->connections; co; co = co->next)
1231             if (co->state == Conn_Open && (!co->client || co->client->session != se))
1232                 break;
1233         if (co)
1234         {
1235             connection_release(co);
1236             cl->connection = co;
1237             co->client = cl;
1238         }
1239         else
1240             co = connection_create(cl);
1241     }
1242     if (co)
1243     {
1244         if (co->state == Conn_Connecting)
1245         {
1246             cl->state = Client_Connecting;
1247             iochan_setflag(co->iochan, EVENT_OUTPUT);
1248         }
1249         else if (co->state == Conn_Open)
1250         {
1251             if (cl->state == Client_Error || cl->state == Client_Disconnected)
1252                 cl->state = Client_Idle;
1253             iochan_setflag(co->iochan, EVENT_OUTPUT);
1254         }
1255         return 1;
1256     }
1257     else
1258         return 0;
1259 }
1260
1261 static struct client *client_create(void)
1262 {
1263     struct client *r;
1264     if (client_freelist)
1265     {
1266         r = client_freelist;
1267         client_freelist = client_freelist->next;
1268     }
1269     else
1270         r = xmalloc(sizeof(struct client));
1271     r->database = 0;
1272     r->connection = 0;
1273     r->session = 0;
1274     r->hits = 0;
1275     r->records = 0;
1276     r->setno = 0;
1277     r->requestid = -1;
1278     r->diagnostic = 0;
1279     r->state = Client_Disconnected;
1280     r->next = 0;
1281     return r;
1282 }
1283
1284 void client_destroy(struct client *c)
1285 {
1286     struct session *se = c->session;
1287     if (c == se->clients)
1288         se->clients = c->next;
1289     else
1290     {
1291         struct client *cc;
1292         for (cc = se->clients; cc && cc->next != c; cc = cc->next)
1293             ;
1294         if (cc)
1295             cc->next = c->next;
1296     }
1297     if (c->connection)
1298         connection_release(c->connection);
1299     c->next = client_freelist;
1300     client_freelist = c;
1301 }
1302
1303 void session_set_watch(struct session *s, int what, session_watchfun fun, void *data)
1304 {
1305     s->watchlist[what].fun = fun;
1306     s->watchlist[what].data = data;
1307 }
1308
1309 void session_alert_watch(struct session *s, int what)
1310 {
1311     if (!s->watchlist[what].fun)
1312         return;
1313     (*s->watchlist[what].fun)(s->watchlist[what].data);
1314     s->watchlist[what].fun = 0;
1315     s->watchlist[what].data = 0;
1316 }
1317
1318 //callback for grep_databases
1319 static void select_targets_callback(void *context, struct session_database *db)
1320 {
1321     struct session *se = (struct session*) context;
1322     struct client *cl = client_create();
1323     cl->database = db;
1324     cl->session = se;
1325     cl->next = se->clients;
1326     se->clients = cl;
1327 }
1328
1329 // Associates a set of clients with a session;
1330 // Note: Session-databases represent databases with per-session setting overrides
1331 int select_targets(struct session *se, struct database_criterion *crit)
1332 {
1333     while (se->clients)
1334         client_destroy(se->clients);
1335
1336     return session_grep_databases(se, crit, select_targets_callback);
1337 }
1338
1339 int session_active_clients(struct session *s)
1340 {
1341     struct client *c;
1342     int res = 0;
1343
1344     for (c = s->clients; c; c = c->next)
1345         if (c->connection && (c->state == Client_Connecting ||
1346                     c->state == Client_Initializing ||
1347                     c->state == Client_Searching ||
1348                     c->state == Client_Presenting))
1349             res++;
1350
1351     return res;
1352 }
1353
1354 // parses crit1=val1,crit2=val2|val3,...
1355 static struct database_criterion *parse_filter(NMEM m, const char *buf)
1356 {
1357     struct database_criterion *res = 0;
1358     char **values;
1359     int num;
1360     int i;
1361
1362     if (!buf || !*buf)
1363         return 0;
1364     nmem_strsplit(m, ",", buf,  &values, &num);
1365     for (i = 0; i < num; i++)
1366     {
1367         char **subvalues;
1368         int subnum;
1369         int subi;
1370         struct database_criterion *new = nmem_malloc(m, sizeof(*new));
1371         char *eq = strchr(values[i], '=');
1372         if (!eq)
1373         {
1374             yaz_log(YLOG_WARN, "Missing equal-sign in filter");
1375             return 0;
1376         }
1377         *(eq++) = '\0';
1378         new->name = values[i];
1379         nmem_strsplit(m, "|", eq, &subvalues, &subnum);
1380         new->values = 0;
1381         for (subi = 0; subi < subnum; subi++)
1382         {
1383             struct database_criterion_value *newv = nmem_malloc(m, sizeof(*newv));
1384             newv->value = subvalues[subi];
1385             newv->next = new->values;
1386             new->values = newv;
1387         }
1388         new->next = res;
1389         res = new;
1390     }
1391     return res;
1392 }
1393
1394 char *search(struct session *se, char *query, char *filter)
1395 {
1396     int live_channels = 0;
1397     struct client *cl;
1398     struct database_criterion *criteria;
1399
1400     yaz_log(YLOG_DEBUG, "Search");
1401
1402     nmem_reset(se->nmem);
1403     criteria = parse_filter(se->nmem, filter);
1404     strcpy(se->query, query);
1405     se->requestid++;
1406     select_targets(se, criteria);
1407     for (cl = se->clients; cl; cl = cl->next)
1408     {
1409         if (client_prep_connection(cl))
1410             live_channels++;
1411     }
1412     if (live_channels)
1413     {
1414         int maxrecs = live_channels * global_parameters.toget;
1415         se->num_termlists = 0;
1416         se->reclist = reclist_create(se->nmem, maxrecs);
1417         // This will be initialized in send_search()
1418         se->relevance = 0;
1419         se->total_records = se->total_hits = se->total_merged = 0;
1420         se->expected_maxrecs = maxrecs;
1421     }
1422     else
1423         return "NOTARGETS";
1424
1425     return 0;
1426 }
1427
1428 // Apply a session override to a database
1429 void session_apply_setting(struct session *se, char *dbname, char *setting, char *value)
1430 {
1431     struct session_database *sdb;
1432
1433     for (sdb = se->databases; sdb; sdb = sdb->next)
1434         if (!strcmp(dbname, sdb->database->url))
1435         {
1436             struct setting *new = nmem_malloc(se->session_nmem, sizeof(*new));
1437             int offset = settings_offset(setting);
1438
1439             if (offset < 0)
1440             {
1441                 yaz_log(YLOG_WARN, "Unknown setting %s", setting);
1442                 return;
1443             }
1444             new->precedence = 0;
1445             new->target = dbname;
1446             new->name = setting;
1447             new->value = value;
1448             new->next = sdb->settings[offset];
1449             sdb->settings[offset] = new;
1450             break;
1451         }
1452     if (!sdb)
1453         yaz_log(YLOG_WARN, "Unknown database in setting override: %s", dbname);
1454 }
1455
1456 void session_init_databases_fun(void *context, struct database *db)
1457 {
1458     struct session *se = (struct session *) context;
1459     struct session_database *new = nmem_malloc(se->session_nmem, sizeof(*new));
1460     int num = settings_num();
1461     int i;
1462
1463     new->database = db;
1464     new->settings = nmem_malloc(se->session_nmem, sizeof(struct settings *) * num);
1465     for (i = 0; i < num; i++)
1466         new->settings[i] = db->settings[i];
1467     new->next = se->databases;
1468     se->databases = new;
1469 }
1470
1471 // Initialize session_database list -- this represents this session's view
1472 // of the database list -- subject to modification by the settings ws command
1473 void session_init_databases(struct session *se)
1474 {
1475     se->databases = 0;
1476     grep_databases(se, 0, session_init_databases_fun);
1477 }
1478
1479 void destroy_session(struct session *s)
1480 {
1481     yaz_log(YLOG_LOG, "Destroying session");
1482     while (s->clients)
1483         client_destroy(s->clients);
1484     nmem_destroy(s->nmem);
1485     wrbuf_destroy(s->wrbuf);
1486 }
1487
1488 struct session *new_session(NMEM nmem)
1489 {
1490     int i;
1491     struct session *session = nmem_malloc(nmem, sizeof(*session));
1492
1493     yaz_log(YLOG_DEBUG, "New Pazpar2 session");
1494
1495     session->total_hits = 0;
1496     session->total_records = 0;
1497     session->num_termlists = 0;
1498     session->reclist = 0;
1499     session->requestid = -1;
1500     session->clients = 0;
1501     session->expected_maxrecs = 0;
1502     session->query[0] = '\0';
1503     session->session_nmem = nmem;
1504     session->nmem = nmem_create();
1505     session->wrbuf = wrbuf_alloc();
1506     session_init_databases(session);
1507     for (i = 0; i <= SESSION_WATCH_MAX; i++)
1508     {
1509         session->watchlist[i].data = 0;
1510         session->watchlist[i].fun = 0;
1511     }
1512
1513     return session;
1514 }
1515
1516 struct hitsbytarget *hitsbytarget(struct session *se, int *count)
1517 {
1518     static struct hitsbytarget res[1000]; // FIXME MM
1519     struct client *cl;
1520
1521     *count = 0;
1522     for (cl = se->clients; cl; cl = cl->next)
1523     {
1524         char *name = session_setting_oneval(cl->database, PZ_NAME);
1525
1526         res[*count].id = cl->database->database->url;
1527         res[*count].name = *name ? name : "Unknown";
1528         res[*count].hits = cl->hits;
1529         res[*count].records = cl->records;
1530         res[*count].diagnostic = cl->diagnostic;
1531         res[*count].state = client_states[cl->state];
1532         res[*count].connected  = cl->connection ? 1 : 0;
1533         (*count)++;
1534     }
1535
1536     return res;
1537 }
1538
1539 struct termlist_score **termlist(struct session *s, const char *name, int *num)
1540 {
1541     int i;
1542
1543     for (i = 0; i < s->num_termlists; i++)
1544         if (!strcmp((const char *) s->termlists[i].name, name))
1545             return termlist_highscore(s->termlists[i].termlist, num);
1546     return 0;
1547 }
1548
1549 #ifdef MISSING_HEADERS
1550 void report_nmem_stats(void)
1551 {
1552     size_t in_use, is_free;
1553
1554     nmem_get_memory_in_use(&in_use);
1555     nmem_get_memory_free(&is_free);
1556
1557     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld",
1558             (long) in_use, (long) is_free);
1559 }
1560 #endif
1561
1562 struct record_cluster *show_single(struct session *s, int id)
1563 {
1564     struct record_cluster *r;
1565
1566     reclist_rewind(s->reclist);
1567     while ((r = reclist_read_record(s->reclist)))
1568         if (r->recid == id)
1569             return r;
1570     return 0;
1571 }
1572
1573 struct record_cluster **show(struct session *s, struct reclist_sortparms *sp, int start,
1574         int *num, int *total, int *sumhits, NMEM nmem_show)
1575 {
1576     struct record_cluster **recs = nmem_malloc(nmem_show, *num
1577                                        * sizeof(struct record_cluster *));
1578     struct reclist_sortparms *spp;
1579     int i;
1580 #if USE_TIMING
1581     yaz_timing_t t = yaz_timing_create();
1582 #endif
1583
1584     for (spp = sp; spp; spp = spp->next)
1585         if (spp->type == Metadata_sortkey_relevance)
1586         {
1587             relevance_prepare_read(s->relevance, s->reclist);
1588             break;
1589         }
1590     reclist_sort(s->reclist, sp);
1591
1592     *total = s->reclist->num_records;
1593     *sumhits = s->total_hits;
1594
1595     for (i = 0; i < start; i++)
1596         if (!reclist_read_record(s->reclist))
1597         {
1598             *num = 0;
1599             recs = 0;
1600             break;
1601         }
1602
1603     for (i = 0; i < *num; i++)
1604     {
1605         struct record_cluster *r = reclist_read_record(s->reclist);
1606         if (!r)
1607         {
1608             *num = i;
1609             break;
1610         }
1611         recs[i] = r;
1612     }
1613 #if USE_TIMING
1614     yaz_timing_stop(t);
1615     yaz_log(YLOG_LOG, "show %6.5f %3.2f %3.2f",
1616             yaz_timing_get_real(t), yaz_timing_get_user(t),
1617             yaz_timing_get_sys(t));
1618     yaz_timing_destroy(&t);
1619 #endif
1620     return recs;
1621 }
1622
1623 void statistics(struct session *se, struct statistics *stat)
1624 {
1625     struct client *cl;
1626     int count = 0;
1627
1628     memset(stat, 0, sizeof(*stat));
1629     for (cl = se->clients; cl; cl = cl->next)
1630     {
1631         if (!cl->connection)
1632             stat->num_no_connection++;
1633         switch (cl->state)
1634         {
1635             case Client_Connecting: stat->num_connecting++; break;
1636             case Client_Initializing: stat->num_initializing++; break;
1637             case Client_Searching: stat->num_searching++; break;
1638             case Client_Presenting: stat->num_presenting++; break;
1639             case Client_Idle: stat->num_idle++; break;
1640             case Client_Failed: stat->num_failed++; break;
1641             case Client_Error: stat->num_error++; break;
1642             default: break;
1643         }
1644         count++;
1645     }
1646     stat->num_hits = se->total_hits;
1647     stat->num_records = se->total_records;
1648
1649     stat->num_clients = count;
1650 }
1651
1652 static void start_http_listener(void)
1653 {
1654     char hp[128] = "";
1655     struct conf_server *ser = global_parameters.server;
1656
1657     if (*global_parameters.listener_override)
1658         strcpy(hp, global_parameters.listener_override);
1659     else
1660     {
1661         strcpy(hp, ser->host ? ser->host : "");
1662         if (ser->port)
1663         {
1664             if (*hp)
1665                 strcat(hp, ":");
1666             sprintf(hp + strlen(hp), "%d", ser->port);
1667         }
1668     }
1669     http_init(hp);
1670 }
1671
1672 static void start_proxy(void)
1673 {
1674     char hp[128] = "";
1675     struct conf_server *ser = global_parameters.server;
1676
1677     if (*global_parameters.proxy_override)
1678         strcpy(hp, global_parameters.proxy_override);
1679     else if (ser->proxy_host || ser->proxy_port)
1680     {
1681         strcpy(hp, ser->proxy_host ? ser->proxy_host : "");
1682         if (ser->proxy_port)
1683         {
1684             if (*hp)
1685                 strcat(hp, ":");
1686             sprintf(hp + strlen(hp), "%d", ser->proxy_port);
1687         }
1688     }
1689     else
1690         return;
1691
1692     http_set_proxyaddr(hp, ser->myurl ? ser->myurl : "");
1693 }
1694
1695 static void start_zproxy(void)
1696 {
1697     struct conf_server *ser = global_parameters.server;
1698
1699     if (*global_parameters.zproxy_override){
1700         yaz_log(YLOG_LOG, "Z39.50 proxy  %s",
1701                 global_parameters.zproxy_override);
1702         return;
1703     }
1704
1705     else if (ser->zproxy_host || ser->zproxy_port)
1706     {
1707         char hp[128] = "";
1708
1709         strcpy(hp, ser->zproxy_host ? ser->zproxy_host : "");
1710         if (ser->zproxy_port)
1711         {
1712             if (*hp)
1713                 strcat(hp, ":");
1714             else
1715                 strcat(hp, "@:");
1716
1717             sprintf(hp + strlen(hp), "%d", ser->zproxy_port);
1718         }
1719         strcpy(global_parameters.zproxy_override, hp);
1720         yaz_log(YLOG_LOG, "Z39.50 proxy  %s",
1721                 global_parameters.zproxy_override);
1722
1723     }
1724     else
1725         return;
1726 }
1727
1728
1729
1730 int main(int argc, char **argv)
1731 {
1732     int ret;
1733     char *arg;
1734
1735     if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
1736         yaz_log(YLOG_WARN|YLOG_ERRNO, "signal");
1737
1738     yaz_log_init(YLOG_DEFAULT_LEVEL, "pazpar2", 0);
1739
1740     while ((ret = options("t:f:x:h:p:z:s:d", argv, argc, &arg)) != -2)
1741     {
1742         switch (ret) {
1743             case 'f':
1744                 if (!read_config(arg))
1745                     exit(1);
1746                 break;
1747             case 'h':
1748                 strcpy(global_parameters.listener_override, arg);
1749                 break;
1750             case 'p':
1751                 strcpy(global_parameters.proxy_override, arg);
1752                 break;
1753             case 'z':
1754                 strcpy(global_parameters.zproxy_override, arg);
1755                 break;
1756             case 't':
1757                 strcpy(global_parameters.settings_path_override, arg);
1758                 break;
1759             case 's':
1760                 load_simpletargets(arg);
1761                 break;
1762             case 'd':
1763                 global_parameters.dump_records = 1;
1764                 break;
1765             default:
1766                 fprintf(stderr, "Usage: pazpar2\n"
1767                         "    -f configfile\n"
1768                         "    -h [host:]port          (REST protocol listener)\n"
1769                         "    -C cclconfig\n"
1770                         "    -s simpletargetfile\n"
1771                         "    -p hostname[:portno]    (HTTP proxy)\n"
1772                         "    -z hostname[:portno]    (Z39.50 proxy)\n"
1773                         "    -d                      (show internal records)\n");
1774                 exit(1);
1775         }
1776     }
1777
1778     if (!config)
1779     {
1780         yaz_log(YLOG_FATAL, "Load config with -f");
1781         exit(1);
1782     }
1783     global_parameters.server = config->servers;
1784
1785     start_http_listener();
1786     start_proxy();
1787     start_zproxy();
1788
1789     if (*global_parameters.settings_path_override)
1790         settings_read(global_parameters.settings_path_override);
1791     else if (global_parameters.server->settings)
1792         settings_read(global_parameters.server->settings);
1793     else
1794         yaz_log(YLOG_WARN, "No settings-directory specified. Problems may well ensue!");
1795     prepare_databases();
1796     global_parameters.odr_in = odr_createmem(ODR_DECODE);
1797     global_parameters.odr_out = odr_createmem(ODR_ENCODE);
1798
1799     event_loop(&channel_list);
1800
1801     return 0;
1802 }
1803
1804 /*
1805  * Local variables:
1806  * c-basic-offset: 4
1807  * indent-tabs-mode: nil
1808  * End:
1809  * vim: shiftwidth=4 tabstop=8 expandtab
1810  */