e7bb2a0b9275d561e9398e4d58f4963c25f16aaf
[pazpar2-moved-to-github.git] / src / client.c
1 /* This file is part of Pazpar2.
2    Copyright (C) 2006-2011 Index Data
3
4 Pazpar2 is free software; you can redistribute it and/or modify it under
5 the terms of the GNU General Public License as published by the Free
6 Software Foundation; either version 2, or (at your option) any later
7 version.
8
9 Pazpar2 is distributed in the hope that it will be useful, but WITHOUT ANY
10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
12 for more details.
13
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
17
18 */
19
20 /** \file client.c
21     \brief Z39.50 client 
22 */
23
24 #if HAVE_CONFIG_H
25 #include <config.h>
26 #endif
27 #include <stdlib.h>
28 #include <stdio.h>
29 #include <string.h>
30 #if HAVE_SYS_TIME_H
31 #include <sys/time.h>
32 #endif
33 #if HAVE_UNISTD_H
34 #include <unistd.h>
35 #endif
36 #include <signal.h>
37 #include <assert.h>
38
39 #include <yaz/marcdisp.h>
40 #include <yaz/comstack.h>
41 #include <yaz/tcpip.h>
42 #include <yaz/proto.h>
43 #include <yaz/readconf.h>
44 #include <yaz/pquery.h>
45 #include <yaz/otherinfo.h>
46 #include <yaz/yaz-util.h>
47 #include <yaz/nmem.h>
48 #include <yaz/query-charset.h>
49 #include <yaz/querytowrbuf.h>
50 #include <yaz/oid_db.h>
51 #include <yaz/diagbib1.h>
52 #include <yaz/snprintf.h>
53 #include <yaz/rpn2cql.h>
54 #include <yaz/rpn2solr.h>
55
56 #define USE_TIMING 0
57 #if USE_TIMING
58 #include <yaz/timing.h>
59 #endif
60
61 #include "ppmutex.h"
62 #include "session.h"
63 #include "parameters.h"
64 #include "client.h"
65 #include "connection.h"
66 #include "settings.h"
67 #include "relevance.h"
68 #include "incref.h"
69
70 /* client counting (1) , disable client counting (0) */
71 #if 1
72 static YAZ_MUTEX g_mutex = 0;
73 static int no_clients = 0;
74
75 static void client_use(int delta)
76 {
77     if (!g_mutex)
78         yaz_mutex_create(&g_mutex);
79     yaz_mutex_enter(g_mutex);
80     no_clients += delta;
81     yaz_mutex_leave(g_mutex);
82     yaz_log(YLOG_DEBUG, "%s clients=%d", delta > 0 ? "INC" : "DEC", no_clients);
83 }
84 #else
85 #define client_use(x)
86 #endif
87
88 /** \brief Represents client state for a connection to one search target */
89 struct client {
90     struct session_database *database;
91     struct connection *connection;
92     struct session *session;
93     char *pquery; // Current search
94     char *cqlquery; // used for SRU targets only
95     Odr_int hits;
96     int record_offset;
97     int maxrecs;
98     int startrecs;
99     int diagnostic;
100     int preferred;
101     enum client_state state;
102     struct show_raw *show_raw;
103     ZOOM_resultset resultset;
104     YAZ_MUTEX mutex;
105     int ref_count;
106 };
107
108 struct show_raw {
109     int active; // whether this request has been sent to the server
110     int position;
111     int binary;
112     char *syntax;
113     char *esn;
114     void (*error_handler)(void *data, const char *addinfo);
115     void (*record_handler)(void *data, const char *buf, size_t sz);
116     void *data;
117     struct show_raw *next;
118 };
119
120 static const char *client_states[] = {
121     "Client_Connecting",
122     "Client_Idle",
123     "Client_Working",
124     "Client_Error",
125     "Client_Failed",
126     "Client_Disconnected"
127 };
128
129 const char *client_get_state_str(struct client *cl)
130 {
131     return client_states[cl->state];
132 }
133
134 enum client_state client_get_state(struct client *cl)
135 {
136     return cl->state;
137 }
138
139 void client_set_state(struct client *cl, enum client_state st)
140 {
141     int was_active = 0;
142     if (client_is_active(cl))
143         was_active = 1;
144     cl->state = st;
145     /* If client is going from being active to inactive and all clients
146        are now idle we fire a watch for the session . The assumption is
147        that session is not mutex locked if client is already active */
148     if (was_active && !client_is_active(cl) && cl->session)
149     {
150
151         int no_active = session_active_clients(cl->session);
152         yaz_log(YLOG_DEBUG, "%s: releasing watches on zero active: %d", client_get_url(cl), no_active);
153         if (no_active == 0) {
154             session_alert_watch(cl->session, SESSION_WATCH_SHOW);
155             session_alert_watch(cl->session, SESSION_WATCH_SHOW_PREF);
156         }
157     }
158 }
159
160 static void client_show_raw_error(struct client *cl, const char *addinfo);
161
162 struct connection *client_get_connection(struct client *cl)
163 {
164     return cl->connection;
165 }
166
167 struct session_database *client_get_database(struct client *cl)
168 {
169     return cl->database;
170 }
171
172 struct session *client_get_session(struct client *cl)
173 {
174     return cl->session;
175 }
176
177 const char *client_get_pquery(struct client *cl)
178 {
179     return cl->pquery;
180 }
181
182 static void client_send_raw_present(struct client *cl);
183 static int nativesyntax_to_type(struct session_database *sdb, char *type,
184                                 ZOOM_record rec);
185
186 static void client_show_immediate(
187     ZOOM_resultset resultset, struct session_database *sdb, int position,
188     void *data,
189     void (*error_handler)(void *data, const char *addinfo),
190     void (*record_handler)(void *data, const char *buf, size_t sz),
191     int binary)
192 {
193     ZOOM_record rec = 0;
194     char type[80];
195     const char *buf;
196     int len;
197
198     if (!resultset)
199     {
200         error_handler(data, "no resultset");
201         return;
202     }
203     rec = ZOOM_resultset_record(resultset, position-1);
204     if (!rec)
205     {
206         error_handler(data, "no record");
207         return;
208     }
209     if (binary)
210         strcpy(type, "raw");
211     else
212         nativesyntax_to_type(sdb, type, rec);
213     buf = ZOOM_record_get(rec, type, &len);
214     if (!buf)
215     {
216         error_handler(data, "no record");
217         return;
218     }
219     record_handler(data, buf, len);
220 }
221
222
223 int client_show_raw_begin(struct client *cl, int position,
224                           const char *syntax, const char *esn,
225                           void *data,
226                           void (*error_handler)(void *data, const char *addinfo),
227                           void (*record_handler)(void *data, const char *buf,
228                                                  size_t sz),
229                           int binary)
230 {
231     if (syntax == 0 && esn == 0)
232         client_show_immediate(cl->resultset, client_get_database(cl),
233                               position, data,
234                               error_handler, record_handler,
235                               binary);
236     else
237     {
238         struct show_raw *rr, **rrp;
239
240         if (!cl->connection)
241             return -1;
242     
243
244         rr = xmalloc(sizeof(*rr));
245         rr->position = position;
246         rr->active = 0;
247         rr->data = data;
248         rr->error_handler = error_handler;
249         rr->record_handler = record_handler;
250         rr->binary = binary;
251         if (syntax)
252             rr->syntax = xstrdup(syntax);
253         else
254             rr->syntax = 0;
255         if (esn)
256             rr->esn = xstrdup(esn);
257         else
258             rr->esn = 0;
259         rr->next = 0;
260         
261         for (rrp = &cl->show_raw; *rrp; rrp = &(*rrp)->next)
262             ;
263         *rrp = rr;
264         
265         if (cl->state == Client_Failed)
266         {
267             client_show_raw_error(cl, "client failed");
268         }
269         else if (cl->state == Client_Disconnected)
270         {
271             client_show_raw_error(cl, "client disconnected");
272         }
273         else
274         {
275             client_send_raw_present(cl);
276         }
277     }
278     return 0;
279 }
280
281 static void client_show_raw_delete(struct show_raw *r)
282 {
283     xfree(r->syntax);
284     xfree(r->esn);
285     xfree(r);
286 }
287
288 void client_show_raw_remove(struct client *cl, void *data)
289 {
290     struct show_raw *rr = data;
291     struct show_raw **rrp = &cl->show_raw;
292     while (*rrp != rr)
293         rrp = &(*rrp)->next;
294     if (*rrp)
295     {
296         *rrp = rr->next;
297         client_show_raw_delete(rr);
298     }
299 }
300
301 void client_show_raw_dequeue(struct client *cl)
302 {
303     struct show_raw *rr = cl->show_raw;
304
305     cl->show_raw = rr->next;
306     client_show_raw_delete(rr);
307 }
308
309 static void client_show_raw_error(struct client *cl, const char *addinfo)
310 {
311     while (cl->show_raw)
312     {
313         cl->show_raw->error_handler(cl->show_raw->data, addinfo);
314         client_show_raw_dequeue(cl);
315     }
316 }
317
318 static void client_send_raw_present(struct client *cl)
319 {
320     struct session_database *sdb = client_get_database(cl);
321     struct connection *co = client_get_connection(cl);
322     ZOOM_resultset set = cl->resultset;
323
324     int offset = cl->show_raw->position;
325     const char *syntax = 0;
326     const char *elements = 0;
327
328     assert(cl->show_raw);
329     assert(set);
330
331     yaz_log(YLOG_DEBUG, "%s: trying to present %d record(s) from %d",
332             client_get_url(cl), 1, offset);
333
334     if (cl->show_raw->syntax)
335         syntax = cl->show_raw->syntax;
336     else
337         syntax = session_setting_oneval(sdb, PZ_REQUESTSYNTAX);
338     ZOOM_resultset_option_set(set, "preferredRecordSyntax", syntax);
339
340     if (cl->show_raw->esn)
341         elements = cl->show_raw->esn;
342     else
343         elements = session_setting_oneval(sdb, PZ_ELEMENTS);
344     if (elements && *elements)
345         ZOOM_resultset_option_set(set, "elementSetName", elements);
346
347     ZOOM_resultset_records(set, 0, offset-1, 1);
348     cl->show_raw->active = 1;
349
350     connection_continue(co);
351 }
352
353 static int nativesyntax_to_type(struct session_database *sdb, char *type,
354                                 ZOOM_record rec)
355 {
356     const char *s = session_setting_oneval(sdb, PZ_NATIVESYNTAX);
357
358     if (s && *s)
359     {
360         if (!strncmp(s, "iso2709", 7))
361         {
362             const char *cp = strchr(s, ';');
363             yaz_snprintf(type, 80, "xml; charset=%s", cp ? cp+1 : "marc-8s");
364         }
365         else if (!strncmp(s, "xml", 3))
366         {
367             strcpy(type, "xml");
368         }
369         else if (!strncmp(s, "txml", 4))
370         {
371             const char *cp = strchr(s, ';');
372             yaz_snprintf(type, 80, "txml; charset=%s", cp ? cp+1 : "marc-8s");
373         }
374         else
375             return -1;
376         return 0;
377     }
378     else  /* attempt to deduce structure */
379     {
380         const char *syntax = ZOOM_record_get(rec, "syntax", NULL);
381         if (syntax)
382         {
383             if (!strcmp(syntax, "XML"))
384             {
385                 strcpy(type, "xml");
386                 return 0;
387             }
388             else if (!strcmp(syntax, "TXML"))
389                 {
390                     strcpy(type, "txml");
391                     return 0;
392                 }
393             else if (!strcmp(syntax, "USmarc") || !strcmp(syntax, "MARC21"))
394             {
395                 strcpy(type, "xml; charset=marc8-s");
396                 return 0;
397             }
398             else return -1;
399         }
400         else return -1;
401     }
402 }
403
404 /**
405  * TODO Consider thread safety!!!
406  *
407  */
408 int client_report_facets(struct client *cl, ZOOM_resultset rs) {
409     int facet_idx;
410     ZOOM_facet_field *facets = ZOOM_resultset_facets(rs);
411     int facet_num;
412     struct session *se = client_get_session(cl);
413     facet_num = ZOOM_resultset_facets_size(rs);
414     yaz_log(YLOG_DEBUG, "client_report_facets: %d", facet_num);
415
416     for (facet_idx = 0; facet_idx < facet_num; facet_idx++) {
417         const char *name = ZOOM_facet_field_name(facets[facet_idx]);
418         size_t term_idx;
419         size_t term_num = ZOOM_facet_field_term_count(facets[facet_idx]);
420         for (term_idx = 0; term_idx < term_num; term_idx++ ) {
421             int freq;
422             const char *term = ZOOM_facet_field_get_term(facets[facet_idx], term_idx, &freq);
423             if (term)
424                 add_facet(se, name, term, freq);
425         }
426     }
427
428     return 0;
429 }
430
431 static void ingest_raw_record(struct client *cl, ZOOM_record rec)
432 {
433     const char *buf;
434     int len;
435     char type[80];
436
437     if (cl->show_raw->binary)
438         strcpy(type, "raw");
439     else
440     {
441         struct session_database *sdb = client_get_database(cl);
442         nativesyntax_to_type(sdb, type, rec);
443     }
444
445     buf = ZOOM_record_get(rec, type, &len);
446     cl->show_raw->record_handler(cl->show_raw->data,  buf, len);
447     client_show_raw_dequeue(cl);
448 }
449
450 void client_check_preferred_watch(struct client *cl)
451 {
452     struct session *se = cl->session;
453     yaz_log(YLOG_DEBUG, "client_check_preferred_watch: %s ", client_get_url(cl));
454     if (se)
455     {
456         client_unlock(cl);
457         if (session_is_preferred_clients_ready(se)) {
458             session_alert_watch(se, SESSION_WATCH_SHOW_PREF);
459         }
460         else
461             yaz_log(YLOG_DEBUG, "client_check_preferred_watch: Still locked on preferred targets.");
462
463         client_lock(cl);
464     }
465     else
466         yaz_log(YLOG_WARN, "client_check_preferred_watch: %s. No session!", client_get_url(cl));
467
468 }
469
470 void client_search_response(struct client *cl)
471 {
472     struct connection *co = cl->connection;
473     struct session *se = cl->session;
474     ZOOM_connection link = connection_get_link(co);
475     ZOOM_resultset resultset = cl->resultset;
476
477     const char *error, *addinfo = 0;
478     
479     if (ZOOM_connection_error(link, &error, &addinfo))
480     {
481         cl->hits = 0;
482         client_set_state(cl, Client_Error);
483         yaz_log(YLOG_WARN, "Search error %s (%s): %s",
484                 error, addinfo, client_get_url(cl));
485     }
486     else
487     {
488         yaz_log(YLOG_DEBUG, "client_search_response: hits "
489                 ODR_INT_PRINTF, cl->hits);
490         client_report_facets(cl, resultset);
491         cl->record_offset = cl->startrecs;
492         cl->hits = ZOOM_resultset_size(resultset);
493         if (se) {
494             se->total_hits += cl->hits;
495             yaz_log(YLOG_DEBUG, "client_search_response: total hits "
496                     ODR_INT_PRINTF, se->total_hits);
497         }
498     }
499 }
500
501 void client_got_records(struct client *cl)
502 {
503     struct session *se = cl->session;
504     if (se)
505     {
506         client_unlock(cl);
507         session_alert_watch(se, SESSION_WATCH_SHOW);
508         session_alert_watch(se, SESSION_WATCH_RECORD);
509         client_lock(cl);
510     }
511 }
512
513 void client_record_response(struct client *cl)
514 {
515     struct connection *co = cl->connection;
516     ZOOM_connection link = connection_get_link(co);
517     ZOOM_resultset resultset = cl->resultset;
518     const char *error, *addinfo;
519
520     if (ZOOM_connection_error(link, &error, &addinfo))
521     {
522         client_set_state(cl, Client_Error);
523         yaz_log(YLOG_WARN, "Search error %s (%s): %s",
524             error, addinfo, client_get_url(cl));
525     }
526     else
527     {
528         ZOOM_record rec = 0;
529         const char *msg, *addinfo;
530         
531         if (cl->show_raw && cl->show_raw->active)
532         {
533             if ((rec = ZOOM_resultset_record(resultset,
534                                              cl->show_raw->position-1)))
535             {
536                 cl->show_raw->active = 0;
537                 ingest_raw_record(cl, rec);
538             }
539             else
540             {
541                 yaz_log(YLOG_WARN, "Expected record, but got NULL, offset=%d",
542                         cl->show_raw->position-1);
543             }
544         }
545         else
546         {
547             int offset = cl->record_offset;
548             if ((rec = ZOOM_resultset_record(resultset, offset)))
549             {
550                 cl->record_offset++;
551                 if (cl->session == 0)
552                     ;
553                 else if (ZOOM_record_error(rec, &msg, &addinfo, 0))
554                 {
555                     yaz_log(YLOG_WARN, "Record error %s (%s): %s (rec #%d)",
556                             msg, addinfo, client_get_url(cl),
557                             cl->record_offset);
558                 }
559                 else
560                 {
561                     struct session_database *sdb = client_get_database(cl);
562                     NMEM nmem = nmem_create();
563                     const char *xmlrec;
564                     char type[80];
565
566                     if (nativesyntax_to_type(sdb, type, rec))
567                         yaz_log(YLOG_WARN, "Failed to determine record type");
568                     xmlrec = ZOOM_record_get(rec, type, NULL);
569                     if (!xmlrec)
570                         yaz_log(YLOG_WARN, "ZOOM_record_get failed from %s",
571                                 client_get_url(cl));
572                     else
573                     {
574                         if (ingest_record(cl, xmlrec, cl->record_offset, nmem))
575                             yaz_log(YLOG_WARN, "Failed to ingest from %s",
576                                     client_get_url(cl));
577                     }
578                     nmem_destroy(nmem);
579                 }
580             }
581             else
582             {
583                 yaz_log(YLOG_WARN, "Expected record, but got NULL, offset=%d",
584                         offset);
585             }
586         }
587     }
588 }
589
590 static int client_set_facets_request(struct client *cl, ZOOM_connection link)
591 {
592     struct session_database *sdb = client_get_database(cl);
593     const char *opt_facet_term_sort  = session_setting_oneval(sdb, PZ_TERMLIST_TERM_SORT);
594     const char *opt_facet_term_count = session_setting_oneval(sdb, PZ_TERMLIST_TERM_COUNT);
595     const char *opt_facet_record_filter = session_setting_oneval(sdb, PZ_RECORDFILTER);
596     /* Disable when no count is set */
597     /* TODO Verify: Do we need to reset the  ZOOM facets if a ZOOM Connection is being reused??? */
598     if (opt_facet_term_count && *opt_facet_term_count)
599     {
600         int index = 0;
601         struct session *session = client_get_session(cl);
602         struct conf_service *service = session->service;
603         int num = service->num_metadata;
604         WRBUF wrbuf = wrbuf_alloc();
605         yaz_log(YLOG_DEBUG, "Facet settings, sort: %s count: %s",
606                 opt_facet_term_sort, opt_facet_term_count);
607         for (index = 0; index < num; index++)
608         {
609             struct conf_metadata *conf_meta = &service->metadata[index];
610             if (conf_meta->termlist)
611             {
612                 if (wrbuf_len(wrbuf))
613                     wrbuf_puts(wrbuf, ", ");
614                 wrbuf_printf(wrbuf, "@attr 1=%s", conf_meta->name);
615                 
616                 if (opt_facet_term_sort && *opt_facet_term_sort)
617                     wrbuf_printf(wrbuf, " @attr 2=%s", opt_facet_term_sort);
618                 wrbuf_printf(wrbuf, " @attr 3=%s", opt_facet_term_count);
619             }
620         }
621         if (wrbuf_len(wrbuf))
622         {
623             yaz_log(YLOG_LOG, "Setting ZOOM facets option: %s", wrbuf_cstr(wrbuf));
624             ZOOM_connection_option_set(link, "facets", wrbuf_cstr(wrbuf));
625             return 1;
626         }
627     }
628     return 0;
629 }
630
631 int client_has_facet(struct client *cl, const char *name) {
632     ZOOM_facet_field facet_field;
633     if (!cl || !cl->resultset || !name) {
634         yaz_log(YLOG_DEBUG, "client has facet: Missing %p %p %s", cl, (cl ? cl->resultset: 0), name);
635         return 0;
636     }
637     facet_field = ZOOM_resultset_get_facet_field(cl->resultset, name);
638     if (facet_field) {
639         yaz_log(YLOG_DEBUG, "client: has facets for %s", name);
640         return 1;
641     }
642     yaz_log(YLOG_DEBUG, "client: No facets for %s", name);
643     return 0;
644 }
645
646
647 void client_start_search(struct client *cl)
648 {
649     struct session_database *sdb = client_get_database(cl);
650     struct connection *co = client_get_connection(cl);
651     ZOOM_connection link = connection_get_link(co);
652     ZOOM_resultset rs;
653     char *databaseName = sdb->database->databases[0];
654     const char *opt_piggyback   = session_setting_oneval(sdb, PZ_PIGGYBACK);
655     const char *opt_queryenc    = session_setting_oneval(sdb, PZ_QUERYENCODING);
656     const char *opt_elements    = session_setting_oneval(sdb, PZ_ELEMENTS);
657     const char *opt_requestsyn  = session_setting_oneval(sdb, PZ_REQUESTSYNTAX);
658     const char *opt_maxrecs     = session_setting_oneval(sdb, PZ_MAXRECS);
659     const char *opt_sru         = session_setting_oneval(sdb, PZ_SRU);
660     const char *opt_sort        = session_setting_oneval(sdb, PZ_SORT);
661     const char *opt_preferred   = session_setting_oneval(sdb, PZ_PREFERRED);
662     char maxrecs_str[24], startrecs_str[24];
663
664     assert(link);
665
666     cl->hits = -1;
667     cl->record_offset = 0;
668     cl->diagnostic = 0;
669
670     if (opt_preferred) {
671         cl->preferred = atoi(opt_preferred);
672         if (cl->preferred)
673             yaz_log(YLOG_LOG, "Target %s has preferred status: %d", sdb->database->url, cl->preferred);
674     }
675     client_set_state(cl, Client_Working);
676
677     if (*opt_piggyback)
678         ZOOM_connection_option_set(link, "piggyback", opt_piggyback);
679     else
680         ZOOM_connection_option_set(link, "piggyback", "1");
681     if (*opt_queryenc)
682         ZOOM_connection_option_set(link, "rpnCharset", opt_queryenc);
683     if (*opt_sru && *opt_elements)
684         ZOOM_connection_option_set(link, "schema", opt_elements);
685     else if (*opt_elements)
686         ZOOM_connection_option_set(link, "elementSetName", opt_elements);
687     if (*opt_requestsyn)
688         ZOOM_connection_option_set(link, "preferredRecordSyntax", opt_requestsyn);
689
690     if (!*opt_maxrecs)
691     {
692         sprintf(maxrecs_str, "%d", cl->maxrecs);
693         opt_maxrecs = maxrecs_str;
694     }
695     ZOOM_connection_option_set(link, "count", opt_maxrecs);
696
697
698     if (atoi(opt_maxrecs) > 20)
699         ZOOM_connection_option_set(link, "presentChunk", "20");
700     else
701         ZOOM_connection_option_set(link, "presentChunk", opt_maxrecs);
702
703     sprintf(startrecs_str, "%d", cl->startrecs);
704     ZOOM_connection_option_set(link, "start", startrecs_str);
705
706     if (databaseName)
707         ZOOM_connection_option_set(link, "databaseName", databaseName);
708
709     /* TODO Verify does it break something for CQL targets(non-SOLR) ? */
710     /* facets definition is in PQF */
711     client_set_facets_request(cl, link);
712
713     if (cl->cqlquery)
714     {
715         ZOOM_query q = ZOOM_query_create();
716         yaz_log(YLOG_LOG, "Search %s CQL: %s", sdb->database->url, cl->cqlquery);
717         ZOOM_query_cql(q, cl->cqlquery);
718         if (*opt_sort)
719             ZOOM_query_sortby(q, opt_sort);
720         rs = ZOOM_connection_search(link, q);
721         ZOOM_query_destroy(q);
722     }
723     else
724     {
725         yaz_log(YLOG_LOG, "Search %s PQF: %s", sdb->database->url, cl->pquery);
726         rs = ZOOM_connection_search_pqf(link, cl->pquery);
727     }
728     ZOOM_resultset_destroy(cl->resultset);
729     cl->resultset = rs;
730     connection_continue(co);
731 }
732
733 struct client *client_create(void)
734 {
735     struct client *cl = xmalloc(sizeof(*cl));
736     cl->maxrecs = 100;
737     cl->startrecs = 0;
738     cl->pquery = 0;
739     cl->cqlquery = 0;
740     cl->database = 0;
741     cl->connection = 0;
742     cl->session = 0;
743     cl->hits = 0;
744     cl->record_offset = 0;
745     cl->diagnostic = 0;
746     cl->state = Client_Disconnected;
747     cl->show_raw = 0;
748     cl->resultset = 0;
749     cl->mutex = 0;
750     pazpar2_mutex_create(&cl->mutex, "client");
751     cl->preferred = 0;
752     cl->ref_count = 1;
753     client_use(1);
754     
755     return cl;
756 }
757
758 void client_lock(struct client *c)
759 {
760     yaz_mutex_enter(c->mutex);
761 }
762
763 void client_unlock(struct client *c)
764 {
765     yaz_mutex_leave(c->mutex);
766 }
767
768 void client_incref(struct client *c)
769 {
770     pazpar2_incref(&c->ref_count, c->mutex);
771     yaz_log(YLOG_DEBUG, "client_incref c=%p %s cnt=%d",
772             c, client_get_url(c), c->ref_count);
773 }
774
775 int client_destroy(struct client *c)
776 {
777     if (c)
778     {
779         yaz_log(YLOG_DEBUG, "client_destroy c=%p %s cnt=%d",
780                 c, client_get_url(c), c->ref_count);
781         if (!pazpar2_decref(&c->ref_count, c->mutex))
782         {
783             xfree(c->pquery);
784             c->pquery = 0;
785             xfree(c->cqlquery);
786             c->cqlquery = 0;
787             assert(!c->connection);
788
789             if (c->resultset)
790             {
791                 ZOOM_resultset_destroy(c->resultset);
792             }
793             yaz_mutex_destroy(&c->mutex);
794             xfree(c);
795             client_use(-1);
796             return 1;
797         }
798     }
799     return 0;
800 }
801
802 void client_set_connection(struct client *cl, struct connection *con)
803 {
804     if (cl->resultset)
805         ZOOM_resultset_release(cl->resultset);
806     if (con)
807     {
808         assert(cl->connection == 0);
809         cl->connection = con;
810         client_incref(cl);
811     }
812     else
813     {
814         cl->connection = con;
815         client_destroy(cl);
816     }
817 }
818
819 void client_disconnect(struct client *cl)
820 {
821     if (cl->state != Client_Idle)
822         client_set_state(cl, Client_Disconnected);
823     client_set_connection(cl, 0);
824 }
825
826 // Extract terms from query into null-terminated termlist
827 static void extract_terms(NMEM nmem, struct ccl_rpn_node *query, char **termlist)
828 {
829     int num = 0;
830
831     pull_terms(nmem, query, termlist, &num);
832     termlist[num] = 0;
833 }
834
835 // Initialize CCL map for a target
836 static CCL_bibset prepare_cclmap(struct client *cl)
837 {
838     struct session_database *sdb = client_get_database(cl);
839     struct setting *s;
840     CCL_bibset res;
841
842     if (!sdb->settings)
843         return 0;
844     res = ccl_qual_mk();
845     for (s = sdb->settings[PZ_CCLMAP]; s; s = s->next)
846     {
847         char *p = strchr(s->name + 3, ':');
848         if (!p)
849         {
850             yaz_log(YLOG_WARN, "Malformed cclmap name: %s", s->name);
851             ccl_qual_rm(&res);
852             return 0;
853         }
854         p++;
855         ccl_qual_fitem(res, s->value, p);
856     }
857     return res;
858 }
859
860 // returns a xmalloced CQL query corresponding to the pquery in client
861 static char *make_cqlquery(struct client *cl)
862 {
863     cql_transform_t cqlt = cql_transform_create();
864     Z_RPNQuery *zquery;
865     char *r;
866     WRBUF wrb = wrbuf_alloc();
867     int status;
868     ODR odr_out = odr_createmem(ODR_ENCODE);
869
870     zquery = p_query_rpn(odr_out, cl->pquery);
871     yaz_log(YLOG_LOG, "PQF: %s", cl->pquery);
872     if ((status = cql_transform_rpn2cql_wrbuf(cqlt, wrb, zquery)))
873     {
874         yaz_log(YLOG_WARN, "Failed to generate CQL query, code=%d", status);
875         r = 0;
876     }
877     else
878     {
879         r = xstrdup(wrbuf_cstr(wrb));
880     }     
881     wrbuf_destroy(wrb);
882     odr_destroy(odr_out);
883     cql_transform_close(cqlt);
884     return r;
885 }
886
887 // returns a xmalloced SOLR query corresponding to the pquery in client
888 // TODO Could prob. be merge with the similar make_cqlquery
889 static char *make_solrquery(struct client *cl)
890 {
891     solr_transform_t sqlt = solr_transform_create();
892     Z_RPNQuery *zquery;
893     char *r;
894     WRBUF wrb = wrbuf_alloc();
895     int status;
896     ODR odr_out = odr_createmem(ODR_ENCODE);
897
898     zquery = p_query_rpn(odr_out, cl->pquery);
899     yaz_log(YLOG_LOG, "PQF: %s", cl->pquery);
900     if ((status = solr_transform_rpn2solr_wrbuf(sqlt, wrb, zquery)))
901     {
902         yaz_log(YLOG_WARN, "Failed to generate SOLR query, code=%d", status);
903         r = 0;
904     }
905     else
906     {
907         r = xstrdup(wrbuf_cstr(wrb));
908     }
909     wrbuf_destroy(wrb);
910     odr_destroy(odr_out);
911     solr_transform_close(sqlt);
912     return r;
913 }
914
915 // Parse the query given the settings specific to this client
916 int client_parse_query(struct client *cl, const char *query)
917 {
918     struct session *se = client_get_session(cl);
919     struct session_database *sdb = client_get_database(cl);
920     struct ccl_rpn_node *cn;
921     int cerror, cpos;
922     CCL_bibset ccl_map = prepare_cclmap(cl);
923     const char *sru = session_setting_oneval(sdb, PZ_SRU);
924     const char *pqf_prefix = session_setting_oneval(sdb, PZ_PQF_PREFIX);
925     const char *pqf_strftime = session_setting_oneval(sdb, PZ_PQF_STRFTIME);
926
927     if (!ccl_map)
928         return -1;
929
930     cn = ccl_find_str(ccl_map, query, &cerror, &cpos);
931     ccl_qual_rm(&ccl_map);
932     if (!cn)
933     {
934         client_set_state(cl, Client_Error);
935         session_log(se, YLOG_WARN, "Failed to parse CCL query '%s' for %s",
936                 query,
937                 client_get_database(cl)->database->url);
938         return -1;
939     }
940     wrbuf_rewind(se->wrbuf);
941     if (*pqf_prefix)
942     {
943         wrbuf_puts(se->wrbuf, pqf_prefix);
944         wrbuf_puts(se->wrbuf, " ");
945     }
946     if (!pqf_strftime || !*pqf_strftime)
947         ccl_pquery(se->wrbuf, cn);
948     else
949     {
950         time_t cur_time = time(0);
951         struct tm *tm =  localtime(&cur_time);
952         char tmp_str[300];
953         const char *cp = tmp_str;
954
955         /* see man strftime(3) for things .. In particular %% gets converted
956          to %.. And That's our original query .. */
957         strftime(tmp_str, sizeof(tmp_str)-1, pqf_strftime, tm);
958         for (; *cp; cp++)
959         {
960             if (cp[0] == '%')
961                 ccl_pquery(se->wrbuf, cn);
962             else
963                 wrbuf_putc(se->wrbuf, cp[0]);
964         }
965     }
966     xfree(cl->pquery);
967     cl->pquery = xstrdup(wrbuf_cstr(se->wrbuf));
968
969     xfree(cl->cqlquery);
970     if (*sru)
971     {
972         if (!strcmp(sru, "solr")) {
973             if (!(cl->cqlquery = make_solrquery(cl)))
974                 return -1;
975         }
976         else {
977             if (!(cl->cqlquery = make_cqlquery(cl)))
978                 return -1;
979         }
980     }
981     else
982         cl->cqlquery = 0;
983
984     /* TODO FIX Not thread safe */
985     if (!se->relevance)
986     {
987         // Initialize relevance structure with query terms
988         char *p[512];
989         extract_terms(se->nmem, cn, p);
990         se->relevance = relevance_create(
991             se->service->relevance_pct,
992             se->nmem, (const char **) p);
993     }
994
995     ccl_rpn_delete(cn);
996     return 0;
997 }
998
999 void client_set_session(struct client *cl, struct session *se)
1000 {
1001     cl->session = se;
1002 }
1003
1004 int client_is_active(struct client *cl)
1005 {
1006     if (cl->connection && (cl->state == Client_Connecting ||
1007                            cl->state == Client_Working))
1008         return 1;
1009     return 0;
1010 }
1011
1012 int client_is_active_preferred(struct client *cl)
1013 {
1014     /* only count if this is a preferred target. */
1015     if (!cl->preferred)
1016         return 0;
1017     /* TODO No sure this the condition that Seb wants */
1018     if (cl->connection && (cl->state == Client_Connecting ||
1019                            cl->state == Client_Working))
1020         return 1;
1021     return 0;
1022 }
1023
1024
1025 Odr_int client_get_hits(struct client *cl)
1026 {
1027     return cl->hits;
1028 }
1029
1030 int client_get_num_records(struct client *cl)
1031 {
1032     return cl->record_offset;
1033 }
1034
1035 void client_set_diagnostic(struct client *cl, int diagnostic)
1036 {
1037     cl->diagnostic = diagnostic;
1038 }
1039
1040 int client_get_diagnostic(struct client *cl)
1041 {
1042     return cl->diagnostic;
1043 }
1044
1045 void client_set_database(struct client *cl, struct session_database *db)
1046 {
1047     cl->database = db;
1048 }
1049
1050 struct host *client_get_host(struct client *cl)
1051 {
1052     return client_get_database(cl)->database->host;
1053 }
1054
1055 const char *client_get_url(struct client *cl)
1056 {
1057     if (cl->database)
1058         return client_get_database(cl)->database->url;
1059     else
1060         return "NOURL";
1061 }
1062
1063 void client_set_maxrecs(struct client *cl, int v)
1064 {
1065     cl->maxrecs = v;
1066 }
1067
1068 int client_get_maxrecs(struct client *cl)
1069 {
1070     return cl->maxrecs;
1071 }
1072
1073 void client_set_startrecs(struct client *cl, int v)
1074 {
1075     cl->startrecs = v;
1076 }
1077
1078 void client_set_preferred(struct client *cl, int v)
1079 {
1080     cl->preferred = v;
1081 }
1082
1083
1084 /*
1085  * Local variables:
1086  * c-basic-offset: 4
1087  * c-file-style: "Stroustrup"
1088  * indent-tabs-mode: nil
1089  * End:
1090  * vim: shiftwidth=4 tabstop=8 expandtab
1091  */
1092