pazpar2.c

   1 /* $Id: pazpar2.c,v 1.7 2006-11-27 19:44:26 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 <signal.h>
  10 #include <ctype.h>
  11 #include <assert.h>
  12
  13 #include <yaz/comstack.h>
  14 #include <yaz/tcpip.h>
  15 #include <yaz/proto.h>
  16 #include <yaz/readconf.h>
  17 #include <yaz/pquery.h>
  18 #include <yaz/yaz-util.h>
  19 #include <yaz/ccl.h>
  20 #include <yaz/yaz-ccl.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_DATABASES 512
  32 #define MAX_CHUNK 10
  33
  34 static void target_destroy(IOCHAN i);
  35
  36 struct target
  37 {
  38     struct session *session;
  39     char fullname[256];
  40     char hostport[128];
  41     char *ibuf;
  42     int ibufsize;
  43     char databases[MAX_DATABASES][128];
  44     COMSTACK link;
  45     ODR odr_in, odr_out;
  46     struct target *next;
  47     void *addr;
  48     int hits;
  49     int records;
  50     int setno;
  51     int requestid;                              // ID of current outstanding request
  52     int diagnostic;
  53     IOCHAN iochan;
  54     enum target_state
  55     {
  56         No_connection,
  57         Connecting,
  58         Connected,
  59         Initializing,
  60         Searching,
  61         Presenting,
  62         Error,
  63         Idle,
  64         Stopped,
  65         Failed
  66     } state;
  67 };
  68
  69 static char *state_strings[] = {
  70     "No_connection",
  71     "Connecting",
  72     "Connected",
  73     "Initializing",
  74     "Searching",
  75     "Presenting",
  76     "Error",
  77     "Idle",
  78     "Failed"
  79 };
  80
  81
  82 IOCHAN channel_list = 0;
  83
  84 static struct parameters {
  85     int timeout;                /* operations timeout, in seconds */
  86     char implementationId[128];
  87     char implementationName[128];
  88     char implementationVersion[128];
  89     struct timeval base_time;
  90     int toget;
  91     int chunk;
  92     CCL_bibset ccl_filter;
  93 } global_parameters =
  94 {
  95     30,
  96     "81",
  97     "Index Data PazPar2 (MasterKey)",
  98     PAZPAR2_VERSION,
  99     {0,0},
 100     100,
 101     MAX_CHUNK,
 102     0
 103 };
 104
 105
 106 static int send_apdu(struct target *t, Z_APDU *a)
 107 {
 108     char *buf;
 109     int len, r;
 110
 111     if (!z_APDU(t->odr_out, &a, 0, 0))
 112     {
 113         odr_perror(t->odr_out, "Encoding APDU");
 114         abort();
 115     }
 116     buf = odr_getbuf(t->odr_out, &len, 0);
 117     r = cs_put(t->link, buf, len);
 118     if (r < 0)
 119     {
 120         yaz_log(YLOG_WARN, "cs_put: %s", cs_errmsg(cs_errno(t->link)));
 121         return -1;
 122     }
 123     else if (r == 1)
 124     {
 125         fprintf(stderr, "cs_put incomplete (ParaZ does not handle that)\n");
 126     }
 127     odr_reset(t->odr_out); /* release the APDU structure  */
 128     return 0;
 129 }
 130
 131
 132 static void send_init(IOCHAN i)
 133 {
 134     struct target *t = iochan_getdata(i);
 135     Z_APDU *a = zget_APDU(t->odr_out, Z_APDU_initRequest);
 136
 137     a->u.initRequest->implementationId = global_parameters.implementationId;
 138     a->u.initRequest->implementationName = global_parameters.implementationName;
 139     a->u.initRequest->implementationVersion =
 140         global_parameters.implementationVersion;
 141     ODR_MASK_SET(a->u.initRequest->options, Z_Options_search);
 142     ODR_MASK_SET(a->u.initRequest->options, Z_Options_present);
 143     ODR_MASK_SET(a->u.initRequest->options, Z_Options_namedResultSets);
 144
 145     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_1);
 146     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_2);
 147     ODR_MASK_SET(a->u.initRequest->protocolVersion, Z_ProtocolVersion_3);
 148     if (send_apdu(t, a) >= 0)
 149     {
 150         iochan_setflags(i, EVENT_INPUT);
 151         t->state = Initializing;
 152     }
 153     else
 154         target_destroy(i);
 155 }
 156
 157 static void send_search(IOCHAN i)
 158 {
 159     struct target *t = iochan_getdata(i);
 160     struct session *s = t->session;
 161     Z_APDU *a = zget_APDU(t->odr_out, Z_APDU_searchRequest);
 162     int ndb, cerror, cpos;
 163     char **databaselist;
 164     Z_Query *zquery;
 165     struct ccl_rpn_node *cn;
 166
 167     yaz_log(YLOG_DEBUG, "Sending search");
 168
 169     cn = ccl_find_str(global_parameters.ccl_filter, s->query, &cerror, &cpos);
 170     if (!cn)
 171         return;
 172     a->u.searchRequest->query = zquery = odr_malloc(t->odr_out, sizeof(Z_Query));
 173     zquery->which = Z_Query_type_1;
 174     zquery->u.type_1 = ccl_rpn_query(t->odr_out, cn);
 175     ccl_rpn_delete(cn);
 176
 177     for (ndb = 0; *t->databases[ndb]; ndb++)
 178         ;
 179     databaselist = odr_malloc(t->odr_out, sizeof(char*) * ndb);
 180     for (ndb = 0; *t->databases[ndb]; ndb++)
 181         databaselist[ndb] = t->databases[ndb];
 182
 183     a->u.searchRequest->resultSetName = "Default";
 184     a->u.searchRequest->databaseNames = databaselist;
 185     a->u.searchRequest->num_databaseNames = ndb;
 186
 187     if (send_apdu(t, a) >= 0)
 188     {
 189         iochan_setflags(i, EVENT_INPUT);
 190         t->state = Searching;
 191         t->requestid = s->requestid;
 192     }
 193     else
 194     {
 195         target_destroy(i);
 196         return;
 197     }
 198     odr_reset(t->odr_out);
 199 }
 200
 201 static void send_present(IOCHAN i)
 202 {
 203     struct target *t = iochan_getdata(i);
 204     Z_APDU *a = zget_APDU(t->odr_out, Z_APDU_presentRequest);
 205     int toget;
 206     int start = t->records + 1;
 207
 208     toget = global_parameters.chunk;
 209     if (toget > t->hits - t->records)
 210         toget = t->hits - t->records;
 211
 212     yaz_log(YLOG_DEBUG, "Trying to present %d records\n", toget);
 213
 214     a->u.presentRequest->resultSetStartPoint = &start;
 215     a->u.presentRequest->numberOfRecordsRequested = &toget;
 216
 217     a->u.presentRequest->resultSetId = "Default";
 218
 219     a->u.presentRequest->preferredRecordSyntax = yaz_oidval_to_z3950oid(t->odr_out,
 220             CLASS_RECSYN, VAL_USMARC);
 221
 222     if (send_apdu(t, a) >= 0)
 223     {
 224         iochan_setflags(i, EVENT_INPUT);
 225         t->state = Presenting;
 226     }
 227     else
 228     {
 229         target_destroy(i);
 230         return;
 231     }
 232     odr_reset(t->odr_out);
 233 }
 234
 235 static void do_initResponse(IOCHAN i, Z_APDU *a)
 236 {
 237     struct target *t = iochan_getdata(i);
 238     Z_InitResponse *r = a->u.initResponse;
 239
 240     yaz_log(YLOG_DEBUG, "Received init response");
 241
 242     if (*r->result)
 243     {
 244         t->state = Idle;
 245     }
 246     else
 247         target_destroy(i);
 248 }
 249
 250 static void do_searchResponse(IOCHAN i, Z_APDU *a)
 251 {
 252     struct target *t = iochan_getdata(i);
 253     Z_SearchResponse *r = a->u.searchResponse;
 254
 255     yaz_log(YLOG_DEBUG, "Searchresponse (status=%d)", *r->searchStatus);
 256
 257     if (*r->searchStatus)
 258     {
 259         t->hits = *r->resultCount;
 260         t->state = Idle;
 261         t->session->total_hits += t->hits;
 262     }
 263     else
 264     {          /*"FAILED"*/
 265         t->hits = 0;
 266         t->state = Failed;
 267         if (r->records) {
 268             Z_Records *recs = r->records;
 269             if (recs->which == Z_Records_NSD)
 270             {
 271                 yaz_log(YLOG_WARN, "Non-surrogate diagnostic");
 272                 t->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 273                 t->state = Error;
 274             }
 275         }
 276     }
 277 }
 278
 279 const char *find_field(const char *rec, const char *field)
 280 {
 281     const char *line = rec;
 282
 283     while (*line)
 284     {
 285         const char *eol;
 286
 287         if (!strncmp(line, field, 3) && line[3] == ' ')
 288             return line;
 289         while (*line && *line != '\n')
 290             line++;
 291         if (!(eol = strchr(line, '\n')))
 292             return 0;
 293         line = eol + 1;
 294     }
 295     return 0;
 296 }
 297
 298 const char *find_subfield(const char *field, char subfield)
 299 {
 300     const char *p = field;
 301
 302     while (*p && *p != '\n')
 303     {
 304         while (*p != '\n' && *p != '\t')
 305             p++;
 306         if (*p == '\t' && *(++p) == subfield) {
 307             if (*(++p) == ' ')
 308             {
 309                 while (isspace(*p))
 310                     p++;
 311                 return p;
 312             }
 313         }
 314     }
 315     return 0;
 316 }
 317
 318 // Extract 245 $a $b 100 $a
 319 char *extract_title(struct session *s, const char *rec)
 320 {
 321     const char *field, *subfield;
 322     char *e, *ef;
 323     unsigned char *obuf, *p;
 324
 325     wrbuf_rewind(s->wrbuf);
 326
 327     if (!(field = find_field(rec, "245")))
 328         return 0;
 329     if (!(subfield = find_subfield(field, 'a')))
 330         return 0;
 331     ef = index(subfield, '\n');
 332     if ((e = index(subfield, '\t')) && e < ef)
 333         ef = e;
 334     if (ef)
 335     {
 336         wrbuf_write(s->wrbuf, subfield, ef - subfield);
 337         if ((subfield = find_subfield(field, 'b')))
 338         {
 339             ef = index(subfield, '\n');
 340             if ((e = index(subfield, '\t')) && e < ef)
 341                 ef = e;
 342             if (ef)
 343             {
 344                 wrbuf_putc(s->wrbuf, ' ');
 345                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 346             }
 347         }
 348     }
 349     if ((field = find_field(rec, "100")))
 350     {
 351         if ((subfield = find_subfield(field, 'a')))
 352         {
 353             ef = index(subfield, '\n');
 354             if ((e = index(subfield, '\t')) && e < ef)
 355                 ef = e;
 356             if (ef)
 357             {
 358                 wrbuf_puts(s->wrbuf, ", by ");
 359                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 360             }
 361         }
 362     }
 363     wrbuf_putc(s->wrbuf, '\0');
 364     obuf = nmem_strdup(s->nmem, wrbuf_buf(s->wrbuf));
 365     for (p = obuf; *p; p++)
 366         if (*p == '&' || *p == '<' || *p > 122 || *p < ' ')
 367             *p = ' ';
 368     return obuf;
 369 }
 370
 371 // Extract 245 $a $b 100 $a
 372 char *extract_mergekey(struct session *s, const char *rec)
 373 {
 374     const char *field, *subfield;
 375     char *e, *ef;
 376     char *out, *p, *pout;
 377
 378     wrbuf_rewind(s->wrbuf);
 379
 380     if (!(field = find_field(rec, "245")))
 381         return 0;
 382     if (!(subfield = find_subfield(field, 'a')))
 383         return 0;
 384     ef = index(subfield, '\n');
 385     if ((e = index(subfield, '\t')) && e < ef)
 386         ef = e;
 387     if (ef)
 388     {
 389         wrbuf_write(s->wrbuf, subfield, ef - subfield);
 390         if ((subfield = find_subfield(field, 'b')))
 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     if ((field = find_field(rec, "100")))
 403     {
 404         if ((subfield = find_subfield(field, 'a')))
 405         {
 406             ef = index(subfield, '\n');
 407             if ((e = index(subfield, '\t')) && e < ef)
 408                 ef = e;
 409             if (ef)
 410             {
 411                 wrbuf_puts(s->wrbuf, " field ");
 412                 wrbuf_write(s->wrbuf, subfield, ef - subfield);
 413             }
 414         }
 415     }
 416     wrbuf_putc(s->wrbuf, '\0');
 417     p = wrbuf_buf(s->wrbuf);
 418     out = pout = nmem_malloc(s->nmem, strlen(p) + 1);
 419
 420     while (*p)
 421     {
 422         while (isalnum(*p))
 423             *(pout++) = tolower(*(p++));
 424         while (*p && !isalnum(*p))
 425             p++;
 426         *(pout++) = ' ';
 427     }
 428     if (out != pout)
 429         *(--pout) = '\0';
 430
 431     return out;
 432 }
 433
 434 #ifdef RECHEAP
 435 static void push_record(struct session *s, struct record *r)
 436 {
 437     int p;
 438     assert(s->recheap_max + 1 < s->recheap_size);
 439
 440     s->recheap[p = ++s->recheap_max] = r;
 441     while (p > 0)
 442     {
 443         int parent = (p - 1) >> 1;
 444         if (strcmp(s->recheap[p]->merge_key, s->recheap[parent]->merge_key) < 0)
 445         {
 446             struct record *tmp;
 447             tmp = s->recheap[parent];
 448             s->recheap[parent] = s->recheap[p];
 449             s->recheap[p] = tmp;
 450             p = parent;
 451         }
 452         else
 453             break;
 454     }
 455 }
 456
 457 static struct record *top_record(struct session *s)
 458 {
 459     return s-> recheap_max >= 0 ?  s->recheap[0] : 0;
 460 }
 461
 462 static struct record *pop_record(struct session *s)
 463 {
 464     struct record *res;
 465     int p = 0;
 466     int lastnonleaf = (s->recheap_max - 1) >> 1;
 467
 468     if (s->recheap_max < 0)
 469         return 0;
 470
 471     res = s->recheap[0];
 472
 473     s->recheap[p] = s->recheap[s->recheap_max--];
 474
 475     while (p <= lastnonleaf)
 476     {
 477         int right = (p + 1) << 1;
 478         int left = right - 1;
 479         int min = left;
 480
 481         if (right < s->recheap_max &&
 482                 strcmp(s->recheap[right]->merge_key, s->recheap[left]->merge_key) < 0)
 483             min = right;
 484         if (strcmp(s->recheap[min]->merge_key, s->recheap[p]->merge_key) < 0)
 485         {
 486             struct record *tmp = s->recheap[min];
 487             s->recheap[min] = s->recheap[p];
 488             s->recheap[p] = tmp;
 489             p = min;
 490         }
 491         else
 492             break;
 493     }
 494     return res;
 495 }
 496
 497 // Like pop_record but collapses identical (merge_key) records
 498 // The heap will contain multiple independent matching records and possibly
 499 // one cluster, created the last time the list was scanned
 500 static struct record *pop_mrecord(struct session *s)
 501 {
 502     struct record *this;
 503     struct record *next;
 504
 505     if (!(this = pop_record(s)))
 506         return 0;
 507
 508     // Collapse identical records
 509     while ((next = top_record(s)))
 510     {
 511         struct record *p, *tmpnext;
 512         if (strcmp(this->merge_key, next->merge_key))
 513             break;
 514         // Absorb record (and clustersiblings) into a supercluster
 515         for (p = next; p; p = tmpnext) {
 516             tmpnext = p->next_cluster;
 517             p->next_cluster = this->next_cluster;
 518             this->next_cluster = p;
 519         }
 520
 521         pop_record(s);
 522     }
 523     return this;
 524 }
 525
 526 // Reads records in sort order. Store records in top of heapspace until rewind is called.
 527 static struct record *read_recheap(struct session *s)
 528 {
 529     struct record *r = pop_mrecord(s);
 530
 531     if (r)
 532     {
 533         if (s->recheap_scratch < 0)
 534             s->recheap_scratch = s->recheap_size;
 535         s->recheap[--s->recheap_scratch] = r;
 536     }
 537
 538     return r;
 539 }
 540
 541 // Return records to heap after read
 542 static void rewind_recheap(struct session *s)
 543 {
 544     while (s->recheap_scratch >= 0) {
 545         push_record(s, s->recheap[s->recheap_scratch++]);
 546         if (s->recheap_scratch >= s->recheap_size)
 547             s->recheap_scratch = -1;
 548     }
 549 }
 550
 551 #endif
 552
 553 // FIXME needs to be generalized. Should flexibly generate X lists per search
 554 static void extract_subject(struct session *s, const char *rec)
 555 {
 556     const char *field, *subfield;
 557
 558     while ((field = find_field(rec, "650")))
 559     {
 560         rec = field + 1; // Crude way to cause a loop through repeating fields
 561         if ((subfield = find_subfield(field, 'a')))
 562         {
 563             char *e, *ef;
 564             char buf[1024];
 565             int len;
 566
 567             ef = index(subfield, '\n');
 568             if (!ef)
 569                 return;
 570             if ((e = index(subfield, '\t')) && e < ef)
 571                 ef = e;
 572             while (ef > subfield && !isalpha(*(ef - 1)) && *(ef - 1) != ')')
 573                 ef--;
 574             len = ef - subfield;
 575             assert(len < 1023);
 576             memcpy(buf, subfield, len);
 577             buf[len] = '\0';
 578             if (*buf)
 579                 termlist_insert(s->termlist, buf);
 580         }
 581     }
 582 }
 583
 584 static void pull_relevance_field(struct session *s, struct record *head, const char *rec,
 585         char *field, int mult)
 586 {
 587     const char *fb;
 588     while ((fb = find_field(rec, field)))
 589     {
 590         char *ffield = strchr(fb, '\t');
 591         if (!ffield)
 592             return;
 593         char *eol = strchr(ffield, '\n');
 594         if (!eol)
 595             return;
 596         relevance_countwords(s->relevance, head, ffield, eol - ffield, mult);
 597         rec = field + 1; // Crude way to cause a loop through repeating fields
 598     }
 599 }
 600
 601 static void pull_relevance_keys(struct session *s, struct record *head,  struct record *rec)
 602 {
 603     relevance_newrec(s->relevance, head);
 604     pull_relevance_field(s, head, rec->buf, "100", 2);
 605     pull_relevance_field(s, head, rec->buf, "245", 4);
 606     //pull_relevance_field(s, head, rec->buf, "530", 1);
 607     pull_relevance_field(s, head, rec->buf, "630", 1);
 608     pull_relevance_field(s, head, rec->buf, "650", 1);
 609     pull_relevance_field(s, head, rec->buf, "700", 1);
 610     relevance_donerecord(s->relevance, head);
 611 }
 612
 613 struct record *ingest_record(struct target *t, char *buf, int len)
 614 {
 615     struct session *s = t->session;
 616     struct record *res;
 617     struct record *head;
 618     const char *recbuf;
 619
 620     wrbuf_rewind(s->wrbuf);
 621     yaz_marc_xml(s->yaz_marc, YAZ_MARC_LINE);
 622     if (yaz_marc_decode_wrbuf(s->yaz_marc, buf, len, s->wrbuf) < 0)
 623     {
 624         yaz_log(YLOG_WARN, "Failed to decode MARC record");
 625         return 0;
 626     }
 627     wrbuf_putc(s->wrbuf, '\0');
 628     recbuf = wrbuf_buf(s->wrbuf);
 629
 630     res = nmem_malloc(s->nmem, sizeof(struct record));
 631     res->buf = nmem_strdup(s->nmem, recbuf);
 632
 633     extract_subject(s, res->buf);
 634
 635     res->title = extract_title(s, res->buf);
 636     res->merge_key = extract_mergekey(s, res->buf);
 637     if (!res->merge_key)
 638         return 0;
 639     res->target = t;
 640     res->next_cluster = 0;
 641     res->target_offset = -1;
 642     res->term_frequency_vec = 0;
 643
 644     head = reclist_insert(s->reclist, res);
 645
 646     pull_relevance_keys(s, head, res);
 647
 648     s->total_records++;
 649
 650     return res;
 651 }
 652
 653 void ingest_records(struct target *t, Z_Records *r)
 654 {
 655     //struct session *s = t->session;
 656     struct record *rec;
 657     Z_NamePlusRecordList *rlist;
 658     int i;
 659
 660     if (r->which != Z_Records_DBOSD)
 661         return;
 662     rlist = r->u.databaseOrSurDiagnostics;
 663     for (i = 0; i < rlist->num_records; i++)
 664     {
 665         Z_NamePlusRecord *npr = rlist->records[i];
 666         Z_External *e;
 667         char *buf;
 668         int len;
 669
 670         if (npr->which != Z_NamePlusRecord_databaseRecord)
 671         {
 672             yaz_log(YLOG_WARN, "Unexpected record type, probably diagnostic");
 673             continue;
 674         }
 675         e = npr->u.databaseRecord;
 676         if (e->which != Z_External_octet)
 677         {
 678             yaz_log(YLOG_WARN, "Unexpected external branch, probably BER");
 679             continue;
 680         }
 681         buf = (char*) e->u.octet_aligned->buf;
 682         len = e->u.octet_aligned->len;
 683
 684         rec = ingest_record(t, buf, len);
 685         if (!rec)
 686             continue;
 687     }
 688 }
 689
 690 static void do_presentResponse(IOCHAN i, Z_APDU *a)
 691 {
 692     struct target *t = iochan_getdata(i);
 693     Z_PresentResponse *r = a->u.presentResponse;
 694
 695     if (r->records) {
 696         Z_Records *recs = r->records;
 697         if (recs->which == Z_Records_NSD)
 698         {
 699             yaz_log(YLOG_WARN, "Non-surrogate diagnostic");
 700             t->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 701             t->state = Error;
 702         }
 703     }
 704
 705     if (!*r->presentStatus && t->state != Error)
 706     {
 707         yaz_log(YLOG_DEBUG, "Good Present response");
 708         t->records += *r->numberOfRecordsReturned;
 709         ingest_records(t, r->records);
 710         t->state = Idle;
 711     }
 712     else if (*r->presentStatus)
 713     {
 714         yaz_log(YLOG_WARN, "Bad Present response");
 715         t->state = Error;
 716     }
 717 }
 718
 719 static void handler(IOCHAN i, int event)
 720 {
 721     struct target *t = iochan_getdata(i);
 722     struct session *s = t->session;
 723     //static int waiting = 0;
 724
 725     if (t->state == No_connection) /* Start connection */
 726     {
 727         int res = cs_connect(t->link, t->addr);
 728
 729         t->state = Connecting;
 730         if (!res) /* we are go */
 731             iochan_setevent(i, EVENT_OUTPUT);
 732         else if (res == 1)
 733             iochan_setflags(i, EVENT_OUTPUT);
 734         else
 735         {
 736             yaz_log(YLOG_WARN|YLOG_ERRNO, "ERROR %s connect\n", t->hostport);
 737             target_destroy(i);
 738             return;
 739         }
 740     }
 741
 742     else if (t->state == Connecting && event & EVENT_OUTPUT)
 743     {
 744         int errcode;
 745         socklen_t errlen = sizeof(errcode);
 746
 747         if (getsockopt(cs_fileno(t->link), SOL_SOCKET, SO_ERROR, &errcode,
 748             &errlen) < 0 || errcode != 0)
 749         {
 750             target_destroy(i);
 751             return;
 752         }
 753         else
 754         {
 755             yaz_log(YLOG_DEBUG, "Connect OK");
 756             t->state = Connected;
 757         }
 758     }
 759
 760     else if (event & EVENT_INPUT)
 761     {
 762         int len = cs_get(t->link, &t->ibuf, &t->ibufsize);
 763
 764         if (len < 0)
 765         {
 766             target_destroy(i);
 767             return;
 768         }
 769         if (len == 0)
 770         {
 771             target_destroy(i);
 772             return;
 773         }
 774         else if (len > 1)
 775         {
 776             if (t->requestid == s->requestid || t->state == Initializing)
 777             {
 778                 Z_APDU *a;
 779
 780                 odr_reset(t->odr_in);
 781                 odr_setbuf(t->odr_in, t->ibuf, len, 0);
 782                 if (!z_APDU(t->odr_in, &a, 0, 0))
 783                 {
 784                     target_destroy(i);
 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                         target_destroy(i);
 801                         return;
 802                 }
 803                 // if (cs_more(t->link))
 804                 //    iochan_setevent(i, EVENT_INPUT);
 805             }
 806             else  // we throw away response and go to idle mode
 807             {
 808                 yaz_log(YLOG_DEBUG, "Ignoring result to previous operation");
 809                 t->state = Idle;
 810             }
 811         }
 812         /* if len==1 we do nothing but wait for more input */
 813     }
 814
 815     else if (t->state == Connected) {
 816         send_init(i);
 817     }
 818
 819     if (t->state == Idle)
 820     {
 821         if (t->requestid != s->requestid && *s->query) {
 822             send_search(i);
 823         }
 824         else if (t->hits > 0 && t->records < global_parameters.toget &&
 825             t->records < t->hits) {
 826             send_present(i);
 827         }
 828     }
 829 }
 830
 831 static void target_destroy(IOCHAN i)
 832 {
 833     struct target *t = iochan_getdata(i);
 834     struct session *s = t->session;
 835     struct target **p;
 836     assert(iochan_getfun(i) == handler);
 837
 838     yaz_log(YLOG_DEBUG, "Destroying target");
 839
 840     if (t->ibuf)
 841         xfree(t->ibuf);
 842     cs_close(t->link);
 843     if (t->odr_in)
 844         odr_destroy(t->odr_in);
 845     if (t->odr_out)
 846         odr_destroy(t->odr_out);
 847     for (p = &s->targets; *p; p = &(*p)->next)
 848         if (*p == t)
 849         {
 850             *p = (*p)->next;
 851             break;
 852         }
 853     xfree(t);
 854     iochan_destroy(i);
 855 }
 856
 857 int load_targets(struct session *s, const char *fn)
 858 {
 859     FILE *f = fopen(fn, "r");
 860     char line[256];
 861     struct target **target_p;
 862
 863     if (!f)
 864     {
 865         yaz_log(YLOG_WARN|YLOG_ERRNO, "open %s", fn);
 866         return -1;
 867     }
 868
 869     while (s->targets)
 870         target_destroy(s->targets->iochan);
 871
 872     s->query[0] = '\0';
 873     target_p = &s->targets;
 874     while (fgets(line, 255, f))
 875     {
 876         char *url, *p;
 877         struct target *target;
 878         IOCHAN new;
 879
 880         if (strncmp(line, "target ", 7))
 881             continue;
 882         url = line + 7;
 883         url[strlen(url) - 1] = '\0';
 884         yaz_log(LOG_DEBUG, "Target: %s", url);
 885
 886         *target_p = target = xmalloc(sizeof(**target_p));
 887         target->next = 0;
 888         target_p = &target->next;
 889         target->state = No_connection;
 890         target->ibuf = 0;
 891         target->ibufsize = 0;
 892         target->odr_in = odr_createmem(ODR_DECODE);
 893         target->odr_out = odr_createmem(ODR_ENCODE);
 894         target->hits = -1;
 895         target->setno = 0;
 896         target->session = s;
 897         target->requestid = -1;
 898         target->records = 0;
 899         target->diagnostic = 0;
 900         strcpy(target->fullname, url);
 901         if ((p = strchr(url, '/')))
 902         {
 903             *p = '\0';
 904             strcpy(target->hostport, url);
 905             *p = '/';
 906             p++;
 907             strcpy(target->databases[0], p);
 908             target->databases[1][0] = '\0';
 909         }
 910         else
 911         {
 912             strcpy(target->hostport, url);
 913             strcpy(target->databases[0], "Default");
 914             target->databases[1][0] = '\0';
 915         }
 916
 917         if (!(target->link = cs_create(tcpip_type, 0, PROTO_Z3950)))
 918         {
 919             yaz_log(YLOG_FATAL|YLOG_ERRNO, "Failed to create comstack");
 920             exit(1);
 921         }
 922         if (!(target->addr = cs_straddr(target->link, target->hostport)))
 923         {
 924             printf("ERROR %s bad-address", target->hostport);
 925             target->state = Failed;
 926             continue;
 927         }
 928         target->iochan = new = iochan_create(cs_fileno(target->link), handler, 0);
 929         assert(new);
 930         iochan_setdata(new, target);
 931         iochan_setevent(new, EVENT_EXCEPT);
 932         new->next = channel_list;
 933         channel_list = new;
 934     }
 935     fclose(f);
 936
 937     return 0;
 938 }
 939
 940 static void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 941 {
 942     switch (n->kind)
 943     {
 944         case CCL_RPN_AND:
 945         case CCL_RPN_OR:
 946         case CCL_RPN_NOT:
 947         case CCL_RPN_PROX:
 948             pull_terms(nmem, n->u.p[0], termlist, num);
 949             pull_terms(nmem, n->u.p[1], termlist, num);
 950             break;
 951         case CCL_RPN_TERM:
 952             termlist[(*num)++] = nmem_strdup(nmem, n->u.t.term);
 953             break;
 954         default: // NOOP
 955             break;
 956     }
 957 }
 958
 959 // Extract terms from query into null-terminated termlist
 960 static int extract_terms(NMEM nmem, char *query, char **termlist)
 961 {
 962     int error, pos;
 963     struct ccl_rpn_node *n;
 964     int num = 0;
 965
 966     n = ccl_find_str(global_parameters.ccl_filter, query, &error, &pos);
 967     if (!n)
 968         return -1;
 969     pull_terms(nmem, n, termlist, &num);
 970     termlist[num] = 0;
 971     ccl_rpn_delete(n);
 972     return 0;
 973 }
 974
 975 char *search(struct session *s, char *query)
 976 {
 977     IOCHAN c;
 978     int live_channels = 0;
 979
 980     yaz_log(YLOG_DEBUG, "Search");
 981
 982     // Determine what iochans belong to this session
 983     // It might have been better to have a list of them
 984
 985     strcpy(s->query, query);
 986     s->requestid++;
 987     nmem_reset(s->nmem);
 988     for (c = channel_list; c; c = c->next)
 989     {
 990         struct target *t;
 991
 992         if (iochan_getfun(c) != handler) // Not a Z target
 993             continue;
 994         t = iochan_getdata(c);
 995         if (t->session == s)
 996         {
 997             t->hits = -1;
 998             t->records = 0;
 999             t->diagnostic = 0;
1000
1001             if (t->state == Error)
1002                 t->state = Idle;
1003
1004             if (t->state == Idle)
1005                 iochan_setflag(c, EVENT_OUTPUT);
1006
1007             live_channels++;
1008         }
1009     }
1010     if (live_channels)
1011     {
1012         char *p[512];
1013         int maxrecs = live_channels * global_parameters.toget;
1014         s->termlist = termlist_create(s->nmem, maxrecs, 15);
1015         s->reclist = reclist_create(s->nmem, maxrecs);
1016         extract_terms(s->nmem, query, p);
1017         s->relevance = relevance_create(s->nmem, (const char **) p, maxrecs);
1018         s->total_records = s->total_hits = 0;
1019     }
1020     else
1021         return "NOTARGETS";
1022
1023     return 0;
1024 }
1025
1026 struct session *new_session()
1027 {
1028     struct session *session = xmalloc(sizeof(*session));
1029
1030     yaz_log(YLOG_DEBUG, "New pazpar2 session");
1031
1032     session->total_hits = 0;
1033     session->total_records = 0;
1034     session->termlist = 0;
1035     session->reclist = 0;
1036     session->requestid = -1;
1037     session->targets = 0;
1038     session->pqf_parser = yaz_pqf_create();
1039     session->query[0] = '\0';
1040     session->nmem = nmem_create();
1041     session->yaz_marc = yaz_marc_create();
1042     yaz_marc_subfield_str(session->yaz_marc, "\t");
1043     session->wrbuf = wrbuf_alloc();
1044
1045     return session;
1046 }
1047
1048 void session_destroy(struct session *s)
1049 {
1050     // FIXME do some shit here!!!!
1051 }
1052
1053 struct hitsbytarget *hitsbytarget(struct session *s, int *count)
1054 {
1055     static struct hitsbytarget res[1000]; // FIXME MM
1056     IOCHAN c;
1057
1058     *count = 0;
1059     for (c = channel_list; c; c = c->next)
1060         if (iochan_getfun(c) == handler)
1061         {
1062             struct target *t = iochan_getdata(c);
1063             if (t->session == s)
1064             {
1065                 strcpy(res[*count].id, t->hostport);
1066                 res[*count].hits = t->hits;
1067                 res[*count].records = t->records;
1068                 res[*count].diagnostic = t->diagnostic;
1069                 res[*count].state = state_strings[(int) t->state];
1070                 (*count)++;
1071             }
1072         }
1073
1074     return res;
1075 }
1076
1077 struct termlist_score **termlist(struct session *s, int *num)
1078 {
1079     return termlist_highscore(s->termlist, num);
1080 }
1081
1082 struct record **show(struct session *s, int start, int *num, int *total, int *sumhits)
1083 {
1084     struct record **recs = nmem_malloc(s->nmem, *num * sizeof(struct record *));
1085     int i;
1086
1087     relevance_prepare_read(s->relevance, s->reclist);
1088
1089     *total = s->reclist->num_records;
1090     *sumhits = s->total_hits;
1091
1092     for (i = 0; i < start; i++)
1093         if (!reclist_read_record(s->reclist))
1094         {
1095             *num = 0;
1096             return 0;
1097         }
1098
1099     for (i = 0; i < *num; i++)
1100     {
1101         struct record *r = reclist_read_record(s->reclist);
1102         if (!r)
1103         {
1104             *num = i;
1105             break;
1106         }
1107         recs[i] = r;
1108     }
1109     return recs;
1110 }
1111
1112 void statistics(struct session *s, struct statistics *stat)
1113 {
1114     IOCHAN c;
1115     int i;
1116
1117     bzero(stat, sizeof(*stat));
1118     for (i = 0, c = channel_list; c; i++, c = c->next)
1119     {
1120         struct target *t;
1121         if (iochan_getfun(c) != handler)
1122             continue;
1123         t = iochan_getdata(c);
1124         switch (t->state)
1125         {
1126             case No_connection: stat->num_no_connection++; break;
1127             case Connecting: stat->num_connecting++; break;
1128             case Initializing: stat->num_initializing++; break;
1129             case Searching: stat->num_searching++; break;
1130             case Presenting: stat->num_presenting++; break;
1131             case Idle: stat->num_idle++; break;
1132             case Failed: stat->num_failed++; break;
1133             case Error: stat->num_error++; break;
1134             default: break;
1135         }
1136     }
1137     stat->num_hits = s->total_hits;
1138     stat->num_records = s->total_records;
1139
1140     stat->num_connections = i;
1141 }
1142
1143 static CCL_bibset load_cclfile(const char *fn)
1144 {
1145     CCL_bibset res = ccl_qual_mk();
1146     if (ccl_qual_fname(res, fn) < 0)
1147     {
1148         yaz_log(YLOG_FATAL|YLOG_ERRNO, "%s", fn);
1149         exit(1);
1150     }
1151     return res;
1152 }
1153
1154 int main(int argc, char **argv)
1155 {
1156     int ret;
1157     char *arg;
1158
1159     if (signal(SIGPIPE, SIG_IGN) < 0)
1160         yaz_log(YLOG_WARN|YLOG_ERRNO, "signal");
1161
1162     yaz_log_init(YLOG_DEFAULT_LEVEL|YLOG_DEBUG, "pazpar2", 0);
1163
1164     while ((ret = options("c:h:p:C:", argv, argc, &arg)) != -2)
1165     {
1166         switch (ret) {
1167             case 0:
1168                 break;
1169             case 'c':
1170                 command_init(atoi(arg));
1171                 break;
1172             case 'C':
1173                 global_parameters.ccl_filter = load_cclfile(arg);
1174                 break;
1175             case 'h':
1176                 http_init(atoi(arg));
1177                 break;
1178             case 'p':
1179                 http_set_proxyaddr(arg);
1180                 break;
1181             default:
1182                 fprintf(stderr, "Usage: pazpar2 -d comport");
1183                 exit(1);
1184         }
1185
1186     }
1187
1188     if (!global_parameters.ccl_filter)
1189         global_parameters.ccl_filter = load_cclfile("default.bib");
1190
1191     event_loop(&channel_list);
1192
1193     return 0;
1194 }
1195
1196 /*
1197  * Local variables:
1198  * c-basic-offset: 4
1199  * indent-tabs-mode: nil
1200  * End:
1201  * vim: shiftwidth=4 tabstop=8 expandtab
1202  */