src/yaz-proxy.cpp

   1 /*
   2  * Copyright (c) 1998-2003, Index Data.
   3  * See the file LICENSE for details.
   4  *
   5  * $Id: yaz-proxy.cpp,v 1.55 2003-10-10 12:37:26 adam Exp $
   6  */
   7
   8 #include <assert.h>
   9 #include <time.h>
  10
  11 #include <yaz/log.h>
  12 #include <yaz/diagbib1.h>
  13 #include <yaz++/proxy.h>
  14
  15 static const char *apdu_name(Z_APDU *apdu)
  16 {
  17     switch (apdu->which)
  18     {
  19     case Z_APDU_initRequest:
  20         return "initRequest";
  21     case Z_APDU_initResponse:
  22         return "initResponse";
  23     case Z_APDU_searchRequest:
  24         return "searchRequest";
  25     case Z_APDU_searchResponse:
  26         return "searchResponse";
  27     case Z_APDU_presentRequest:
  28         return "presentRequest";
  29     case Z_APDU_presentResponse:
  30         return "presentResponse";
  31     case Z_APDU_deleteResultSetRequest:
  32         return "deleteResultSetRequest";
  33     case Z_APDU_deleteResultSetResponse:
  34         return "deleteResultSetResponse";
  35     case Z_APDU_scanRequest:
  36         return "scanRequest";
  37     case Z_APDU_scanResponse:
  38         return "scanResponse";
  39     case Z_APDU_sortRequest:
  40         return "sortRequest";
  41     case Z_APDU_sortResponse:
  42         return "sortResponse";
  43     case Z_APDU_extendedServicesRequest:
  44         return "extendedServicesRequest";
  45     case Z_APDU_extendedServicesResponse:
  46         return "extendedServicesResponse";
  47     case Z_APDU_close:
  48         return "close";
  49     }
  50     return "other";
  51 }
  52
  53 Yaz_Proxy::Yaz_Proxy(IYaz_PDU_Observable *the_PDU_Observable) :
  54     Yaz_Z_Assoc(the_PDU_Observable), m_bw_stat(60), m_pdu_stat(60)
  55 {
  56     m_PDU_Observable = the_PDU_Observable;
  57     m_client = 0;
  58     m_parent = 0;
  59     m_clientPool = 0;
  60     m_seqno = 1;
  61     m_keepalive_limit_bw = 500000;
  62     m_keepalive_limit_pdu = 1000;
  63     m_proxyTarget = 0;
  64     m_default_target = 0;
  65     m_proxy_authentication = 0;
  66     m_max_clients = 150;
  67     m_seed = time(0);
  68     m_client_idletime = 600;
  69     m_target_idletime = 600;
  70     m_optimize = xstrdup ("1");
  71     strcpy(m_session_str, "x");
  72     m_session_no=0;
  73     m_bytes_sent = m_bytes_recv = 0;
  74     m_bw_hold_PDU = 0;
  75     m_bw_max = 0;
  76     m_pdu_max = 0;
  77     m_max_record_retrieve = 0;
  78     m_reconfig_flag = 0;
  79     m_config_fname = 0;
  80     m_request_no = 0;
  81     m_invalid_session = 0;
  82 }
  83
  84 Yaz_Proxy::~Yaz_Proxy()
  85 {
  86     yaz_log(LOG_LOG, "%sClosed %d/%d sent/recv bytes total", m_session_str,
  87             m_bytes_sent, m_bytes_recv);
  88     xfree (m_proxyTarget);
  89     xfree (m_default_target);
  90     xfree (m_proxy_authentication);
  91     xfree (m_optimize);
  92     if (m_parent)
  93         m_parent->check_reconfigure();
  94 }
  95
  96 int Yaz_Proxy::set_config(const char *config)
  97 {
  98     xfree(m_config_fname);
  99     m_config_fname = xstrdup(config);
 100     int r = m_config.read_xml(config);
 101     return r;
 102 }
 103
 104 void Yaz_Proxy::set_default_target(const char *target)
 105 {
 106     xfree (m_default_target);
 107     m_default_target = 0;
 108     if (target)
 109         m_default_target = (char *) xstrdup (target);
 110 }
 111
 112 void Yaz_Proxy::set_proxy_authentication (const char *auth)
 113 {
 114     xfree (m_proxy_authentication);
 115     m_proxy_authentication = 0;
 116     if (auth)
 117         m_proxy_authentication = (char *) xstrdup (auth);
 118 }
 119
 120 void Yaz_Proxy::check_reconfigure()
 121 {
 122     if (m_reconfig_flag)
 123     {
 124         yaz_log(LOG_LOG, "reconfigure");
 125         yaz_log_reopen();
 126         if (m_config_fname)
 127         {
 128             yaz_log(LOG_LOG, "reconfigure config %s", m_config_fname);
 129             int r = m_config.read_xml(m_config_fname);
 130             if (r)
 131                 yaz_log(LOG_WARN, "reconfigure failed");
 132         }
 133         else
 134             yaz_log(LOG_LOG, "reconfigure");
 135         m_reconfig_flag = 0;
 136     }
 137 }
 138
 139 IYaz_PDU_Observer *Yaz_Proxy::sessionNotify(IYaz_PDU_Observable
 140                                             *the_PDU_Observable, int fd)
 141 {
 142     check_reconfigure();
 143     Yaz_Proxy *new_proxy = new Yaz_Proxy(the_PDU_Observable);
 144     new_proxy->m_parent = this;
 145     new_proxy->m_config = m_config;
 146     new_proxy->m_config_fname = 0;
 147     new_proxy->timeout(m_client_idletime);
 148     new_proxy->m_target_idletime = m_target_idletime;
 149     new_proxy->set_default_target(m_default_target);
 150     new_proxy->set_APDU_log(get_APDU_log());
 151     new_proxy->set_proxy_authentication(m_proxy_authentication);
 152     sprintf(new_proxy->m_session_str, "%ld:%d ", (long) time(0), m_session_no);
 153     m_session_no++;
 154     yaz_log (LOG_LOG, "%sNew session %s", new_proxy->m_session_str,
 155              the_PDU_Observable->getpeername());
 156     return new_proxy;
 157 }
 158
 159 char *Yaz_Proxy::get_cookie(Z_OtherInformation **otherInfo)
 160 {
 161     int oid[OID_SIZE];
 162     Z_OtherInformationUnit *oi;
 163     struct oident ent;
 164     ent.proto = PROTO_Z3950;
 165     ent.oclass = CLASS_USERINFO;
 166     ent.value = (oid_value) VAL_COOKIE;
 167     assert (oid_ent_to_oid (&ent, oid));
 168
 169     if (oid_ent_to_oid (&ent, oid) &&
 170         (oi = update_otherInformation(otherInfo, 0, oid, 1, 1)) &&
 171         oi->which == Z_OtherInfo_characterInfo)
 172         return oi->information.characterInfo;
 173     return 0;
 174 }
 175
 176 char *Yaz_Proxy::get_proxy(Z_OtherInformation **otherInfo)
 177 {
 178     int oid[OID_SIZE];
 179     Z_OtherInformationUnit *oi;
 180     struct oident ent;
 181     ent.proto = PROTO_Z3950;
 182     ent.oclass = CLASS_USERINFO;
 183     ent.value = (oid_value) VAL_PROXY;
 184     if (oid_ent_to_oid (&ent, oid) &&
 185         (oi = update_otherInformation(otherInfo, 0, oid, 1, 1)) &&
 186         oi->which == Z_OtherInfo_characterInfo)
 187         return oi->information.characterInfo;
 188     return 0;
 189 }
 190
 191 const char *Yaz_Proxy::load_balance(const char **url)
 192 {
 193     int zurl_in_use[MAX_ZURL_PLEX];
 194     Yaz_ProxyClient *c;
 195     int i;
 196
 197     for (i = 0; i<MAX_ZURL_PLEX; i++)
 198         zurl_in_use[i] = 0;
 199     for (c = m_parent->m_clientPool; c; c = c->m_next)
 200     {
 201         for (i = 0; url[i]; i++)
 202             if (!strcmp(url[i], c->get_hostname()))
 203                 zurl_in_use[i]++;
 204     }
 205     int min = 100000;
 206     const char *ret = 0;
 207     for (i = 0; url[i]; i++)
 208     {
 209         yaz_log(LOG_DEBUG, "%szurl=%s use=%d",
 210                 m_session_str, url[i], zurl_in_use[i]);
 211         if (min > zurl_in_use[i])
 212         {
 213             ret = url[i];
 214             min = zurl_in_use[i];
 215         }
 216     }
 217     return ret;
 218 }
 219
 220 Yaz_ProxyClient *Yaz_Proxy::get_client(Z_APDU *apdu)
 221 {
 222     assert (m_parent);
 223     Yaz_Proxy *parent = m_parent;
 224     Z_OtherInformation **oi;
 225     Yaz_ProxyClient *c = m_client;
 226
 227     get_otherInfoAPDU(apdu, &oi);
 228     char *cookie = get_cookie(oi);
 229
 230     if (!m_proxyTarget)
 231     {
 232         const char *url[MAX_ZURL_PLEX];
 233         const char *proxy_host = get_proxy(oi);
 234         if (proxy_host)
 235         {
 236             xfree(m_default_target);
 237             m_default_target = xstrdup(proxy_host);
 238             proxy_host = m_default_target;
 239         }
 240
 241         int client_idletime = -1;
 242         m_config.get_target_info(proxy_host, url, &m_bw_max,
 243                                  &m_pdu_max, &m_max_record_retrieve,
 244                                  &m_target_idletime, &client_idletime,
 245                                  &parent->m_max_clients,
 246                                  &m_keepalive_limit_bw, &m_keepalive_limit_pdu);
 247         if (client_idletime != -1)
 248         {
 249             m_client_idletime = client_idletime;
 250             timeout(m_client_idletime);
 251         }
 252         if (!url[0])
 253         {
 254             yaz_log(LOG_LOG, "%sNo default target", m_session_str);
 255             return 0;
 256         }
 257         // we don't handle multiplexing for cookie session, so we just
 258         // pick the first one in this case (anonymous users will be able
 259         // to use any backend)
 260         if (cookie && *cookie)
 261             m_proxyTarget = (char*) xstrdup(url[0]);
 262         else
 263             m_proxyTarget = (char*) xstrdup(load_balance(url));
 264     }
 265     if (cookie && *cookie)
 266     {
 267         Yaz_ProxyClient *cc = 0;
 268
 269         for (c = parent->m_clientPool; c; c = c->m_next)
 270         {
 271             assert (c->m_prev);
 272             assert (*c->m_prev == c);
 273             if (c->m_cookie && !strcmp(cookie,c->m_cookie) &&
 274                 !strcmp(m_proxyTarget, c->get_hostname()))
 275             {
 276                 cc = c;
 277             }
 278         }
 279         if (cc)
 280         {
 281             // found it in cache
 282             c = cc;
 283             // The following handles "cancel"
 284             // If connection is busy (waiting for PDU) and
 285             // we have an initRequest we can safely do re-open
 286             if (c->m_waiting && apdu->which == Z_APDU_initRequest)
 287             {
 288                 yaz_log (LOG_LOG, "%s REOPEN target=%s", m_session_str,
 289                          c->get_hostname());
 290                 c->close();
 291                 c->m_init_flag = 0;
 292
 293                 c->m_last_ok = 0;
 294                 c->m_cache.clear();
 295                 c->m_last_resultCount = 0;
 296                 c->m_sr_transform = 0;
 297                 c->m_waiting = 0;
 298                 c->m_resultSetStartPoint = 0;
 299                 if (c->client(m_proxyTarget))
 300                 {
 301                     delete c;
 302                     return 0;
 303                 }
 304                 c->timeout(m_target_idletime);
 305             }
 306             c->m_seqno = parent->m_seqno;
 307             if (c->m_server && c->m_server != this)
 308                 c->m_server->m_client = 0;
 309             c->m_server = this;
 310             (parent->m_seqno)++;
 311             yaz_log (LOG_DEBUG, "get_client 1 %p %p", this, c);
 312             return c;
 313         }
 314     }
 315     else if (!c)
 316     {
 317         Yaz_ProxyClient *cc = 0;
 318
 319         for (c = parent->m_clientPool; c; c = c->m_next)
 320         {
 321             assert (c->m_prev);
 322             assert (*c->m_prev == c);
 323             if (c->m_server == 0 && c->m_cookie == 0 &&
 324                 !strcmp(m_proxyTarget, c->get_hostname()))
 325             {
 326                 cc = c;
 327             }
 328         }
 329         if (cc)
 330         {
 331             // found it in cache
 332             c = cc;
 333
 334             yaz_log (LOG_LOG, "%sREUSE %d %d %s",
 335                      m_session_str,
 336                      c->m_seqno, parent->m_seqno, c->get_hostname());
 337
 338             c->m_seqno = parent->m_seqno;
 339             assert(c->m_server == 0);
 340             c->m_server = this;
 341
 342             (parent->m_seqno)++;
 343             return c;
 344         }
 345     }
 346     if (!m_client)
 347     {
 348         if (apdu->which != Z_APDU_initRequest)
 349         {
 350             yaz_log (LOG_LOG, "no first INIT!");
 351             return 0;
 352         }
 353         Z_InitRequest *initRequest = apdu->u.initRequest;
 354
 355         if (!initRequest->idAuthentication)
 356         {
 357             if (m_proxy_authentication)
 358             {
 359                 initRequest->idAuthentication =
 360                     (Z_IdAuthentication *)
 361                     odr_malloc (odr_encode(),
 362                                 sizeof(*initRequest->idAuthentication));
 363                 initRequest->idAuthentication->which =
 364                     Z_IdAuthentication_open;
 365                 initRequest->idAuthentication->u.open =
 366                     odr_strdup (odr_encode(), m_proxy_authentication);
 367             }
 368         }
 369
 370         // go through list of clients - and find the lowest/oldest one.
 371         Yaz_ProxyClient *c_min = 0;
 372         int min_seq = -1;
 373         int no_of_clients = 0;
 374         if (parent->m_clientPool)
 375             yaz_log (LOG_DEBUG, "Existing sessions");
 376         for (c = parent->m_clientPool; c; c = c->m_next)
 377         {
 378             yaz_log (LOG_DEBUG, " Session %-3d wait=%d %s cookie=%s", c->m_seqno,
 379                                c->m_waiting, c->get_hostname(),
 380                                c->m_cookie ? c->m_cookie : "");
 381             no_of_clients++;
 382             if (min_seq < 0 || c->m_seqno < min_seq)
 383             {
 384                 min_seq = c->m_seqno;
 385                 c_min = c;
 386             }
 387         }
 388         if (no_of_clients >= parent->m_max_clients)
 389         {
 390             c = c_min;
 391             if (c->m_waiting || strcmp(m_proxyTarget, c->get_hostname()))
 392             {
 393                 yaz_log (LOG_LOG, "%sMAXCLIENTS Destroy %d",
 394                          m_session_str, c->m_seqno);
 395                 if (c->m_server && c->m_server != this)
 396                     delete c->m_server;
 397                 c->m_server = 0;
 398             }
 399             else
 400             {
 401                 yaz_log (LOG_LOG, "%sMAXCLIENTS Reuse %d %d %s",
 402                          m_session_str,
 403                          c->m_seqno, parent->m_seqno, c->get_hostname());
 404                 xfree (c->m_cookie);
 405                 c->m_cookie = 0;
 406                 if (cookie)
 407                     c->m_cookie = xstrdup(cookie);
 408                 c->m_seqno = parent->m_seqno;
 409                 if (c->m_server && c->m_server != this)
 410                 {
 411                     c->m_server->m_client = 0;
 412                     delete c->m_server;
 413                 }
 414                 (parent->m_seqno)++;
 415                 return c;
 416             }
 417         }
 418         else
 419         {
 420             yaz_log (LOG_LOG, "%sNEW %d %s",
 421                      m_session_str, parent->m_seqno, m_proxyTarget);
 422             c = new Yaz_ProxyClient(m_PDU_Observable->clone());
 423             c->m_next = parent->m_clientPool;
 424             if (c->m_next)
 425                 c->m_next->m_prev = &c->m_next;
 426             parent->m_clientPool = c;
 427             c->m_prev = &parent->m_clientPool;
 428         }
 429
 430         xfree (c->m_cookie);
 431         c->m_cookie = 0;
 432         if (cookie)
 433             c->m_cookie = xstrdup(cookie);
 434
 435         c->m_seqno = parent->m_seqno;
 436         c->m_init_flag = 0;
 437         c->m_last_resultCount = 0;
 438         c->m_last_ok = 0;
 439         c->m_cache.clear();
 440         c->m_sr_transform = 0;
 441         c->m_waiting = 0;
 442         c->m_resultSetStartPoint = 0;
 443         (parent->m_seqno)++;
 444         if (c->client(m_proxyTarget))
 445         {
 446             delete c;
 447             return 0;
 448         }
 449         c->timeout(30);
 450
 451     }
 452     yaz_log (LOG_DEBUG, "get_client 3 %p %p", this, c);
 453     return c;
 454 }
 455
 456 void Yaz_Proxy::display_diagrecs(Z_DiagRec **pp, int num)
 457 {
 458     int i;
 459     for (i = 0; i<num; i++)
 460     {
 461         oident *ent;
 462         Z_DefaultDiagFormat *r;
 463         Z_DiagRec *p = pp[i];
 464         if (p->which != Z_DiagRec_defaultFormat)
 465         {
 466             yaz_log(LOG_LOG, "%sError no diagnostics", m_session_str);
 467             return;
 468         }
 469         else
 470             r = p->u.defaultFormat;
 471         if (!(ent = oid_getentbyoid(r->diagnosticSetId)) ||
 472             ent->oclass != CLASS_DIAGSET || ent->value != VAL_BIB1)
 473             yaz_log(LOG_LOG, "%sError unknown diagnostic set", m_session_str);
 474         switch (r->which)
 475         {
 476         case Z_DefaultDiagFormat_v2Addinfo:
 477             yaz_log(LOG_LOG, "%sError %d %s:%s",
 478                     m_session_str,
 479                     *r->condition, diagbib1_str(*r->condition),
 480                     r->u.v2Addinfo);
 481             break;
 482         case Z_DefaultDiagFormat_v3Addinfo:
 483             yaz_log(LOG_LOG, "%sError %d %s:%s",
 484                     m_session_str,
 485                     *r->condition, diagbib1_str(*r->condition),
 486                     r->u.v3Addinfo);
 487             break;
 488         }
 489     }
 490 }
 491
 492 int Yaz_Proxy::send_to_client(Z_APDU *apdu)
 493 {
 494     int len = 0;
 495     if (apdu->which == Z_APDU_searchResponse)
 496     {
 497         Z_SearchResponse *sr = apdu->u.searchResponse;
 498         Z_Records *p = sr->records;
 499         if (p && p->which == Z_Records_NSD)
 500         {
 501             Z_DiagRec dr, *dr_p = &dr;
 502             dr.which = Z_DiagRec_defaultFormat;
 503             dr.u.defaultFormat = p->u.nonSurrogateDiagnostic;
 504
 505             display_diagrecs(&dr_p, 1);
 506         }
 507         else
 508         {
 509             if (sr->resultCount)
 510             {
 511                 yaz_log(LOG_LOG, "%s%d hits", m_session_str,
 512                         *sr->resultCount);
 513                 if (*sr->resultCount < 0)
 514                     m_invalid_session = 1;
 515             }
 516         }
 517     }
 518     else if (apdu->which == Z_APDU_presentResponse)
 519     {
 520         Z_PresentResponse *sr = apdu->u.presentResponse;
 521         Z_Records *p = sr->records;
 522         if (p && p->which == Z_Records_NSD)
 523         {
 524             Z_DiagRec dr, *dr_p = &dr;
 525             dr.which = Z_DiagRec_defaultFormat;
 526             dr.u.defaultFormat = p->u.nonSurrogateDiagnostic;
 527
 528             display_diagrecs(&dr_p, 1);
 529         }
 530     }
 531     int r = send_Z_PDU(apdu, &len);
 532     yaz_log (LOG_DEBUG, "%sSending %s to client %d bytes", m_session_str,
 533              apdu_name(apdu), len);
 534     m_bytes_sent += len;
 535     m_bw_stat.add_bytes(len);
 536     return r;
 537 }
 538
 539 int Yaz_ProxyClient::send_to_target(Z_APDU *apdu)
 540 {
 541     int len = 0;
 542     int r = send_Z_PDU(apdu, &len);
 543     yaz_log (LOG_DEBUG, "%sSending %s to %s %d bytes",
 544              get_session_str(),
 545              apdu_name(apdu), get_hostname(), len);
 546     m_bytes_sent += len;
 547     return r;
 548 }
 549
 550 Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
 551 {
 552     if (*m_parent->m_optimize == '0')
 553         return apdu;      // don't optimize result sets..
 554     if (apdu->which == Z_APDU_presentRequest)
 555     {
 556         Z_PresentRequest *pr = apdu->u.presentRequest;
 557         Z_NamePlusRecordList *npr;
 558         int toget = *pr->numberOfRecordsRequested;
 559         int start = *pr->resultSetStartPoint;
 560
 561         if (m_client->m_last_resultSetId &&
 562             !strcmp(m_client->m_last_resultSetId, pr->resultSetId))
 563         {
 564             if (m_client->m_cache.lookup (odr_encode(), &npr, start, toget,
 565                                           pr->preferredRecordSyntax,
 566                                           pr->recordComposition))
 567             {
 568                 yaz_log (LOG_LOG, "%sReturned cached records for present request",
 569                          m_session_str);
 570                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentResponse);
 571                 new_apdu->u.presentResponse->referenceId = pr->referenceId;
 572
 573                 new_apdu->u.presentResponse->numberOfRecordsReturned
 574                     = odr_intdup(odr_encode(), toget);
 575
 576                 new_apdu->u.presentResponse->records = (Z_Records*)
 577                     odr_malloc(odr_encode(), sizeof(Z_Records));
 578                 new_apdu->u.presentResponse->records->which = Z_Records_DBOSD;
 579                 new_apdu->u.presentResponse->records->u.databaseOrSurDiagnostics = npr;
 580                 new_apdu->u.presentResponse->nextResultSetPosition =
 581                     odr_intdup(odr_encode(), start+toget);
 582
 583                 send_to_client(new_apdu);
 584                 return 0;
 585             }
 586         }
 587     }
 588
 589     if (apdu->which != Z_APDU_searchRequest)
 590         return apdu;
 591     Z_SearchRequest *sr = apdu->u.searchRequest;
 592     Yaz_Z_Query *this_query = new Yaz_Z_Query;
 593     Yaz_Z_Databases this_databases;
 594
 595     this_databases.set(sr->num_databaseNames, (const char **)
 596                        sr->databaseNames);
 597
 598     this_query->set_Z_Query(sr->query);
 599
 600     char query_str[80];
 601     this_query->print(query_str, sizeof(query_str)-1);
 602     yaz_log(LOG_LOG, "%sQuery %s", m_session_str, query_str);
 603
 604     if (m_client->m_last_ok && m_client->m_last_query &&
 605         m_client->m_last_query->match(this_query) &&
 606         !strcmp(m_client->m_last_resultSetId, sr->resultSetName) &&
 607         m_client->m_last_databases.match(this_databases))
 608     {
 609         delete this_query;
 610         if (m_client->m_last_resultCount > *sr->smallSetUpperBound &&
 611             m_client->m_last_resultCount < *sr->largeSetLowerBound)
 612         {
 613             Z_NamePlusRecordList *npr;
 614             int toget = *sr->mediumSetPresentNumber;
 615             Z_RecordComposition *comp = 0;
 616
 617             if (toget > m_client->m_last_resultCount)
 618                 toget = m_client->m_last_resultCount;
 619
 620             if (sr->mediumSetElementSetNames)
 621             {
 622                 comp = (Z_RecordComposition *)
 623                     odr_malloc(odr_encode(), sizeof(Z_RecordComposition));
 624                 comp->which = Z_RecordComp_simple;
 625                 comp->u.simple = sr->mediumSetElementSetNames;
 626             }
 627
 628             if (m_client->m_cache.lookup (odr_encode(), &npr, 1, toget,
 629                                           sr->preferredRecordSyntax, comp))
 630             {
 631                 yaz_log (LOG_LOG, "%sReturned cached records for medium set",
 632                          m_session_str);
 633                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
 634                 new_apdu->u.searchResponse->referenceId = sr->referenceId;
 635                 new_apdu->u.searchResponse->resultCount =
 636                     &m_client->m_last_resultCount;
 637
 638                 new_apdu->u.searchResponse->numberOfRecordsReturned
 639                     = odr_intdup(odr_encode(), toget);
 640
 641                 new_apdu->u.searchResponse->presentStatus =
 642                     odr_intdup(odr_encode(), Z_PresentStatus_success);
 643                 new_apdu->u.searchResponse->records = (Z_Records*)
 644                     odr_malloc(odr_encode(), sizeof(Z_Records));
 645                 new_apdu->u.searchResponse->records->which = Z_Records_DBOSD;
 646                 new_apdu->u.searchResponse->records->u.databaseOrSurDiagnostics = npr;
 647                 new_apdu->u.searchResponse->nextResultSetPosition =
 648                     odr_intdup(odr_encode(), toget+1);
 649                 send_to_client(new_apdu);
 650                 return 0;
 651             }
 652             else
 653             {
 654                 // medium Set
 655                 // send present request (medium size)
 656                 yaz_log (LOG_LOG, "%sOptimizing search for medium set",
 657                          m_session_str);
 658
 659                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentRequest);
 660                 Z_PresentRequest *pr = new_apdu->u.presentRequest;
 661                 pr->referenceId = sr->referenceId;
 662                 pr->resultSetId = sr->resultSetName;
 663                 pr->preferredRecordSyntax = sr->preferredRecordSyntax;
 664                 *pr->numberOfRecordsRequested = toget;
 665                 pr->recordComposition = comp;
 666                 m_client->m_sr_transform = 1;
 667                 return new_apdu;
 668             }
 669         }
 670         else if (m_client->m_last_resultCount >= *sr->largeSetLowerBound ||
 671             m_client->m_last_resultCount <= 0)
 672         {
 673             // large set. Return pseudo-search response immediately
 674             yaz_log (LOG_LOG, "%sOptimizing search for large set",
 675                      m_session_str);
 676             Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
 677             new_apdu->u.searchResponse->referenceId = sr->referenceId;
 678             new_apdu->u.searchResponse->resultCount =
 679                 &m_client->m_last_resultCount;
 680             send_to_client(new_apdu);
 681             return 0;
 682         }
 683         else
 684         {
 685             Z_NamePlusRecordList *npr;
 686             int toget = m_client->m_last_resultCount;
 687             Z_RecordComposition *comp = 0;
 688             // small set
 689             // send a present request (small set)
 690
 691             if (sr->smallSetElementSetNames)
 692             {
 693                 comp = (Z_RecordComposition *)
 694                     odr_malloc(odr_encode(), sizeof(Z_RecordComposition));
 695                 comp->which = Z_RecordComp_simple;
 696                 comp->u.simple = sr->smallSetElementSetNames;
 697             }
 698
 699             if (m_client->m_cache.lookup (odr_encode(), &npr, 1, toget,
 700                                           sr->preferredRecordSyntax, comp))
 701             {
 702                 yaz_log (LOG_LOG, "%sReturned cached records for small set",
 703                          m_session_str);
 704                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
 705                 new_apdu->u.searchResponse->referenceId = sr->referenceId;
 706                 new_apdu->u.searchResponse->resultCount =
 707                     &m_client->m_last_resultCount;
 708
 709                 new_apdu->u.searchResponse->numberOfRecordsReturned
 710                     = odr_intdup(odr_encode(), toget);
 711
 712                 new_apdu->u.searchResponse->presentStatus =
 713                     odr_intdup(odr_encode(), Z_PresentStatus_success);
 714                 new_apdu->u.searchResponse->records = (Z_Records*)
 715                     odr_malloc(odr_encode(), sizeof(Z_Records));
 716                 new_apdu->u.searchResponse->records->which = Z_Records_DBOSD;
 717                 new_apdu->u.searchResponse->records->u.databaseOrSurDiagnostics = npr;
 718                 new_apdu->u.searchResponse->nextResultSetPosition =
 719                     odr_intdup(odr_encode(), toget+1);
 720                 send_to_client(new_apdu);
 721                 return 0;
 722             }
 723             else
 724             {
 725                 yaz_log (LOG_LOG, "%sOptimizing search for small set",
 726                          m_session_str);
 727                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentRequest);
 728                 Z_PresentRequest *pr = new_apdu->u.presentRequest;
 729                 pr->referenceId = sr->referenceId;
 730                 pr->resultSetId = sr->resultSetName;
 731                 pr->preferredRecordSyntax = sr->preferredRecordSyntax;
 732                 *pr->numberOfRecordsRequested = toget;
 733                 pr->recordComposition = comp;
 734                 m_client->m_sr_transform = 1;
 735                 return new_apdu;
 736             }
 737         }
 738     }
 739     else  // query doesn't match
 740     {
 741         delete m_client->m_last_query;
 742         m_client->m_last_query = this_query;
 743         m_client->m_last_ok = 0;
 744         m_client->m_cache.clear();
 745         m_client->m_resultSetStartPoint = 0;
 746
 747         xfree (m_client->m_last_resultSetId);
 748         m_client->m_last_resultSetId = xstrdup (sr->resultSetName);
 749
 750         m_client->m_last_databases.set(sr->num_databaseNames,
 751                                        (const char **) sr->databaseNames);
 752     }
 753     return apdu;
 754 }
 755
 756
 757 void Yaz_Proxy::recv_Z_PDU(Z_APDU *apdu, int len)
 758 {
 759     char *cp = strchr(m_session_str, ' ');
 760     m_request_no++;
 761     if (cp)
 762         sprintf(cp+1, "%d ", m_request_no);
 763
 764     int reduce = 0;
 765     m_bytes_recv += len;
 766
 767     yaz_log (LOG_DEBUG, "%sReceiving %s from client %d bytes", m_session_str,
 768              apdu_name(apdu), len);
 769
 770     if (m_bw_hold_PDU)     // double incoming PDU. shutdown now.
 771         shutdown();
 772
 773     m_bw_stat.add_bytes(len);
 774     m_pdu_stat.add_bytes(1);
 775
 776     int bw_total = m_bw_stat.get_total();
 777     int pdu_total = m_pdu_stat.get_total();
 778
 779     yaz_log(LOG_LOG, "%sstat bw=%d pdu=%d limit-bw=%d limit-pdu=%d",
 780             m_session_str, bw_total, pdu_total, m_bw_max, m_pdu_max);
 781     if (m_bw_max)
 782     {
 783         if (bw_total > m_bw_max)
 784         {
 785             reduce = (bw_total/m_bw_max);
 786         }
 787     }
 788     if (m_pdu_max)
 789     {
 790         if (pdu_total > m_pdu_max)
 791         {
 792             int nreduce = (60/m_pdu_max);
 793             reduce = (reduce > nreduce) ? reduce : nreduce;
 794         }
 795     }
 796     if (reduce)
 797     {
 798         yaz_log(LOG_LOG, "%sLimit delay=%d", m_session_str, reduce);
 799         m_bw_hold_PDU = apdu;  // save PDU and signal "on hold"
 800         timeout(reduce);       // call us reduce seconds later
 801     }
 802     else
 803         recv_Z_PDU_0(apdu);    // all fine. Proceed receive PDU as usual
 804 }
 805
 806 void Yaz_Proxy::handle_max_record_retrieve(Z_APDU *apdu)
 807 {
 808     if (m_max_record_retrieve)
 809     {
 810         if (apdu->which == Z_APDU_presentRequest)
 811         {
 812             Z_PresentRequest *pr = apdu->u.presentRequest;
 813             if (pr->numberOfRecordsRequested &&
 814                 *pr->numberOfRecordsRequested > m_max_record_retrieve)
 815                 *pr->numberOfRecordsRequested = m_max_record_retrieve;
 816         }
 817     }
 818 }
 819
 820 Z_Records *Yaz_Proxy::create_nonSurrogateDiagnostics(ODR odr,
 821                                                      int error,
 822                                                      const char *addinfo)
 823 {
 824     Z_Records *rec = (Z_Records *)
 825         odr_malloc (odr, sizeof(*rec));
 826     int *err = (int *)
 827         odr_malloc (odr, sizeof(*err));
 828     Z_DiagRec *drec = (Z_DiagRec *)
 829         odr_malloc (odr, sizeof(*drec));
 830     Z_DefaultDiagFormat *dr = (Z_DefaultDiagFormat *)
 831         odr_malloc (odr, sizeof(*dr));
 832     *err = error;
 833     rec->which = Z_Records_NSD;
 834     rec->u.nonSurrogateDiagnostic = dr;
 835     dr->diagnosticSetId =
 836         yaz_oidval_to_z3950oid (odr, CLASS_DIAGSET, VAL_BIB1);
 837     dr->condition = err;
 838     dr->which = Z_DefaultDiagFormat_v2Addinfo;
 839     dr->u.v2Addinfo = odr_strdup (odr, addinfo ? addinfo : "");
 840     return rec;
 841 }
 842
 843 Z_APDU *Yaz_Proxy::handle_query_validation(Z_APDU *apdu)
 844 {
 845     if (apdu->which == Z_APDU_searchRequest)
 846     {
 847         Z_SearchRequest *sr = apdu->u.searchRequest;
 848         int err;
 849         char *addinfo = 0;
 850         err = m_config.check_query(odr_encode(), m_default_target, sr->query,
 851                                    &addinfo);
 852         if (err)
 853         {
 854             Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
 855
 856             new_apdu->u.searchResponse->referenceId = sr->referenceId;
 857             new_apdu->u.searchResponse->records =
 858                 create_nonSurrogateDiagnostics(odr_encode(), err, addinfo);
 859             *new_apdu->u.searchResponse->searchStatus = 0;
 860
 861             send_to_client(new_apdu);
 862
 863             return 0;
 864         }
 865     }
 866     return apdu;
 867 }
 868
 869 Z_APDU *Yaz_Proxy::handle_syntax_validation(Z_APDU *apdu)
 870 {
 871     if (apdu->which == Z_APDU_searchRequest)
 872     {
 873         Z_SearchRequest *sr = apdu->u.searchRequest;
 874         if (*sr->smallSetUpperBound > 0 || *sr->largeSetLowerBound > 1)
 875         {
 876             int err;
 877             char *addinfo = 0;
 878             err = m_config.check_syntax(odr_encode(), m_default_target,
 879                                         sr->preferredRecordSyntax,
 880                                         &addinfo);
 881             if (err)
 882             {
 883                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
 884
 885                 new_apdu->u.searchResponse->referenceId = sr->referenceId;
 886                 new_apdu->u.searchResponse->records =
 887                     create_nonSurrogateDiagnostics(odr_encode(), err, addinfo);
 888                 *new_apdu->u.searchResponse->searchStatus = 0;
 889
 890                 send_to_client(new_apdu);
 891
 892                 return 0;
 893             }
 894         }
 895     }
 896     else if (apdu->which == Z_APDU_presentRequest)
 897     {
 898         Z_PresentRequest *pr = apdu->u.presentRequest;
 899         int err;
 900         char *addinfo = 0;
 901         err = m_config.check_syntax(odr_encode(), m_default_target,
 902                                     pr->preferredRecordSyntax,
 903                                     &addinfo);
 904         if (err)
 905         {
 906             Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentResponse);
 907
 908             new_apdu->u.presentResponse->referenceId = pr->referenceId;
 909             new_apdu->u.presentResponse->records =
 910                 create_nonSurrogateDiagnostics(odr_encode(), err, addinfo);
 911             *new_apdu->u.presentResponse->presentStatus =
 912                 Z_PresentStatus_failure;
 913
 914             send_to_client(new_apdu);
 915
 916             return 0;
 917         }
 918     }
 919     return apdu;
 920 }
 921
 922 void Yaz_Proxy::recv_Z_PDU_0(Z_APDU *apdu)
 923 {
 924     // Determine our client.
 925     m_client = get_client(apdu);
 926     if (!m_client)
 927     {
 928         delete this;
 929         return;
 930     }
 931     m_client->m_server = this;
 932
 933     if (apdu->which == Z_APDU_initRequest)
 934     {
 935         if (apdu->u.initRequest->implementationId)
 936             yaz_log(LOG_LOG, "%simplementationId: %s",
 937                     m_session_str, apdu->u.initRequest->implementationId);
 938         if (apdu->u.initRequest->implementationName)
 939             yaz_log(LOG_LOG, "%simplementationName: %s",
 940                     m_session_str, apdu->u.initRequest->implementationName);
 941         if (apdu->u.initRequest->implementationVersion)
 942             yaz_log(LOG_LOG, "%simplementationVersion: %s",
 943                     m_session_str, apdu->u.initRequest->implementationVersion);
 944         if (m_client->m_init_flag)
 945         {
 946             Z_APDU *apdu = m_client->m_initResponse;
 947             apdu->u.initResponse->otherInfo = 0;
 948             if (m_client->m_cookie && *m_client->m_cookie)
 949                 set_otherInformationString(apdu, VAL_COOKIE, 1,
 950                                            m_client->m_cookie);
 951             send_to_client(apdu);
 952             return;
 953         }
 954         m_client->m_init_flag = 1;
 955     }
 956     handle_max_record_retrieve(apdu);
 957
 958     if (apdu)
 959         apdu = handle_syntax_validation(apdu);
 960
 961     if (apdu)
 962         apdu = handle_query_validation(apdu);
 963
 964     if (apdu)
 965         apdu = result_set_optimize(apdu);
 966     if (!apdu)
 967     {
 968         m_client->timeout(m_target_idletime);  // mark it active even
 969         // though we didn't use it
 970         return;
 971     }
 972
 973     // delete other info part from PDU before sending to target
 974     Z_OtherInformation **oi;
 975     get_otherInfoAPDU(apdu, &oi);
 976     if (oi)
 977         *oi = 0;
 978
 979     if (apdu->which == Z_APDU_presentRequest &&
 980         m_client->m_resultSetStartPoint == 0)
 981     {
 982         Z_PresentRequest *pr = apdu->u.presentRequest;
 983         m_client->m_resultSetStartPoint = *pr->resultSetStartPoint;
 984         m_client->m_cache.copy_presentRequest(apdu->u.presentRequest);
 985     } else {
 986         m_client->m_resultSetStartPoint = 0;
 987     }
 988     if (m_client->send_to_target(apdu) < 0)
 989     {
 990         delete m_client;
 991         m_client = 0;
 992         delete this;
 993     }
 994     else
 995         m_client->m_waiting = 1;
 996 }
 997
 998 void Yaz_Proxy::connectNotify()
 999 {
1000 }
1001
1002 void Yaz_Proxy::shutdown()
1003 {
1004     // only keep if keep_alive flag is set...
1005     if (m_client &&
1006         !m_invalid_session &&
1007         m_client->m_pdu_recv < m_keepalive_limit_pdu &&
1008         m_client->m_bytes_recv+m_client->m_bytes_sent < m_keepalive_limit_bw &&
1009         m_client->m_waiting == 0)
1010     {
1011         yaz_log(LOG_LOG, "%sShutdown (client to proxy) keepalive %s",
1012                  m_session_str,
1013                  m_client->get_hostname());
1014         yaz_log(LOG_LOG, "%sbw=%d pdu=%d limit-bw=%d limit-pdu=%d",
1015                 m_session_str, m_client->m_pdu_recv,
1016                 m_client->m_bytes_sent + m_client->m_bytes_recv,
1017                 m_keepalive_limit_bw, m_keepalive_limit_pdu);
1018         assert (m_client->m_waiting != 2);
1019         // Tell client (if any) that no server connection is there..
1020         m_client->m_server = 0;
1021     }
1022     else if (m_client)
1023     {
1024         yaz_log (LOG_LOG, "%sShutdown (client to proxy) close %s",
1025                  m_session_str,
1026                  m_client->get_hostname());
1027         assert (m_client->m_waiting != 2);
1028         delete m_client;
1029     }
1030     else if (!m_parent)
1031     {
1032         yaz_log (LOG_LOG, "%sshutdown (client to proxy) bad state",
1033                  m_session_str);
1034         assert (m_parent);
1035     }
1036     else
1037     {
1038         yaz_log (LOG_LOG, "%sShutdown (client to proxy)",
1039                  m_session_str);
1040     }
1041     delete this;
1042 }
1043
1044 const char *Yaz_ProxyClient::get_session_str()
1045 {
1046     if (!m_server)
1047         return "0";
1048     return m_server->get_session_str();
1049 }
1050
1051 void Yaz_ProxyClient::shutdown()
1052 {
1053     yaz_log (LOG_LOG, "%sShutdown (proxy to target) %s", get_session_str(),
1054              get_hostname());
1055     delete m_server;
1056     delete this;
1057 }
1058
1059 void Yaz_Proxy::failNotify()
1060 {
1061     yaz_log (LOG_LOG, "%sConnection closed by client",
1062              get_session_str());
1063     shutdown();
1064 }
1065
1066 void Yaz_ProxyClient::failNotify()
1067 {
1068     yaz_log (LOG_LOG, "%sConnection closed by target %s",
1069              get_session_str(), get_hostname());
1070     shutdown();
1071 }
1072
1073 void Yaz_ProxyClient::connectNotify()
1074 {
1075     yaz_log (LOG_LOG, "%sConnection accepted by %s", get_session_str(),
1076              get_hostname());
1077     int to;
1078     if (m_server)
1079         to = m_server->get_target_idletime();
1080     else
1081         to = 600;
1082     timeout(to);
1083 }
1084
1085 IYaz_PDU_Observer *Yaz_ProxyClient::sessionNotify(IYaz_PDU_Observable
1086                                                   *the_PDU_Observable, int fd)
1087 {
1088     return new Yaz_ProxyClient(the_PDU_Observable);
1089 }
1090
1091 Yaz_ProxyClient::~Yaz_ProxyClient()
1092 {
1093     if (m_prev)
1094         *m_prev = m_next;
1095     if (m_next)
1096         m_next->m_prev = m_prev;
1097     m_waiting = 2;     // for debugging purposes only.
1098     odr_destroy(m_init_odr);
1099     delete m_last_query;
1100     xfree (m_last_resultSetId);
1101     xfree (m_cookie);
1102 }
1103
1104 void Yaz_Proxy::timeoutNotify()
1105 {
1106     if (m_bw_hold_PDU)
1107     {
1108         timeout(m_client_idletime);
1109         Z_APDU *apdu = m_bw_hold_PDU;
1110         m_bw_hold_PDU = 0;
1111         recv_Z_PDU_0(apdu);
1112     }
1113     else
1114     {
1115         yaz_log (LOG_LOG, "%sTimeout (client to proxy)", m_session_str);
1116         shutdown();
1117     }
1118 }
1119
1120 void Yaz_ProxyClient::timeoutNotify()
1121 {
1122     yaz_log (LOG_LOG, "%sTimeout (proxy to target) %s", get_session_str(),
1123              get_hostname());
1124     shutdown();
1125 }
1126
1127 Yaz_ProxyClient::Yaz_ProxyClient(IYaz_PDU_Observable *the_PDU_Observable) :
1128     Yaz_Z_Assoc (the_PDU_Observable)
1129 {
1130     m_cookie = 0;
1131     m_next = 0;
1132     m_prev = 0;
1133     m_init_flag = 0;
1134     m_last_query = 0;
1135     m_last_resultSetId = 0;
1136     m_last_resultCount = 0;
1137     m_last_ok = 0;
1138     m_sr_transform = 0;
1139     m_waiting = 0;
1140     m_init_odr = odr_createmem (ODR_DECODE);
1141     m_initResponse = 0;
1142     m_resultSetStartPoint = 0;
1143     m_bytes_sent = m_bytes_recv = 0;
1144     m_pdu_recv = 0;
1145 }
1146
1147 const char *Yaz_Proxy::option(const char *name, const char *value)
1148 {
1149     if (!strcmp (name, "optimize")) {
1150         if (value) {
1151             xfree (m_optimize);
1152             m_optimize = xstrdup (value);
1153         }
1154         return m_optimize;
1155     }
1156     return 0;
1157 }
1158
1159 void Yaz_ProxyClient::recv_Z_PDU(Z_APDU *apdu, int len)
1160 {
1161     m_bytes_recv += len;
1162     m_pdu_recv++;
1163     m_waiting = 0;
1164     yaz_log (LOG_DEBUG, "%sReceiving %s from %s %d bytes", get_session_str(),
1165              apdu_name(apdu), get_hostname(), len);
1166     if (apdu->which == Z_APDU_initResponse)
1167     {
1168         NMEM nmem = odr_extract_mem (odr_decode());
1169         odr_reset (m_init_odr);
1170         nmem_transfer (m_init_odr->mem, nmem);
1171         m_initResponse = apdu;
1172
1173         Z_InitResponse *ir = apdu->u.initResponse;
1174         char *im0 = ir->implementationName;
1175
1176         char *im1 = (char*)
1177             odr_malloc(m_init_odr, 20 + (im0 ? strlen(im0) : 0));
1178         *im1 = '\0';
1179         if (im0)
1180         {
1181             strcat(im1, im0);
1182             strcat(im1, " ");
1183         }
1184         strcat(im1, "(YAZ Proxy)");
1185         ir->implementationName = im1;
1186
1187         nmem_destroy (nmem);
1188     }
1189     if (apdu->which == Z_APDU_searchResponse)
1190     {
1191         Z_SearchResponse *sr = apdu->u.searchResponse;
1192         m_last_resultCount = *sr->resultCount;
1193         int status = *sr->searchStatus;
1194         if (status && (!sr->records || sr->records->which == Z_Records_DBOSD))
1195         {
1196             m_last_ok = 1;
1197
1198             if (sr->records && sr->records->which == Z_Records_DBOSD)
1199             {
1200                 m_cache.add(odr_decode(),
1201                             sr->records->u.databaseOrSurDiagnostics, 1,
1202                             *sr->resultCount);
1203             }
1204         }
1205     }
1206     if (apdu->which == Z_APDU_presentResponse)
1207     {
1208         Z_PresentResponse *pr = apdu->u.presentResponse;
1209         if (m_sr_transform)
1210         {
1211             m_sr_transform = 0;
1212             Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
1213             Z_SearchResponse *sr = new_apdu->u.searchResponse;
1214             sr->referenceId = pr->referenceId;
1215             *sr->resultCount = m_last_resultCount;
1216             sr->records = pr->records;
1217             sr->nextResultSetPosition = pr->nextResultSetPosition;
1218             sr->numberOfRecordsReturned = pr->numberOfRecordsReturned;
1219             apdu = new_apdu;
1220         }
1221         if (pr->records &&
1222             pr->records->which == Z_Records_DBOSD && m_resultSetStartPoint)
1223         {
1224             m_cache.add(odr_decode(),
1225                         pr->records->u.databaseOrSurDiagnostics,
1226                         m_resultSetStartPoint, -1);
1227             m_resultSetStartPoint = 0;
1228         }
1229     }
1230     if (m_cookie)
1231         set_otherInformationString (apdu, VAL_COOKIE, 1, m_cookie);
1232     if (m_server)
1233     {
1234         m_server->send_to_client(apdu);
1235     }
1236     if (apdu->which == Z_APDU_close)
1237     {
1238         shutdown();
1239     }
1240 }