pazpar2.c

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