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