pazpar2.c

   1 /* $Id: pazpar2.c,v 1.6 2006-11-27 14:35:15 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             termlist_insert(s->termlist, buf);
 579         }
 580     }
 581 }
 582
 583 static void pull_relevance_field(struct session *s, struct record *head, const char *rec,
 584         char *field, int mult)
 585 {
 586     const char *fb;
 587     while ((fb = find_field(rec, field)))
 588     {
 589         char *ffield = strchr(fb, '\t');
 590         if (!ffield)
 591             return;
 592         char *eol = strchr(ffield, '\n');
 593         if (!eol)
 594             return;
 595         relevance_countwords(s->relevance, head, ffield, eol - ffield, mult);
 596         rec = field + 1; // Crude way to cause a loop through repeating fields
 597     }
 598 }
 599
 600 static void pull_relevance_keys(struct session *s, struct record *head,  struct record *rec)
 601 {
 602     relevance_newrec(s->relevance, head);
 603     pull_relevance_field(s, head, rec->buf, "100", 2);
 604     pull_relevance_field(s, head, rec->buf, "245", 4);
 605     //pull_relevance_field(s, head, rec->buf, "530", 1);
 606     pull_relevance_field(s, head, rec->buf, "630", 1);
 607     pull_relevance_field(s, head, rec->buf, "650", 1);
 608     pull_relevance_field(s, head, rec->buf, "700", 1);
 609     relevance_donerecord(s->relevance, head);
 610 }
 611
 612 struct record *ingest_record(struct target *t, char *buf, int len)
 613 {
 614     struct session *s = t->session;
 615     struct record *res;
 616     struct record *head;
 617     const char *recbuf;
 618
 619     wrbuf_rewind(s->wrbuf);
 620     yaz_marc_xml(s->yaz_marc, YAZ_MARC_LINE);
 621     if (yaz_marc_decode_wrbuf(s->yaz_marc, buf, len, s->wrbuf) < 0)
 622     {
 623         yaz_log(YLOG_WARN, "Failed to decode MARC record");
 624         return 0;
 625     }
 626     wrbuf_putc(s->wrbuf, '\0');
 627     recbuf = wrbuf_buf(s->wrbuf);
 628
 629     res = nmem_malloc(s->nmem, sizeof(struct record));
 630     res->buf = nmem_strdup(s->nmem, recbuf);
 631
 632     extract_subject(s, res->buf);
 633
 634     res->title = extract_title(s, res->buf);
 635     res->merge_key = extract_mergekey(s, res->buf);
 636     if (!res->merge_key)
 637         return 0;
 638     res->target = t;
 639     res->next_cluster = 0;
 640     res->target_offset = -1;
 641     res->term_frequency_vec = 0;
 642
 643     head = reclist_insert(s->reclist, res);
 644
 645     pull_relevance_keys(s, head, res);
 646
 647     s->total_records++;
 648
 649     return res;
 650 }
 651
 652 void ingest_records(struct target *t, Z_Records *r)
 653 {
 654     //struct session *s = t->session;
 655     struct record *rec;
 656     Z_NamePlusRecordList *rlist;
 657     int i;
 658
 659     if (r->which != Z_Records_DBOSD)
 660         return;
 661     rlist = r->u.databaseOrSurDiagnostics;
 662     for (i = 0; i < rlist->num_records; i++)
 663     {
 664         Z_NamePlusRecord *npr = rlist->records[i];
 665         Z_External *e;
 666         char *buf;
 667         int len;
 668
 669         if (npr->which != Z_NamePlusRecord_databaseRecord)
 670         {
 671             yaz_log(YLOG_WARN, "Unexpected record type, probably diagnostic");
 672             continue;
 673         }
 674         e = npr->u.databaseRecord;
 675         if (e->which != Z_External_octet)
 676         {
 677             yaz_log(YLOG_WARN, "Unexpected external branch, probably BER");
 678             continue;
 679         }
 680         buf = (char*) e->u.octet_aligned->buf;
 681         len = e->u.octet_aligned->len;
 682
 683         rec = ingest_record(t, buf, len);
 684         if (!rec)
 685             continue;
 686     }
 687 }
 688
 689 static void do_presentResponse(IOCHAN i, Z_APDU *a)
 690 {
 691     struct target *t = iochan_getdata(i);
 692     Z_PresentResponse *r = a->u.presentResponse;
 693
 694     if (r->records) {
 695         Z_Records *recs = r->records;
 696         if (recs->which == Z_Records_NSD)
 697         {
 698             yaz_log(YLOG_WARN, "Non-surrogate diagnostic");
 699             t->diagnostic = *recs->u.nonSurrogateDiagnostic->condition;
 700             t->state = Error;
 701         }
 702     }
 703
 704     if (!*r->presentStatus && t->state != Error)
 705     {
 706         yaz_log(YLOG_DEBUG, "Good Present response");
 707         t->records += *r->numberOfRecordsReturned;
 708         ingest_records(t, r->records);
 709         t->state = Idle;
 710     }
 711     else if (*r->presentStatus)
 712     {
 713         yaz_log(YLOG_WARN, "Bad Present response");
 714         t->state = Error;
 715     }
 716 }
 717
 718 static void handler(IOCHAN i, int event)
 719 {
 720     struct target *t = iochan_getdata(i);
 721     struct session *s = t->session;
 722     //static int waiting = 0;
 723
 724     if (t->state == No_connection) /* Start connection */
 725     {
 726         int res = cs_connect(t->link, t->addr);
 727
 728         t->state = Connecting;
 729         if (!res) /* we are go */
 730             iochan_setevent(i, EVENT_OUTPUT);
 731         else if (res == 1)
 732             iochan_setflags(i, EVENT_OUTPUT);
 733         else
 734         {
 735             yaz_log(YLOG_WARN|YLOG_ERRNO, "ERROR %s connect\n", t->hostport);
 736             target_destroy(i);
 737             return;
 738         }
 739     }
 740
 741     else if (t->state == Connecting && event & EVENT_OUTPUT)
 742     {
 743         int errcode;
 744         socklen_t errlen = sizeof(errcode);
 745
 746         if (getsockopt(cs_fileno(t->link), SOL_SOCKET, SO_ERROR, &errcode,
 747             &errlen) < 0 || errcode != 0)
 748         {
 749             target_destroy(i);
 750             return;
 751         }
 752         else
 753         {
 754             yaz_log(YLOG_DEBUG, "Connect OK");
 755             t->state = Connected;
 756         }
 757     }
 758
 759     else if (event & EVENT_INPUT)
 760     {
 761         int len = cs_get(t->link, &t->ibuf, &t->ibufsize);
 762
 763         if (len < 0)
 764         {
 765             target_destroy(i);
 766             return;
 767         }
 768         if (len == 0)
 769         {
 770             target_destroy(i);
 771             return;
 772         }
 773         else if (len > 1)
 774         {
 775             if (t->requestid == s->requestid || t->state == Initializing)
 776             {
 777                 Z_APDU *a;
 778
 779                 odr_reset(t->odr_in);
 780                 odr_setbuf(t->odr_in, t->ibuf, len, 0);
 781                 if (!z_APDU(t->odr_in, &a, 0, 0))
 782                 {
 783                     target_destroy(i);
 784                     return;
 785                 }
 786                 switch (a->which)
 787                 {
 788                     case Z_APDU_initResponse:
 789                         do_initResponse(i, a);
 790                         break;
 791                     case Z_APDU_searchResponse:
 792                         do_searchResponse(i, a);
 793                         break;
 794                     case Z_APDU_presentResponse:
 795                         do_presentResponse(i, a);
 796                         break;
 797                     default:
 798                         yaz_log(YLOG_WARN, "Unexpected result from server");
 799                         target_destroy(i);
 800                         return;
 801                 }
 802                 // if (cs_more(t->link))
 803                 //    iochan_setevent(i, EVENT_INPUT);
 804             }
 805             else  // we throw away response and go to idle mode
 806             {
 807                 yaz_log(YLOG_DEBUG, "Ignoring result to previous operation");
 808                 t->state = Idle;
 809             }
 810         }
 811         /* if len==1 we do nothing but wait for more input */
 812     }
 813
 814     else if (t->state == Connected) {
 815         send_init(i);
 816     }
 817
 818     if (t->state == Idle)
 819     {
 820         if (t->requestid != s->requestid && *s->query) {
 821             send_search(i);
 822         }
 823         else if (t->hits > 0 && t->records < global_parameters.toget &&
 824             t->records < t->hits) {
 825             send_present(i);
 826         }
 827     }
 828 }
 829
 830 static void target_destroy(IOCHAN i)
 831 {
 832     struct target *t = iochan_getdata(i);
 833     struct session *s = t->session;
 834     struct target **p;
 835     assert(iochan_getfun(i) == handler);
 836
 837     yaz_log(YLOG_DEBUG, "Destroying target");
 838
 839     if (t->ibuf)
 840         xfree(t->ibuf);
 841     cs_close(t->link);
 842     if (t->odr_in)
 843         odr_destroy(t->odr_in);
 844     if (t->odr_out)
 845         odr_destroy(t->odr_out);
 846     for (p = &s->targets; *p; p = &(*p)->next)
 847         if (*p == t)
 848         {
 849             *p = (*p)->next;
 850             break;
 851         }
 852     xfree(t);
 853     iochan_destroy(i);
 854 }
 855
 856 int load_targets(struct session *s, const char *fn)
 857 {
 858     FILE *f = fopen(fn, "r");
 859     char line[256];
 860     struct target **target_p;
 861
 862     if (!f)
 863     {
 864         yaz_log(YLOG_WARN|YLOG_ERRNO, "open %s", fn);
 865         return -1;
 866     }
 867
 868     while (s->targets)
 869         target_destroy(s->targets->iochan);
 870
 871     s->query[0] = '\0';
 872     target_p = &s->targets;
 873     while (fgets(line, 255, f))
 874     {
 875         char *url, *p;
 876         struct target *target;
 877         IOCHAN new;
 878
 879         if (strncmp(line, "target ", 7))
 880             continue;
 881         url = line + 7;
 882         url[strlen(url) - 1] = '\0';
 883         yaz_log(LOG_DEBUG, "Target: %s", url);
 884
 885         *target_p = target = xmalloc(sizeof(**target_p));
 886         target->next = 0;
 887         target_p = &target->next;
 888         target->state = No_connection;
 889         target->ibuf = 0;
 890         target->ibufsize = 0;
 891         target->odr_in = odr_createmem(ODR_DECODE);
 892         target->odr_out = odr_createmem(ODR_ENCODE);
 893         target->hits = -1;
 894         target->setno = 0;
 895         target->session = s;
 896         target->requestid = -1;
 897         target->records = 0;
 898         target->diagnostic = 0;
 899         strcpy(target->fullname, url);
 900         if ((p = strchr(url, '/')))
 901         {
 902             *p = '\0';
 903             strcpy(target->hostport, url);
 904             *p = '/';
 905             p++;
 906             strcpy(target->databases[0], p);
 907             target->databases[1][0] = '\0';
 908         }
 909         else
 910         {
 911             strcpy(target->hostport, url);
 912             strcpy(target->databases[0], "Default");
 913             target->databases[1][0] = '\0';
 914         }
 915
 916         if (!(target->link = cs_create(tcpip_type, 0, PROTO_Z3950)))
 917         {
 918             yaz_log(YLOG_FATAL|YLOG_ERRNO, "Failed to create comstack");
 919             exit(1);
 920         }
 921         if (!(target->addr = cs_straddr(target->link, target->hostport)))
 922         {
 923             printf("ERROR %s bad-address", target->hostport);
 924             target->state = Failed;
 925             continue;
 926         }
 927         target->iochan = new = iochan_create(cs_fileno(target->link), handler, 0);
 928         assert(new);
 929         iochan_setdata(new, target);
 930         iochan_setevent(new, EVENT_EXCEPT);
 931         new->next = channel_list;
 932         channel_list = new;
 933     }
 934     fclose(f);
 935
 936     return 0;
 937 }
 938
 939 static void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 940 {
 941     switch (n->kind)
 942     {
 943         case CCL_RPN_AND:
 944         case CCL_RPN_OR:
 945         case CCL_RPN_NOT:
 946         case CCL_RPN_PROX:
 947             pull_terms(nmem, n->u.p[0], termlist, num);
 948             pull_terms(nmem, n->u.p[1], termlist, num);
 949             break;
 950         case CCL_RPN_TERM:
 951             termlist[(*num)++] = nmem_strdup(nmem, n->u.t.term);
 952             break;
 953         default: // NOOP
 954             break;
 955     }
 956 }
 957
 958 // Extract terms from query into null-terminated termlist
 959 static int extract_terms(NMEM nmem, char *query, char **termlist)
 960 {
 961     int error, pos;
 962     struct ccl_rpn_node *n;
 963     int num = 0;
 964
 965     n = ccl_find_str(global_parameters.ccl_filter, query, &error, &pos);
 966     if (!n)
 967         return -1;
 968     pull_terms(nmem, n, termlist, &num);
 969     termlist[num] = 0;
 970     ccl_rpn_delete(n);
 971     return 0;
 972 }
 973
 974 char *search(struct session *s, char *query)
 975 {
 976     IOCHAN c;
 977     int live_channels = 0;
 978
 979     yaz_log(YLOG_DEBUG, "Search");
 980
 981     // Determine what iochans belong to this session
 982     // It might have been better to have a list of them
 983
 984     strcpy(s->query, query);
 985     s->requestid++;
 986     nmem_reset(s->nmem);
 987     for (c = channel_list; c; c = c->next)
 988     {
 989         struct target *t;
 990
 991         if (iochan_getfun(c) != handler) // Not a Z target
 992             continue;
 993         t = iochan_getdata(c);
 994         if (t->session == s)
 995         {
 996             t->hits = -1;
 997             t->records = 0;
 998             t->diagnostic = 0;
 999
1000             if (t->state == Error)
1001                 t->state = Idle;
1002
1003             if (t->state == Idle)
1004                 iochan_setflag(c, EVENT_OUTPUT);
1005
1006             live_channels++;
1007         }
1008     }
1009     if (live_channels)
1010     {
1011         char *p[512];
1012         int maxrecs = live_channels * global_parameters.toget;
1013         s->termlist = termlist_create(s->nmem, maxrecs, 15);
1014         s->reclist = reclist_create(s->nmem, maxrecs);
1015         extract_terms(s->nmem, query, p);
1016         s->relevance = relevance_create(s->nmem, (const char **) p, maxrecs);
1017         s->total_records = s->total_hits = 0;
1018     }
1019     else
1020         return "NOTARGETS";
1021
1022     return 0;
1023 }
1024
1025 struct session *new_session()
1026 {
1027     struct session *session = xmalloc(sizeof(*session));
1028
1029     yaz_log(YLOG_DEBUG, "New pazpar2 session");
1030
1031     session->total_hits = 0;
1032     session->total_records = 0;
1033     session->termlist = 0;
1034     session->reclist = 0;
1035     session->requestid = -1;
1036     session->targets = 0;
1037     session->pqf_parser = yaz_pqf_create();
1038     session->query[0] = '\0';
1039     session->nmem = nmem_create();
1040     session->yaz_marc = yaz_marc_create();
1041     yaz_marc_subfield_str(session->yaz_marc, "\t");
1042     session->wrbuf = wrbuf_alloc();
1043
1044     return session;
1045 }
1046
1047 void session_destroy(struct session *s)
1048 {
1049     // FIXME do some shit here!!!!
1050 }
1051
1052 struct hitsbytarget *hitsbytarget(struct session *s, int *count)
1053 {
1054     static struct hitsbytarget res[1000]; // FIXME MM
1055     IOCHAN c;
1056
1057     *count = 0;
1058     for (c = channel_list; c; c = c->next)
1059         if (iochan_getfun(c) == handler)
1060         {
1061             struct target *t = iochan_getdata(c);
1062             if (t->session == s)
1063             {
1064                 strcpy(res[*count].id, t->hostport);
1065                 res[*count].hits = t->hits;
1066                 res[*count].records = t->records;
1067                 res[*count].diagnostic = t->diagnostic;
1068                 res[*count].state = state_strings[(int) t->state];
1069                 (*count)++;
1070             }
1071         }
1072
1073     return res;
1074 }
1075
1076 struct termlist_score **termlist(struct session *s, int *num)
1077 {
1078     return termlist_highscore(s->termlist, num);
1079 }
1080
1081 struct record **show(struct session *s, int start, int *num)
1082 {
1083     struct record **recs = nmem_malloc(s->nmem, *num * sizeof(struct record *));
1084     int i;
1085
1086     // FIXME -- skip initial records
1087
1088     relevance_prepare_read(s->relevance, s->reclist);
1089     for (i = 0; i < *num; i++)
1090     {
1091         struct record *r = reclist_read_record(s->reclist);
1092         if (!r)
1093         {
1094             *num = i;
1095             break;
1096         }
1097         recs[i] = r;
1098     }
1099     return recs;
1100 }
1101
1102 void statistics(struct session *s, struct statistics *stat)
1103 {
1104     IOCHAN c;
1105     int i;
1106
1107     bzero(stat, sizeof(*stat));
1108     for (i = 0, c = channel_list; c; i++, c = c->next)
1109     {
1110         struct target *t;
1111         if (iochan_getfun(c) != handler)
1112             continue;
1113         t = iochan_getdata(c);
1114         switch (t->state)
1115         {
1116             case No_connection: stat->num_no_connection++; break;
1117             case Connecting: stat->num_connecting++; break;
1118             case Initializing: stat->num_initializing++; break;
1119             case Searching: stat->num_searching++; break;
1120             case Presenting: stat->num_presenting++; break;
1121             case Idle: stat->num_idle++; break;
1122             case Failed: stat->num_failed++; break;
1123             case Error: stat->num_error++; break;
1124             default: break;
1125         }
1126     }
1127     stat->num_hits = s->total_hits;
1128     stat->num_records = s->total_records;
1129
1130     stat->num_connections = i;
1131 }
1132
1133 static CCL_bibset load_cclfile(const char *fn)
1134 {
1135     CCL_bibset res = ccl_qual_mk();
1136     if (ccl_qual_fname(res, fn) < 0)
1137     {
1138         yaz_log(YLOG_FATAL|YLOG_ERRNO, "%s", fn);
1139         exit(1);
1140     }
1141     return res;
1142 }
1143
1144 int main(int argc, char **argv)
1145 {
1146     int ret;
1147     char *arg;
1148
1149     if (signal(SIGPIPE, SIG_IGN) < 0)
1150         yaz_log(YLOG_WARN|YLOG_ERRNO, "signal");
1151
1152     yaz_log_init(YLOG_DEFAULT_LEVEL|YLOG_DEBUG, "pazpar2", 0);
1153
1154     while ((ret = options("c:h:p:C:", argv, argc, &arg)) != -2)
1155     {
1156         switch (ret) {
1157             case 0:
1158                 break;
1159             case 'c':
1160                 command_init(atoi(arg));
1161                 break;
1162             case 'C':
1163                 global_parameters.ccl_filter = load_cclfile(arg);
1164                 break;
1165             case 'h':
1166                 http_init(atoi(arg));
1167                 break;
1168             case 'p':
1169                 http_set_proxyaddr(arg);
1170                 break;
1171             default:
1172                 fprintf(stderr, "Usage: pazpar2 -d comport");
1173                 exit(1);
1174         }
1175
1176     }
1177
1178     if (!global_parameters.ccl_filter)
1179         global_parameters.ccl_filter = load_cclfile("default.bib");
1180
1181     event_loop(&channel_list);
1182
1183     return 0;
1184 }
1185
1186 /*
1187  * Local variables:
1188  * c-basic-offset: 4
1189  * indent-tabs-mode: nil
1190  * End:
1191  * vim: shiftwidth=4 tabstop=8 expandtab
1192  */