src/pazpar2.c

   1 /* $Id: pazpar2.c,v 1.5 2006-12-22 04:43:11 quinn Exp $ */;
   2
   3 #include <stdlib.h>
   4 #include <stdio.h>
   5 #include <string.h>
   6 #include <sys/time.h>
   7 #include <unistd.h>
   8 #include <sys/socket.h>
   9 #include <netdb.h>
  10 #include <signal.h>
  11 #include <ctype.h>
  12 #include <assert.h>
  13
  14 #include <yaz/comstack.h>
  15 #include <yaz/tcpip.h>
  16 #include <yaz/proto.h>
  17 #include <yaz/readconf.h>
  18 #include <yaz/pquery.h>
  19 #include <yaz/yaz-util.h>
  20 #include <yaz/nmem.h>
  21
  22 #include "pazpar2.h"
  23 #include "eventl.h"
  24 #include "command.h"
  25 #include "http.h"
  26 #include "termlists.h"
  27 #include "reclists.h"
  28 #include "relevance.h"
  29
  30 #define PAZPAR2_VERSION "0.1"
  31 #define MAX_CHUNK 15
  32
  33 static void client_fatal(struct client *cl);
  34 static void connection_destroy(struct connection *co);
  35 static int client_prep_connection(struct client *cl);
  36 static void ingest_records(struct client *cl, Z_Records *r);
  37 void session_alert_watch(struct session *s, int what);
  38
  39 IOCHAN channel_list = 0;  // Master list of connections we're handling events to
  40
  41 static struct connection *connection_freelist = 0;
  42 static struct client *client_freelist = 0;
  43
  44 static struct host *hosts = 0;  // The hosts we know about
  45 static struct database *databases = 0; // The databases we know about
  46
  47 static char *client_states[] = {
  48     "Client_Connecting",
  49     "Client_Connected",
  50     "Client_Idle",
  51     "Client_Initializing",
  52     "Client_Searching",
  53     "Client_Presenting",
  54     "Client_Error",
  55     "Client_Failed",
  56     "Client_Disconnected",
  57     "Client_Stopped"
  58 };
  59
  60 struct parameters global_parameters =
  61 {
  62     30,
  63     "81",
  64     "Index Data PazPar2 (MasterKey)",
  65     PAZPAR2_VERSION,
  66     600, // 10 minutes
  67     60,
  68     100,
  69     MAX_CHUNK,
  70     0,
  71     0,
  72     0,
  73     0,
  74     0
  75 };
  76
  77
  78 static int send_apdu(struct client *c, Z_APDU *a)
  79 {
  80     struct connection *co = c->connection;
  81     char *buf;
  82     int len, r;
  83
  84     if (!z_APDU(global_parameters.odr_out, &a, 0, 0))
  85     {
  86         odr_perror(global_parameters.odr_out, "Encoding APDU");
  87         abort();
  88     }
  89     buf = odr_getbuf(global_parameters.odr_out, &len, 0);
  90     r = cs_put(co->link, buf, len);
  91     if (r < 0)
  92     {
  93         yaz_log(YLOG_WARN, "cs_put: %s", cs_errmsg(cs_errno(co->link)));
  94         return -1;
  95     }
  96     else if (r == 1)
  97     {
  98         fprintf(stderr, "cs_put incomplete (ParaZ does not handle that)\n");
  99         exit(1);
 100     }
 101     odr_reset(global_parameters.odr_out); /* release the APDU structure  */
 102     co->state = Conn_Waiting;
 103     return 0;
 104 }
 105
 106
 107 static void send_init(IOCHAN i)
 108 {
 109     struct connection *co = iochan_getdata(i);
 110     struct client *cl = co->client;
 111     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_initRequest);
 112
 113     a->u.initRequest->implementationId = global_parameters.implementationId;
 114     a->u.initRequest->implementationName = global_parameters.implementationName;
 115     a->u.initRequest->implementationVersion =
 116         global_parameters.implementationVersion;
 117     ODR_MASK_SET(a->u.initRequest->options, Z_Options_search);
 118     ODR_MASK_SET(a->u.initRequest->options, Z_Options_present);
 119     ODR_MASK_SET(a->u.initRequest->options, Z_Options_namedResultSets);
 120
 121     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_1);
 122     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_2);
 123     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_3);
 124     if (send_apdu(cl, a) >= 0)
 125     {
 126         iochan_setflags(i, EVENT_INPUT);
 127         cl->state = Client_Initializing;
 128     }
 129     else
 130         cl->state = Client_Error;
 131     odr_reset(global_parameters.odr_out);
 132 }
 133
 134 static void send_search(IOCHAN i)
 135 {
 136     struct connection *co = iochan_getdata(i);
 137     struct client *cl = co->client;
 138     struct session *se = cl->session;
 139     struct database *db = cl->database;
 140     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_searchRequest);
 141     int ndb, cerror, cpos;
 142     char **databaselist;
 143     Z_Query *zquery;
 144     struct ccl_rpn_node *cn;
 145     int ssub = 0, lslb = 100000, mspn = 10;
 146
 147     yaz_log(YLOG_DEBUG, "Sending search");
 148
 149     cn = ccl_find_str(global_parameters.ccl_filter, se->query, &cerror, &cpos);
 150     if (!cn)
 151         return;
 152     a->u.searchRequest->query = zquery = odr_malloc(global_parameters.odr_out,
 153             sizeof(Z_Query));
 154     zquery->which = Z_Query_type_1;
 155     zquery->u.type_1 = ccl_rpn_query(global_parameters.odr_out, cn);
 156     ccl_rpn_delete(cn);
 157
 158     for (ndb = 0; db->databases[ndb]; ndb++)
 159         ;
 160     databaselist = odr_malloc(global_parameters.odr_out, sizeof(char*) * ndb);
 161     for (ndb = 0; db->databases[ndb]; ndb++)
 162         databaselist[ndb] = db->databases[ndb];
 163
 164     a->u.presentRequest->preferredRecordSyntax =
 165             yaz_oidval_to_z3950oid(global_parameters.odr_out,
 166             CLASS_RECSYN, VAL_USMARC);
 167     a->u.searchRequest->smallSetUpperBound = &ssub;
 168     a->u.searchRequest->largeSetLowerBound = &lslb;
 169     a->u.searchRequest->mediumSetPresentNumber = &mspn;
 170     a->u.searchRequest->resultSetName = "Default";
 171     a->u.searchRequest->databaseNames = databaselist;
 172     a->u.searchRequest->num_databaseNames = ndb;
 173
 174     if (send_apdu(cl, a) >= 0)
 175     {
 176         iochan_setflags(i, EVENT_INPUT);
 177         cl->state = Client_Searching;
 178         cl->requestid = se->requestid;
 179     }
 180     else
 181         cl->state = Client_Error;
 182
 183     odr_reset(global_parameters.odr_out);
 184 }
 185
 186 static void send_present(IOCHAN i)
 187 {
 188     struct connection *co = iochan_getdata(i);
 189     struct client *cl = co->client;
 190     Z_APDU *a = zget_APDU(global_parameters.odr_out, Z_APDU_presentRequest);
 191     int toget;
 192     int start = cl->records + 1;
 193
 194     toget = global_parameters.chunk;
 195     if (toget > cl->hits - cl->records)
 196         toget = cl->hits - cl->records;
 197
 198     yaz_log(YLOG_DEBUG, "Trying to present %d records\n", toget);
 199
 200     a->u.presentRequest->resultSetStartPoint = &start;
 201     a->u.presentRequest->numberOfRecordsRequested = &toget;
 202
 203     a->u.presentRequest->resultSetId = "Default";
 204
 205     a->u.presentRequest->preferredRecordSyntax =
 206             yaz_oidval_to_z3950oid(global_parameters.odr_out,
 207             CLASS_RECSYN, VAL_USMARC);
 208
 209     if (send_apdu(cl, a) >= 0)
 210     {
 211         iochan_setflags(i, EVENT_INPUT);
 212         cl->state = Client_Presenting;
 213     }
 214     else
 215         cl->state = Client_Error;
 216     odr_reset(global_parameters.odr_out);
 217 }
 218
 219 static void do_initResponse(IOCHAN i, Z_APDU *a)
 220 {
 221     struct connection *co = iochan_getdata(i);
 222     struct client *cl = co->client;
 223     Z_InitResponse *r = a->u.initResponse;
 224
 225     yaz_log(YLOG_DEBUG, "Received init response");
 226
 227     if (*r->result)
 228     {
 229         cl->state = Client_Idle;
 230     }
 231     else
 232         cl->state = Client_Failed; // FIXME need to do something to the connection
 233 }
 234
 235 static void do_searchResponse(IOCHAN i, Z_APDU *a)
 236 {
 237     struct connection *co = iochan_getdata(i);
 238     struct client *cl = co->client;
 239     struct session *se = cl->session;
 240     Z_SearchResponse *r = a->u.searchResponse;
 241
 242     yaz_log(YLOG_DEBUG, "Searchresponse (status=%d)", *r->searchStatus);
 243
 244     if (*r->searchStatus)
 245     {
 246         cl->hits = *r->resultCount;
 247         se->total_hits += cl->hits;
 248         if (r->presentStatus && !*r->presentStatus && r->records)
 249         {
 250             yaz_log(YLOG_DEBUG, "Records in search response");
 251             cl->records += *r->numberOfRecordsReturned;
 252             ingest_records(cl, r->records);
 253         }
 254         cl->state = Client_Idle;
 255     }
 256     else
 257     {          /*"FAILED"*/
 258         cl->hits = 0;
 259         cl->state = Client_Error;
 260         if (r->records) {
 261             Z_Records *recs = r->records;
 262             if (recs->which == Z_Records_NSD)
 263             {
 264                 yaz_log(YLOG_WARN, "Non-surrogate diagnostic");
 265                 cl->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 266                 cl->state = Client_Error;
 267             }
 268         }
 269     }
 270 }
 271
 272 const char *find_field(const char *rec, const char *field)
 273 {
 274     char lbuf[5];
 275     char *line;
 276
 277     lbuf[0] = '\n';
 278     strcpy(lbuf + 1, field);
 279
 280     if ((line = strstr(rec, lbuf)))
 281         return ++line;
 282     else
 283         return 0;
 284 }
 285
 286 const char *find_subfield(const char *field, char subfield)
 287 {
 288     const char *p = field;
 289
 290     while (*p && *p != '\n')
 291     {
 292         while (*p != '\n' && *p != '\t')
 293             p++;
 294         if (*p == '\t' && *(++p) == subfield) {
 295             if (*(++p) == ' ')
 296             {
 297                 while (isspace(*p))
 298                     p++;
 299                 return p;
 300             }
 301         }
 302     }
 303     return 0;
 304 }
 305
 306 // Extract 245 $a $b 100 $a
 307 char *extract_title(struct session *s, const char *rec)
 308 {
 309     const char *field, *subfield;
 310     char *e, *ef;
 311     unsigned char *obuf, *p;
 312
 313     wrbuf_rewind(s->wrbuf);
 314
 315     if (!(field = find_field(rec, "245")))
 316         return 0;
 317     if (!(subfield = find_subfield(field, 'a')))
 318         return 0;
 319     ef = index(subfield, '\n');
 320     if ((e = index(subfield, '\t')) && e < ef)
 321         ef = e;
 322     if (ef)
 323     {
 324         wrbuf_write(s->wrbuf, subfield, ef - subfield);
 325         if ((subfield = find_subfield(field, 'b')))
 326         {
 327             ef = index(subfield, '\n');
 328             if ((e = index(subfield, '\t')) && e < ef)
 329                 ef = e;
 330             if (ef)
 331             {
 332                 wrbuf_putc(s->wrbuf, ' ');
 333                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 334             }
 335         }
 336     }
 337     if ((field = find_field(rec, "100")))
 338     {
 339         if ((subfield = find_subfield(field, 'a')))
 340         {
 341             ef = index(subfield, '\n');
 342             if ((e = index(subfield, '\t')) && e < ef)
 343                 ef = e;
 344             if (ef)
 345             {
 346                 wrbuf_puts(s->wrbuf, ", by ");
 347                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 348             }
 349         }
 350     }
 351     wrbuf_putc(s->wrbuf, '\0');
 352     obuf = (unsigned char*) nmem_strdup(s->nmem, wrbuf_buf(s->wrbuf));
 353     for (p = obuf; *p; p++)
 354         if (*p == '&' || *p == '<' || *p > 122 || *p < ' ')
 355             *p = ' ';
 356     return (char*) obuf;
 357 }
 358
 359 // Extract 245 $a $b 100 $a
 360 char *extract_mergekey(struct session *s, const char *rec)
 361 {
 362     const char *field, *subfield;
 363     char *e, *ef;
 364     char *out, *p, *pout;
 365
 366     wrbuf_rewind(s->wrbuf);
 367
 368     if (!(field = find_field(rec, "245")))
 369         return 0;
 370     if (!(subfield = find_subfield(field, 'a')))
 371         return 0;
 372     ef = index(subfield, '\n');
 373     if ((e = index(subfield, '\t')) && e < ef)
 374         ef = e;
 375     if (ef)
 376     {
 377         wrbuf_write(s->wrbuf, subfield, ef - subfield);
 378         if ((subfield = find_subfield(field, 'b')))
 379         {
 380             ef = index(subfield, '\n');
 381             if ((e = index(subfield, '\t')) && e < ef)
 382                 ef = e;
 383             if (ef)
 384             {
 385                 wrbuf_puts(s->wrbuf, " field ");
 386                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 387             }
 388         }
 389     }
 390     if ((field = find_field(rec, "100")))
 391     {
 392         if ((subfield = find_subfield(field, 'a')))
 393         {
 394             ef = index(subfield, '\n');
 395             if ((e = index(subfield, '\t')) && e < ef)
 396                 ef = e;
 397             if (ef)
 398             {
 399                 wrbuf_puts(s->wrbuf, " field ");
 400                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 401             }
 402         }
 403     }
 404     wrbuf_putc(s->wrbuf, '\0');
 405     p = wrbuf_buf(s->wrbuf);
 406     out = pout = nmem_malloc(s->nmem, strlen(p) + 1);
 407
 408     while (*p)
 409     {
 410         while (isalnum(*p))
 411             *(pout++) = tolower(*(p++));
 412         while (*p && !isalnum(*p))
 413             p++;
 414         *(pout++) = ' ';
 415     }
 416     if (out != pout)
 417         *(--pout) = '\0';
 418
 419     return out;
 420 }
 421
 422 #ifdef RECHEAP
 423 static void push_record(struct session *s, struct record *r)
 424 {
 425     int p;
 426     assert(s->recheap_max + 1 < s->recheap_size);
 427
 428     s->recheap[p = ++s->recheap_max] = r;
 429     while (p > 0)
 430     {
 431         int parent = (p - 1) >> 1;
 432         if (strcmp(s->recheap[p]->merge_key, s->recheap[parent]->merge_key) < 0)
 433         {
 434             struct record *tmp;
 435             tmp = s->recheap[parent];
 436             s->recheap[parent] = s->recheap[p];
 437             s->recheap[p] = tmp;
 438             p = parent;
 439         }
 440         else
 441             break;
 442     }
 443 }
 444
 445 static struct record *top_record(struct session *s)
 446 {
 447     return s-> recheap_max >= 0 ?  s->recheap[0] : 0;
 448 }
 449
 450 static struct record *pop_record(struct session *s)
 451 {
 452     struct record *res;
 453     int p = 0;
 454     int lastnonleaf = (s->recheap_max - 1) >> 1;
 455
 456     if (s->recheap_max < 0)
 457         return 0;
 458
 459     res = s->recheap[0];
 460
 461     s->recheap[p] = s->recheap[s->recheap_max--];
 462
 463     while (p <= lastnonleaf)
 464     {
 465         int right = (p + 1) << 1;
 466         int left = right - 1;
 467         int min = left;
 468
 469         if (right < s->recheap_max &&
 470                 strcmp(s->recheap[right]->merge_key, s->recheap[left]->merge_key) < 0)
 471             min = right;
 472         if (strcmp(s->recheap[min]->merge_key, s->recheap[p]->merge_key) < 0)
 473         {
 474             struct record *tmp = s->recheap[min];
 475             s->recheap[min] = s->recheap[p];
 476             s->recheap[p] = tmp;
 477             p = min;
 478         }
 479         else
 480             break;
 481     }
 482     return res;
 483 }
 484
 485 // Like pop_record but collapses identical (merge_key) records
 486 // The heap will contain multiple independent matching records and possibly
 487 // one cluster, created the last time the list was scanned
 488 static struct record *pop_mrecord(struct session *s)
 489 {
 490     struct record *this;
 491     struct record *next;
 492
 493     if (!(this = pop_record(s)))
 494         return 0;
 495
 496     // Collapse identical records
 497     while ((next = top_record(s)))
 498     {
 499         struct record *p, *tmpnext;
 500         if (strcmp(this->merge_key, next->merge_key))
 501             break;
 502         // Absorb record (and clustersiblings) into a supercluster
 503         for (p = next; p; p = tmpnext) {
 504             tmpnext = p->next_cluster;
 505             p->next_cluster = this->next_cluster;
 506             this->next_cluster = p;
 507         }
 508
 509         pop_record(s);
 510     }
 511     return this;
 512 }
 513
 514 // Reads records in sort order. Store records in top of heapspace until rewind is called.
 515 static struct record *read_recheap(struct session *s)
 516 {
 517     struct record *r = pop_mrecord(s);
 518
 519     if (r)
 520     {
 521         if (s->recheap_scratch < 0)
 522             s->recheap_scratch = s->recheap_size;
 523         s->recheap[--s->recheap_scratch] = r;
 524     }
 525
 526     return r;
 527 }
 528
 529 // Return records to heap after read
 530 static void rewind_recheap(struct session *s)
 531 {
 532     while (s->recheap_scratch >= 0) {
 533         push_record(s, s->recheap[s->recheap_scratch++]);
 534         if (s->recheap_scratch >= s->recheap_size)
 535             s->recheap_scratch = -1;
 536     }
 537 }
 538
 539 #endif
 540
 541 // FIXME needs to be generalized. Should flexibly generate X lists per search
 542 static void extract_subject(struct session *s, const char *rec)
 543 {
 544     const char *field, *subfield;
 545
 546     while ((field = find_field(rec, "650")))
 547     {
 548         rec = field;
 549         if ((subfield = find_subfield(field, 'a')))
 550         {
 551             char *e, *ef;
 552             char buf[1024];
 553             int len;
 554
 555             ef = index(subfield, '\n');
 556             if (!ef)
 557                 return;
 558             if ((e = index(subfield, '\t')) && e < ef)
 559                 ef = e;
 560             while (ef > subfield && !isalpha(*(ef - 1)) && *(ef - 1) != ')')
 561                 ef--;
 562             len = ef - subfield;
 563             assert(len < 1023);
 564             memcpy(buf, subfield, len);
 565             buf[len] = '\0';
 566             if (*buf)
 567                 termlist_insert(s->termlist, buf);
 568         }
 569     }
 570 }
 571
 572 static void pull_relevance_field(struct session *s, struct record *head, const char *rec,
 573         char *field, int mult)
 574 {
 575     const char *fb;
 576     while ((fb = find_field(rec, field)))
 577     {
 578         char *ffield = strchr(fb, '\t');
 579         if (!ffield)
 580             return;
 581         char *eol = strchr(ffield, '\n');
 582         if (!eol)
 583             return;
 584         relevance_countwords(s->relevance, head, ffield, eol - ffield, mult);
 585         rec = field + 1; // Crude way to cause a loop through repeating fields
 586     }
 587 }
 588
 589 static void pull_relevance_keys(struct session *s, struct record *head,  struct record *rec)
 590 {
 591     relevance_newrec(s->relevance, head);
 592     pull_relevance_field(s, head, rec->buf, "100", 2);
 593     pull_relevance_field(s, head, rec->buf, "245", 4);
 594     //pull_relevance_field(s, head, rec->buf, "530", 1);
 595     pull_relevance_field(s, head, rec->buf, "630", 1);
 596     pull_relevance_field(s, head, rec->buf, "650", 1);
 597     pull_relevance_field(s, head, rec->buf, "700", 1);
 598     relevance_donerecord(s->relevance, head);
 599 }
 600
 601 static struct record *ingest_record(struct client *cl, char *buf, int len)
 602 {
 603     struct session *se = cl->session;
 604     struct record *res;
 605     struct record *head;
 606     const char *recbuf;
 607
 608     wrbuf_rewind(se->wrbuf);
 609     yaz_marc_xml(global_parameters.yaz_marc, YAZ_MARC_LINE);
 610     if (yaz_marc_decode_wrbuf(global_parameters.yaz_marc, buf, len, se->wrbuf) < 0)
 611     {
 612         yaz_log(YLOG_WARN, "Failed to decode MARC record");
 613         return 0;
 614     }
 615     wrbuf_putc(se->wrbuf, '\0');
 616     recbuf = wrbuf_buf(se->wrbuf);
 617
 618     res = nmem_malloc(se->nmem, sizeof(struct record));
 619     res->buf = nmem_strdup(se->nmem, recbuf);
 620
 621     extract_subject(se, res->buf);
 622
 623     res->title = extract_title(se, res->buf);
 624     res->merge_key = extract_mergekey(se, res->buf);
 625     if (!res->merge_key)
 626         return 0;
 627     res->client = cl;
 628     res->next_cluster = 0;
 629     res->target_offset = -1;
 630     res->term_frequency_vec = 0;
 631
 632     head = reclist_insert(se->reclist, res);
 633
 634     pull_relevance_keys(se, head, res);
 635
 636     se->total_records++;
 637
 638     return res;
 639 }
 640
 641 static void ingest_records(struct client *cl, Z_Records *r)
 642 {
 643     struct record *rec;
 644     struct session *s = cl->session;
 645     Z_NamePlusRecordList *rlist;
 646     int i;
 647
 648     if (r->which != Z_Records_DBOSD)
 649         return;
 650     rlist = r->u.databaseOrSurDiagnostics;
 651     for (i = 0; i < rlist->num_records; i++)
 652     {
 653         Z_NamePlusRecord *npr = rlist->records[i];
 654         Z_External *e;
 655         char *buf;
 656         int len;
 657
 658         if (npr->which != Z_NamePlusRecord_databaseRecord)
 659         {
 660             yaz_log(YLOG_WARN, "Unexpected record type, probably diagnostic");
 661             continue;
 662         }
 663         e = npr->u.databaseRecord;
 664         if (e->which != Z_External_octet)
 665         {
 666             yaz_log(YLOG_WARN, "Unexpected external branch, probably BER");
 667             continue;
 668         }
 669         buf = (char*) e->u.octet_aligned->buf;
 670         len = e->u.octet_aligned->len;
 671
 672         rec = ingest_record(cl, buf, len);
 673         if (!rec)
 674             continue;
 675     }
 676     if (s->watchlist[SESSION_WATCH_RECORDS].fun && rlist->num_records)
 677         session_alert_watch(s, SESSION_WATCH_RECORDS);
 678 }
 679
 680 xsltStylesheetPtr load_stylesheet(const char *fname)
 681 {
 682     xsltStylesheetPtr ret;
 683     if (!(ret = xsltParseStylesheetFile((const xmlChar *) fname)))
 684     {
 685         yaz_log(YLOG_FATAL|YLOG_ERRNO, "Failed to load stylesheet %s", fname);
 686         exit(1);
 687     }
 688     return ret;
 689 }
 690
 691 static void do_presentResponse(IOCHAN i, Z_APDU *a)
 692 {
 693     struct connection *co = iochan_getdata(i);
 694     struct client *cl = co->client;
 695     Z_PresentResponse *r = a->u.presentResponse;
 696
 697     if (r->records) {
 698         Z_Records *recs = r->records;
 699         if (recs->which == Z_Records_NSD)
 700         {
 701             yaz_log(YLOG_WARN, "Non-surrogate diagnostic");
 702             cl->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 703             cl->state = Client_Error;
 704         }
 705     }
 706
 707     if (!*r->presentStatus && cl->state != Client_Error)
 708     {
 709         yaz_log(YLOG_DEBUG, "Good Present response");
 710         cl->records += *r->numberOfRecordsReturned;
 711         ingest_records(cl, r->records);
 712         cl->state = Client_Idle;
 713     }
 714     else if (*r->presentStatus)
 715     {
 716         yaz_log(YLOG_WARN, "Bad Present response");
 717         cl->state = Client_Error;
 718     }
 719 }
 720
 721 static void handler(IOCHAN i, int event)
 722 {
 723     struct connection *co = iochan_getdata(i);
 724     struct client *cl = co->client;
 725     struct session *se = 0;
 726
 727     if (cl)
 728         se = cl->session;
 729     else
 730     {
 731         yaz_log(YLOG_WARN, "Destroying orphan connection");
 732         connection_destroy(co);
 733         return;
 734     }
 735
 736     if (co->state == Conn_Connecting && event & EVENT_OUTPUT)
 737     {
 738         int errcode;
 739         socklen_t errlen = sizeof(errcode);
 740
 741         if (getsockopt(cs_fileno(co->link), SOL_SOCKET, SO_ERROR, &errcode,
 742             &errlen) < 0 || errcode != 0)
 743         {
 744             client_fatal(cl);
 745             return;
 746         }
 747         else
 748         {
 749             yaz_log(YLOG_DEBUG, "Connect OK");
 750             co->state = Conn_Open;
 751             if (cl)
 752                 cl->state = Client_Connected;
 753         }
 754     }
 755
 756     else if (event & EVENT_INPUT)
 757     {
 758         int len = cs_get(co->link, &co->ibuf, &co->ibufsize);
 759
 760         if (len < 0)
 761         {
 762             yaz_log(YLOG_WARN|YLOG_ERRNO, "Error reading from Z server");
 763             connection_destroy(co);
 764             return;
 765         }
 766         else if (len == 0)
 767         {
 768             yaz_log(YLOG_WARN, "EOF reading from Z server");
 769             connection_destroy(co);
 770             return;
 771         }
 772         else if (len > 1) // We discard input if we have no connection
 773         {
 774             co->state = Conn_Open;
 775
 776             if (cl && (cl->requestid == se->requestid || cl->state == Client_Initializing))
 777             {
 778                 Z_APDU *a;
 779
 780                 odr_reset(global_parameters.odr_in);
 781                 odr_setbuf(global_parameters.odr_in, co->ibuf, len, 0);
 782                 if (!z_APDU(global_parameters.odr_in, &a, 0, 0))
 783                 {
 784                     client_fatal(cl);
 785                     return;
 786                 }
 787                 switch (a->which)
 788                 {
 789                     case Z_APDU_initResponse:
 790                         do_initResponse(i, a);
 791                         break;
 792                     case Z_APDU_searchResponse:
 793                         do_searchResponse(i, a);
 794                         break;
 795                     case Z_APDU_presentResponse:
 796                         do_presentResponse(i, a);
 797                         break;
 798                     default:
 799                         yaz_log(YLOG_WARN, "Unexpected result from server");
 800                         client_fatal(cl);
 801                         return;
 802                 }
 803                 // We aren't expecting staggered output from target
 804                 // if (cs_more(t->link))
 805                 //    iochan_setevent(i, EVENT_INPUT);
 806             }
 807             else  // we throw away response and go to idle mode
 808             {
 809                 yaz_log(YLOG_DEBUG, "Ignoring result of expired operation");
 810                 cl->state = Client_Idle;
 811             }
 812         }
 813         /* if len==1 we do nothing but wait for more input */
 814     }
 815
 816     if (cl->state == Client_Connected) {
 817         send_init(i);
 818     }
 819
 820     if (cl->state == Client_Idle)
 821     {
 822         if (cl->requestid != se->requestid && *se->query) {
 823             send_search(i);
 824         }
 825         else if (cl->hits > 0 && cl->records < global_parameters.toget &&
 826             cl->records < cl->hits) {
 827             send_present(i);
 828         }
 829     }
 830 }
 831
 832 // Disassociate connection from client
 833 static void connection_release(struct connection *co)
 834 {
 835     struct client *cl = co->client;
 836
 837     yaz_log(YLOG_DEBUG, "Connection release %s", co->host->hostport);
 838     if (!cl)
 839         return;
 840     cl->connection = 0;
 841     co->client = 0;
 842 }
 843
 844 // Close connection and recycle structure
 845 static void connection_destroy(struct connection *co)
 846 {
 847     struct host *h = co->host;
 848     cs_close(co->link);
 849     iochan_destroy(co->iochan);
 850
 851     yaz_log(YLOG_DEBUG, "Connection destroy %s", co->host->hostport);
 852     if (h->connections == co)
 853         h->connections = co->next;
 854     else
 855     {
 856         struct connection *pco;
 857         for (pco = h->connections; pco && pco->next != co; pco = pco->next)
 858             ;
 859         if (pco)
 860             pco->next = co->next;
 861         else
 862             abort();
 863     }
 864     if (co->client)
 865     {
 866         if (co->client->state != Client_Idle)
 867             co->client->state = Client_Disconnected;
 868         co->client->connection = 0;
 869     }
 870     co->next = connection_freelist;
 871     connection_freelist = co;
 872 }
 873
 874 // Creates a new connection for client, associated with the host of
 875 // client's database
 876 static struct connection *connection_create(struct client *cl)
 877 {
 878     struct connection *new;
 879     COMSTACK link;
 880     int res;
 881     void *addr;
 882
 883     yaz_log(YLOG_DEBUG, "Connection create %s", cl->database->url);
 884     if (!(link = cs_create(tcpip_type, 0, PROTO_Z3950)))
 885     {
 886         yaz_log(YLOG_FATAL|YLOG_ERRNO, "Failed to create comstack");
 887         exit(1);
 888     }
 889
 890     if (!(addr = cs_straddr(link, cl->database->host->ipport)))
 891     {
 892         yaz_log(YLOG_WARN|YLOG_ERRNO, "Lookup of IP address failed?");
 893         return 0;
 894     }
 895
 896     res = cs_connect(link, addr);
 897     if (res < 0)
 898     {
 899         yaz_log(YLOG_WARN|YLOG_ERRNO, "cs_connect %s", cl->database->url);
 900         return 0;
 901     }
 902
 903     if ((new = connection_freelist))
 904         connection_freelist = new->next;
 905     else
 906     {
 907         new = xmalloc(sizeof (struct connection));
 908         new->ibuf = 0;
 909         new->ibufsize = 0;
 910     }
 911     new->state = Conn_Connecting;
 912     new->host = cl->database->host;
 913     new->next = new->host->connections;
 914     new->host->connections = new;
 915     new->client = cl;
 916     cl->connection = new;
 917     new->link = link;
 918
 919     new->iochan = iochan_create(cs_fileno(link), handler, 0);
 920     iochan_setdata(new->iochan, new);
 921     new->iochan->next = channel_list;
 922     channel_list = new->iochan;
 923     return new;
 924 }
 925
 926 // Close connection and set state to error
 927 static void client_fatal(struct client *cl)
 928 {
 929     yaz_log(YLOG_WARN, "Fatal error from %s", cl->database->url);
 930     connection_destroy(cl->connection);
 931     cl->state = Client_Error;
 932 }
 933
 934 // Ensure that client has a connection associated
 935 static int client_prep_connection(struct client *cl)
 936 {
 937     struct connection *co;
 938     struct session *se = cl->session;
 939     struct host *host = cl->database->host;
 940
 941     co = cl->connection;
 942
 943     yaz_log(YLOG_DEBUG, "Client prep %s", cl->database->url);
 944
 945     if (!co)
 946     {
 947         // See if someone else has an idle connection
 948         // We should look at timestamps here to select the longest-idle connection
 949         for (co = host->connections; co; co = co->next)
 950             if (co->state == Conn_Open && (!co->client || co->client->session != se))
 951                 break;
 952         if (co)
 953         {
 954             connection_release(co);
 955             cl->connection = co;
 956             co->client = cl;
 957         }
 958         else
 959             co = connection_create(cl);
 960     }
 961     if (co)
 962     {
 963         if (co->state == Conn_Connecting)
 964             cl->state = Client_Connecting;
 965         else if (co->state == Conn_Open)
 966         {
 967             if (cl->state == Client_Error || cl->state == Client_Disconnected)
 968                 cl->state = Client_Idle;
 969         }
 970         iochan_setflag(co->iochan, EVENT_OUTPUT);
 971         return 1;
 972     }
 973     else
 974         return 0;
 975 }
 976
 977 void load_simpletargets(const char *fn)
 978 {
 979     FILE *f = fopen(fn, "r");
 980     char line[256];
 981
 982     if (!f)
 983     {
 984         yaz_log(YLOG_WARN|YLOG_ERRNO, "open %s", fn);
 985         exit(1);
 986     }
 987
 988     while (fgets(line, 255, f))
 989     {
 990         char *url, *db;
 991         struct host *host;
 992         struct database *database;
 993
 994         if (strncmp(line, "target ", 7))
 995             continue;
 996         url = line + 7;
 997         url[strlen(url) - 1] = '\0';
 998         yaz_log(YLOG_DEBUG, "Target: %s", url);
 999         if ((db = strchr(url, '/')))
1000             *(db++) = '\0';
1001         else
1002             db = "Default";
1003
1004         for (host = hosts; host; host = host->next)
1005             if (!strcmp(url, host->hostport))
1006                 break;
1007         if (!host)
1008         {
1009             struct addrinfo *addrinfo, hints;
1010             char *port;
1011             char ipport[128];
1012             unsigned char addrbuf[4];
1013             int res;
1014
1015             host = xmalloc(sizeof(struct host));
1016             host->hostport = xstrdup(url);
1017             host->connections = 0;
1018
1019             if ((port = strchr(url, ':')))
1020                 *(port++) = '\0';
1021             else
1022                 port = "210";
1023
1024             hints.ai_flags = 0;
1025             hints.ai_family = PF_INET;
1026             hints.ai_socktype = SOCK_STREAM;
1027             hints.ai_protocol = IPPROTO_TCP;
1028             hints.ai_addrlen = 0;
1029             hints.ai_addr = 0;
1030             hints.ai_canonname = 0;
1031             hints.ai_next = 0;
1032             // This is not robust code. It assumes that getaddrinfo returns AF_INET
1033             // address.
1034             if ((res = getaddrinfo(url, port, &hints, &addrinfo)))
1035             {
1036                 yaz_log(YLOG_WARN, "Failed to resolve %s: %s", url, gai_strerror(res));
1037                 xfree(host->hostport);
1038                 xfree(host);
1039                 continue;
1040             }
1041             assert(addrinfo->ai_family == PF_INET);
1042             memcpy(addrbuf, &((struct sockaddr_in*)addrinfo->ai_addr)->sin_addr.s_addr, 4);
1043             sprintf(ipport, "%hhd.%hhd.%hhd.%hhd:%s",
1044                     addrbuf[0], addrbuf[1], addrbuf[2], addrbuf[3], port);
1045             host->ipport = xstrdup(ipport);
1046             freeaddrinfo(addrinfo);
1047             host->next = hosts;
1048             hosts = host;
1049         }
1050         database = xmalloc(sizeof(struct database));
1051         database->host = host;
1052         database->url = xmalloc(strlen(url) + strlen(db) + 2);
1053         strcpy(database->url, url);
1054         strcat(database->url, "/");
1055         strcat(database->url, db);
1056
1057         database->databases = xmalloc(2 * sizeof(char *));
1058         database->databases[0] = xstrdup(db);
1059         database->databases[1] = 0;
1060         database->errors = 0;
1061         database->next = databases;
1062         databases = database;
1063
1064     }
1065     fclose(f);
1066 }
1067
1068 static void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
1069 {
1070     switch (n->kind)
1071     {
1072         case CCL_RPN_AND:
1073         case CCL_RPN_OR:
1074         case CCL_RPN_NOT:
1075         case CCL_RPN_PROX:
1076             pull_terms(nmem, n->u.p[0], termlist, num);
1077             pull_terms(nmem, n->u.p[1], termlist, num);
1078             break;
1079         case CCL_RPN_TERM:
1080             termlist[(*num)++] = nmem_strdup(nmem, n->u.t.term);
1081             break;
1082         default: // NOOP
1083             break;
1084     }
1085 }
1086
1087 // Extract terms from query into null-terminated termlist
1088 static int extract_terms(NMEM nmem, char *query, char **termlist)
1089 {
1090     int error, pos;
1091     struct ccl_rpn_node *n;
1092     int num = 0;
1093
1094     n = ccl_find_str(global_parameters.ccl_filter, query, &error, &pos);
1095     if (!n)
1096         return -1;
1097     pull_terms(nmem, n, termlist, &num);
1098     termlist[num] = 0;
1099     ccl_rpn_delete(n);
1100     return 0;
1101 }
1102
1103 static struct client *client_create(void)
1104 {
1105     struct client *r;
1106     if (client_freelist)
1107     {
1108         r = client_freelist;
1109         client_freelist = client_freelist->next;
1110     }
1111     else
1112         r = xmalloc(sizeof(struct client));
1113     r->database = 0;
1114     r->connection = 0;
1115     r->session = 0;
1116     r->hits = 0;
1117     r->records = 0;
1118     r->setno = 0;
1119     r->requestid = -1;
1120     r->diagnostic = 0;
1121     r->state = Client_Disconnected;
1122     r->next = 0;
1123     return r;
1124 }
1125
1126 void client_destroy(struct client *c)
1127 {
1128     struct session *se = c->session;
1129     if (c == se->clients)
1130         se->clients = c->next;
1131     else
1132     {
1133         struct client *cc;
1134         for (cc = se->clients; cc && cc->next != c; cc = cc->next)
1135             ;
1136         if (cc)
1137             cc->next = c->next;
1138     }
1139     if (c->connection)
1140         connection_release(c->connection);
1141     c->next = client_freelist;
1142     client_freelist = c;
1143 }
1144
1145 void session_set_watch(struct session *s, int what, session_watchfun fun, void *data)
1146 {
1147     s->watchlist[what].fun = fun;
1148     s->watchlist[what].data = data;
1149 }
1150
1151 void session_alert_watch(struct session *s, int what)
1152 {
1153     if (!s->watchlist[what].fun)
1154         return;
1155     (*s->watchlist[what].fun)(s->watchlist[what].data);
1156     s->watchlist[what].fun = 0;
1157     s->watchlist[what].data = 0;
1158 }
1159
1160 // This should be extended with parameters to control selection criteria
1161 // Associates a set of clients with a session;
1162 int select_targets(struct session *se)
1163 {
1164     struct database *db;
1165     int c = 0;
1166
1167     while (se->clients)
1168         client_destroy(se->clients);
1169     for (db = databases; db; db = db->next)
1170     {
1171         struct client *cl = client_create();
1172         cl->database = db;
1173         cl->session = se;
1174         cl->next = se->clients;
1175         se->clients = cl;
1176         c++;
1177     }
1178     return c;
1179 }
1180
1181 char *search(struct session *se, char *query)
1182 {
1183     int live_channels = 0;
1184     struct client *cl;
1185
1186     yaz_log(YLOG_DEBUG, "Search");
1187
1188     strcpy(se->query, query);
1189     se->requestid++;
1190     nmem_reset(se->nmem);
1191     for (cl = se->clients; cl; cl = cl->next)
1192     {
1193         cl->hits = -1;
1194         cl->records = 0;
1195         cl->diagnostic = 0;
1196
1197         if (client_prep_connection(cl))
1198             live_channels++;
1199     }
1200     if (live_channels)
1201     {
1202         char *p[512];
1203         int maxrecs = live_channels * global_parameters.toget;
1204         se->termlist = termlist_create(se->nmem, maxrecs, 15);
1205         se->reclist = reclist_create(se->nmem, maxrecs);
1206         extract_terms(se->nmem, query, p);
1207         se->relevance = relevance_create(se->nmem, (const char **) p, maxrecs);
1208         se->total_records = se->total_hits = 0;
1209     }
1210     else
1211         return "NOTARGETS";
1212
1213     return 0;
1214 }
1215
1216 void destroy_session(struct session *s)
1217 {
1218     yaz_log(YLOG_LOG, "Destroying session");
1219     while (s->clients)
1220         client_destroy(s->clients);
1221     nmem_destroy(s->nmem);
1222     wrbuf_free(s->wrbuf, 1);
1223 }
1224
1225 struct session *new_session()
1226 {
1227     int i;
1228     struct session *session = xmalloc(sizeof(*session));
1229
1230     yaz_log(YLOG_DEBUG, "New pazpar2 session");
1231
1232     session->total_hits = 0;
1233     session->total_records = 0;
1234     session->termlist = 0;
1235     session->reclist = 0;
1236     session->requestid = -1;
1237     session->clients = 0;
1238     session->query[0] = '\0';
1239     session->nmem = nmem_create();
1240     session->wrbuf = wrbuf_alloc();
1241     for (i = 0; i <= SESSION_WATCH_MAX; i++)
1242     {
1243         session->watchlist[i].data = 0;
1244         session->watchlist[i].fun = 0;
1245     }
1246
1247     select_targets(session);
1248
1249     return session;
1250 }
1251
1252 struct hitsbytarget *hitsbytarget(struct session *se, int *count)
1253 {
1254     static struct hitsbytarget res[1000]; // FIXME MM
1255     struct client *cl;
1256
1257     *count = 0;
1258     for (cl = se->clients; cl; cl = cl->next)
1259     {
1260         strcpy(res[*count].id, cl->database->host->hostport);
1261         res[*count].hits = cl->hits;
1262         res[*count].records = cl->records;
1263         res[*count].diagnostic = cl->diagnostic;
1264         res[*count].state = client_states[cl->state];
1265         res[*count].connected  = cl->connection ? 1 : 0;
1266         (*count)++;
1267     }
1268
1269     return res;
1270 }
1271
1272 struct termlist_score **termlist(struct session *s, int *num)
1273 {
1274     return termlist_highscore(s->termlist, num);
1275 }
1276
1277 #ifdef REPORT_NMEM
1278 // conditional compilation by SH: This lead to a warning with currently installed
1279 // YAZ header files on us1
1280 void report_nmem_stats(void)
1281 {
1282     size_t in_use, is_free;
1283
1284     nmem_get_memory_in_use(&in_use);
1285     nmem_get_memory_free(&is_free);
1286
1287     yaz_log(YLOG_LOG, "nmem stat: use=%ld free=%ld",
1288             (long) in_use, (long) is_free);
1289 }
1290 #endif
1291
1292 struct record **show(struct session *s, int start, int *num, int *total,
1293                      int *sumhits, NMEM nmem_show)
1294 {
1295     struct record **recs = nmem_malloc(nmem_show, *num
1296                                        * sizeof(struct record *));
1297     int i;
1298
1299     relevance_prepare_read(s->relevance, s->reclist);
1300
1301     *total = s->reclist->num_records;
1302     *sumhits = s->total_hits;
1303
1304     for (i = 0; i < start; i++)
1305         if (!reclist_read_record(s->reclist))
1306         {
1307             *num = 0;
1308             return 0;
1309         }
1310
1311     for (i = 0; i < *num; i++)
1312     {
1313         struct record *r = reclist_read_record(s->reclist);
1314         if (!r)
1315         {
1316             *num = i;
1317             break;
1318         }
1319         recs[i] = r;
1320     }
1321     return recs;
1322 }
1323
1324 void statistics(struct session *se, struct statistics *stat)
1325 {
1326     struct client *cl;
1327     int count = 0;
1328
1329     bzero(stat, sizeof(*stat));
1330     for (cl = se->clients; cl; cl = cl->next)
1331     {
1332         if (!cl->connection)
1333             stat->num_no_connection++;
1334         switch (cl->state)
1335         {
1336             case Client_Connecting: stat->num_connecting++; break;
1337             case Client_Initializing: stat->num_initializing++; break;
1338             case Client_Searching: stat->num_searching++; break;
1339             case Client_Presenting: stat->num_presenting++; break;
1340             case Client_Idle: stat->num_idle++; break;
1341             case Client_Failed: stat->num_failed++; break;
1342             case Client_Error: stat->num_error++; break;
1343             default: break;
1344         }
1345         count++;
1346     }
1347     stat->num_hits = se->total_hits;
1348     stat->num_records = se->total_records;
1349
1350     stat->num_clients = count;
1351 }
1352
1353 static CCL_bibset load_cclfile(const char *fn)
1354 {
1355     CCL_bibset res = ccl_qual_mk();
1356     if (ccl_qual_fname(res, fn) < 0)
1357     {
1358         yaz_log(YLOG_FATAL|YLOG_ERRNO, "%s", fn);
1359         exit(1);
1360     }
1361     return res;
1362 }
1363
1364 int main(int argc, char **argv)
1365 {
1366     int ret;
1367     char *arg;
1368     int setport = 0;
1369
1370     if (signal(SIGPIPE, SIG_IGN) < 0)
1371         yaz_log(YLOG_WARN|YLOG_ERRNO, "signal");
1372
1373     yaz_log_init(YLOG_DEFAULT_LEVEL, "pazpar2", 0);
1374
1375     while ((ret = options("x:c:h:p:C:s:", argv, argc, &arg)) != -2)
1376     {
1377         switch (ret) {
1378             case 'c':
1379                 command_init(atoi(arg));
1380                 setport++;
1381                 break;
1382             case 'h':
1383                 http_init(arg);
1384                 setport++;
1385                 break;
1386             case 'C':
1387                 global_parameters.ccl_filter = load_cclfile(arg);
1388                 break;
1389             case 'p':
1390                 http_set_proxyaddr(arg);
1391                 break;
1392             case 's':
1393                 load_simpletargets(arg);
1394                 break;
1395             case 'x':
1396                 global_parameters.xsl = load_stylesheet(arg);
1397                 break;
1398             default:
1399                 fprintf(stderr, "Usage: pazpar2\n"
1400                         "    -h [host:]port          (REST protocol listener)\n"
1401                         "    -c cmdport              (telnet-style)\n"
1402                         "    -C cclconfig\n"
1403                         "    -s simpletargetfile\n"
1404                         "    -p hostname[:portno]    (HTTP proxy)\n");
1405                 exit(1);
1406         }
1407     }
1408
1409     if (!setport)
1410     {
1411         fprintf(stderr, "Set command port with -h or -c\n");
1412         exit(1);
1413     }
1414
1415     if (!global_parameters.xsl)
1416         global_parameters.xsl = load_stylesheet("../etc/default.xsl");
1417     global_parameters.ccl_filter = load_cclfile("../etc/default.bib");
1418     global_parameters.yaz_marc = yaz_marc_create();
1419     yaz_marc_subfield_str(global_parameters.yaz_marc, "\t");
1420     global_parameters.odr_in = odr_createmem(ODR_DECODE);
1421     global_parameters.odr_out = odr_createmem(ODR_ENCODE);
1422
1423     event_loop(&channel_list);
1424
1425     return 0;
1426 }
1427
1428 /*
1429  * Local variables:
1430  * c-basic-offset: 4
1431  * indent-tabs-mode: nil
1432  * End:
1433  * vim: shiftwidth=4 tabstop=8 expandtab
1434  */