pazpar2.c

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