src/logic.c

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