pazpar2.c

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