pazpar2.c

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