c9ac78f837f7be2404a7af7d41e9664c8363a3cd
[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 static YAZ_MUTEX g_mutex = 0;
71 static int no_clients = 0;
72 static int no_clients_total = 0;
73
74 static int client_use(int delta)
75 {
76     int clients;
77     if (!g_mutex)
78         yaz_mutex_create(&g_mutex);
79     yaz_mutex_enter(g_mutex);
80     no_clients += delta;
81     if (delta > 0)
82         no_clients_total += delta;
83     clients = no_clients;
84     yaz_mutex_leave(g_mutex);
85     yaz_log(YLOG_DEBUG, "%s clients=%d", delta == 0 ? "" : (delta > 0 ? "INC" : "DEC"), clients);
86     return clients;
87 }
88
89 int  clients_count(void) {
90     return client_use(0);
91 }
92
93 int  clients_count_total(void) {
94     int total = 0;
95     if (!g_mutex)
96         return 0;
97     yaz_mutex_enter(g_mutex);
98     total = no_clients_total;
99     yaz_mutex_leave(g_mutex);
100     return total;
101 }
102
103
104 /** \brief Represents client state for a connection to one search target */
105 struct client {
106     struct session_database *database;
107     struct connection *connection;
108     struct session *session;
109     char *pquery; // Current search
110     char *cqlquery; // used for SRU targets only
111     Odr_int hits;
112     int record_offset;
113     int maxrecs;
114     int startrecs;
115     int diagnostic;
116     int preferred;
117     struct suggestions *suggestions;
118     enum client_state state;
119     struct show_raw *show_raw;
120     ZOOM_resultset resultset;
121     YAZ_MUTEX mutex;
122     int ref_count;
123     char *id;
124 };
125
126 struct suggestions {
127     NMEM nmem;
128     int num;
129     char **misspelled;
130     char **suggest;
131     char *passthrough;
132 };
133
134 struct show_raw {
135     int active; // whether this request has been sent to the server
136     int position;
137     int binary;
138     char *syntax;
139     char *esn;
140     void (*error_handler)(void *data, const char *addinfo);
141     void (*record_handler)(void *data, const char *buf, size_t sz);
142     void *data;
143     struct show_raw *next;
144 };
145
146 static const char *client_states[] = {
147     "Client_Connecting",
148     "Client_Idle",
149     "Client_Working",
150     "Client_Error",
151     "Client_Failed",
152     "Client_Disconnected"
153 };
154
155 const char *client_get_state_str(struct client *cl)
156 {
157     return client_states[cl->state];
158 }
159
160 enum client_state client_get_state(struct client *cl)
161 {
162     return cl->state;
163 }
164
165 void client_set_state_nb(struct client *cl, enum client_state st)
166 {
167     cl->state = st;
168 }
169
170 void client_set_state(struct client *cl, enum client_state st)
171 {
172     int was_active = 0;
173     if (client_is_active(cl))
174         was_active = 1;
175     cl->state = st;
176     /* If client is going from being active to inactive and all clients
177        are now idle we fire a watch for the session . The assumption is
178        that session is not mutex locked if client is already active */
179     if (was_active && !client_is_active(cl) && cl->session)
180     {
181
182         int no_active = session_active_clients(cl->session);
183         yaz_log(YLOG_DEBUG, "%s: releasing watches on zero active: %d",
184                 client_get_id(cl), no_active);
185         if (no_active == 0) {
186             session_alert_watch(cl->session, SESSION_WATCH_SHOW);
187             session_alert_watch(cl->session, SESSION_WATCH_SHOW_PREF);
188         }
189     }
190 }
191
192 static void client_show_raw_error(struct client *cl, const char *addinfo);
193
194 struct connection *client_get_connection(struct client *cl)
195 {
196     return cl->connection;
197 }
198
199 struct session_database *client_get_database(struct client *cl)
200 {
201     return cl->database;
202 }
203
204 struct session *client_get_session(struct client *cl)
205 {
206     return cl->session;
207 }
208
209 const char *client_get_pquery(struct client *cl)
210 {
211     return cl->pquery;
212 }
213
214 static void client_send_raw_present(struct client *cl);
215 static int nativesyntax_to_type(struct session_database *sdb, char *type,
216                                 ZOOM_record rec);
217
218 static void client_show_immediate(
219     ZOOM_resultset resultset, struct session_database *sdb, int position,
220     void *data,
221     void (*error_handler)(void *data, const char *addinfo),
222     void (*record_handler)(void *data, const char *buf, size_t sz),
223     int binary)
224 {
225     ZOOM_record rec = 0;
226     char type[80];
227     const char *buf;
228     int len;
229
230     if (!resultset)
231     {
232         error_handler(data, "no resultset");
233         return;
234     }
235     rec = ZOOM_resultset_record(resultset, position-1);
236     if (!rec)
237     {
238         error_handler(data, "no record");
239         return;
240     }
241     if (binary)
242         strcpy(type, "raw");
243     else
244         nativesyntax_to_type(sdb, type, rec);
245     buf = ZOOM_record_get(rec, type, &len);
246     if (!buf)
247     {
248         error_handler(data, "no record");
249         return;
250     }
251     record_handler(data, buf, len);
252 }
253
254
255 int client_show_raw_begin(struct client *cl, int position,
256                           const char *syntax, const char *esn,
257                           void *data,
258                           void (*error_handler)(void *data, const char *addinfo),
259                           void (*record_handler)(void *data, const char *buf,
260                                                  size_t sz),
261                           int binary)
262 {
263     if (syntax == 0 && esn == 0)
264         client_show_immediate(cl->resultset, client_get_database(cl),
265                               position, data,
266                               error_handler, record_handler,
267                               binary);
268     else
269     {
270         struct show_raw *rr, **rrp;
271
272         if (!cl->connection)
273             return -1;
274     
275
276         rr = xmalloc(sizeof(*rr));
277         rr->position = position;
278         rr->active = 0;
279         rr->data = data;
280         rr->error_handler = error_handler;
281         rr->record_handler = record_handler;
282         rr->binary = binary;
283         if (syntax)
284             rr->syntax = xstrdup(syntax);
285         else
286             rr->syntax = 0;
287         if (esn)
288             rr->esn = xstrdup(esn);
289         else
290             rr->esn = 0;
291         rr->next = 0;
292         
293         for (rrp = &cl->show_raw; *rrp; rrp = &(*rrp)->next)
294             ;
295         *rrp = rr;
296         
297         if (cl->state == Client_Failed)
298         {
299             client_show_raw_error(cl, "client failed");
300         }
301         else if (cl->state == Client_Disconnected)
302         {
303             client_show_raw_error(cl, "client disconnected");
304         }
305         else
306         {
307             client_send_raw_present(cl);
308         }
309     }
310     return 0;
311 }
312
313 static void client_show_raw_delete(struct show_raw *r)
314 {
315     xfree(r->syntax);
316     xfree(r->esn);
317     xfree(r);
318 }
319
320 void client_show_raw_remove(struct client *cl, void *data)
321 {
322     struct show_raw *rr = data;
323     struct show_raw **rrp = &cl->show_raw;
324     while (*rrp != rr)
325         rrp = &(*rrp)->next;
326     if (*rrp)
327     {
328         *rrp = rr->next;
329         client_show_raw_delete(rr);
330     }
331 }
332
333 void client_show_raw_dequeue(struct client *cl)
334 {
335     struct show_raw *rr = cl->show_raw;
336
337     cl->show_raw = rr->next;
338     client_show_raw_delete(rr);
339 }
340
341 static void client_show_raw_error(struct client *cl, const char *addinfo)
342 {
343     while (cl->show_raw)
344     {
345         cl->show_raw->error_handler(cl->show_raw->data, addinfo);
346         client_show_raw_dequeue(cl);
347     }
348 }
349
350 static void client_send_raw_present(struct client *cl)
351 {
352     struct session_database *sdb = client_get_database(cl);
353     struct connection *co = client_get_connection(cl);
354     ZOOM_resultset set = cl->resultset;
355
356     int offset = cl->show_raw->position;
357     const char *syntax = 0;
358     const char *elements = 0;
359
360     assert(cl->show_raw);
361     assert(set);
362
363     yaz_log(YLOG_DEBUG, "%s: trying to present %d record(s) from %d",
364             client_get_id(cl), 1, offset);
365
366     if (cl->show_raw->syntax)
367         syntax = cl->show_raw->syntax;
368     else
369         syntax = session_setting_oneval(sdb, PZ_REQUESTSYNTAX);
370     ZOOM_resultset_option_set(set, "preferredRecordSyntax", syntax);
371
372     if (cl->show_raw->esn)
373         elements = cl->show_raw->esn;
374     else
375         elements = session_setting_oneval(sdb, PZ_ELEMENTS);
376     if (elements && *elements)
377         ZOOM_resultset_option_set(set, "elementSetName", elements);
378
379     ZOOM_resultset_records(set, 0, offset-1, 1);
380     cl->show_raw->active = 1;
381
382     connection_continue(co);
383 }
384
385 static int nativesyntax_to_type(struct session_database *sdb, char *type,
386                                 ZOOM_record rec)
387 {
388     const char *s = session_setting_oneval(sdb, PZ_NATIVESYNTAX);
389
390     if (s && *s)
391     {
392         if (!strncmp(s, "iso2709", 7))
393         {
394             const char *cp = strchr(s, ';');
395             yaz_snprintf(type, 80, "xml; charset=%s", cp ? cp+1 : "marc-8s");
396         }
397         else if (!strncmp(s, "xml", 3))
398         {
399             strcpy(type, "xml");
400         }
401         else if (!strncmp(s, "txml", 4))
402         {
403             const char *cp = strchr(s, ';');
404             yaz_snprintf(type, 80, "txml; charset=%s", cp ? cp+1 : "marc-8s");
405         }
406         else
407             return -1;
408         return 0;
409     }
410     else  /* attempt to deduce structure */
411     {
412         const char *syntax = ZOOM_record_get(rec, "syntax", NULL);
413         if (syntax)
414         {
415             if (!strcmp(syntax, "XML"))
416             {
417                 strcpy(type, "xml");
418                 return 0;
419             }
420             else if (!strcmp(syntax, "USmarc") || !strcmp(syntax, "MARC21"))
421             {
422                 strcpy(type, "xml; charset=marc8-s");
423                 return 0;
424             }
425             else return -1;
426         }
427         else return -1;
428     }
429 }
430
431 /**
432  * TODO Consider thread safety!!!
433  *
434  */
435 void client_report_facets(struct client *cl, ZOOM_resultset rs)
436 {
437     struct session_database *sdb = client_get_database(cl);
438     ZOOM_facet_field *facets = ZOOM_resultset_facets(rs);
439
440     if (sdb && facets)
441     {
442         struct session *se = client_get_session(cl);
443         int facet_num = ZOOM_resultset_facets_size(rs);
444         struct setting *s;
445
446         for (s = sdb->settings[PZ_FACETMAP]; s; s = s->next)
447         {
448             const char *p = strchr(s->name + 3, ':');
449             if (p && p[1] && s->value && s->value[0])
450             {
451                 int facet_idx;
452                 p++; /* p now holds logical facet name */
453                 for (facet_idx = 0; facet_idx < facet_num; facet_idx++)
454                 {
455                     const char *native_name =
456                         ZOOM_facet_field_name(facets[facet_idx]);
457                     if (native_name && !strcmp(s->value, native_name))
458                     {
459                         size_t term_idx;
460                         size_t term_num =
461                             ZOOM_facet_field_term_count(facets[facet_idx]);
462                         for (term_idx = 0; term_idx < term_num; term_idx++ )
463                         {
464                             int freq;
465                             const char *term =
466                                 ZOOM_facet_field_get_term(facets[facet_idx],
467                                                           term_idx, &freq);
468                             if (term)
469                                 add_facet(se, p, term, freq);
470                         }
471                         break;
472                     }
473                 }
474             }
475         }
476     }
477 }
478
479 static void ingest_raw_record(struct client *cl, ZOOM_record rec)
480 {
481     const char *buf;
482     int len;
483     char type[80];
484
485     if (cl->show_raw->binary)
486         strcpy(type, "raw");
487     else
488     {
489         struct session_database *sdb = client_get_database(cl);
490         nativesyntax_to_type(sdb, type, rec);
491     }
492
493     buf = ZOOM_record_get(rec, type, &len);
494     cl->show_raw->record_handler(cl->show_raw->data,  buf, len);
495     client_show_raw_dequeue(cl);
496 }
497
498 void client_check_preferred_watch(struct client *cl)
499 {
500     struct session *se = cl->session;
501     yaz_log(YLOG_DEBUG, "client_check_preferred_watch: %s ", client_get_id(cl));
502     if (se)
503     {
504         client_unlock(cl);
505         /* TODO possible threading issue. Session can have been destroyed */
506         if (session_is_preferred_clients_ready(se)) {
507             session_alert_watch(se, SESSION_WATCH_SHOW_PREF);
508         }
509         else
510             yaz_log(YLOG_DEBUG, "client_check_preferred_watch: Still locked on preferred targets.");
511
512         client_lock(cl);
513     }
514     else
515         yaz_log(YLOG_WARN, "client_check_preferred_watch: %s. No session!", client_get_id(cl));
516
517 }
518
519 struct suggestions* client_suggestions_create(const char* suggestions_string);
520 static void client_suggestions_destroy(struct client *cl);
521
522 void client_search_response(struct client *cl)
523 {
524     struct connection *co = cl->connection;
525     ZOOM_connection link = connection_get_link(co);
526     ZOOM_resultset resultset = cl->resultset;
527
528     const char *error, *addinfo = 0;
529     
530     if (ZOOM_connection_error(link, &error, &addinfo))
531     {
532         cl->hits = 0;
533         client_set_state(cl, Client_Error);
534         yaz_log(YLOG_WARN, "Search error %s (%s): %s",
535                 error, addinfo, client_get_id(cl));
536     }
537     else
538     {
539         yaz_log(YLOG_DEBUG, "client_search_response: hits "
540                 ODR_INT_PRINTF, cl->hits);
541         client_report_facets(cl, resultset);
542         cl->record_offset = cl->startrecs;
543         cl->hits = ZOOM_resultset_size(resultset);
544         if (cl->suggestions)
545             client_suggestions_destroy(cl);
546         cl->suggestions = client_suggestions_create(ZOOM_resultset_option_get(resultset, "suggestions"));
547     }
548 }
549
550 void client_got_records(struct client *cl)
551 {
552     struct session *se = cl->session;
553     if (se)
554     {
555         client_unlock(cl);
556         session_alert_watch(se, SESSION_WATCH_SHOW);
557         session_alert_watch(se, SESSION_WATCH_RECORD);
558         client_lock(cl);
559     }
560 }
561
562 static void client_record_ingest(struct client *cl)
563 {
564     const char *msg, *addinfo;
565     ZOOM_record rec = 0;
566     ZOOM_resultset resultset = cl->resultset;
567     int offset = cl->record_offset;
568     if ((rec = ZOOM_resultset_record(resultset, offset)))
569     {
570         cl->record_offset++;
571         if (cl->session == 0)
572             ;
573         else if (ZOOM_record_error(rec, &msg, &addinfo, 0))
574         {
575             yaz_log(YLOG_WARN, "Record error %s (%s): %s (rec #%d)",
576                     msg, addinfo, client_get_id(cl),
577                     cl->record_offset);
578         }
579         else
580         {
581             struct session_database *sdb = client_get_database(cl);
582             NMEM nmem = nmem_create();
583             const char *xmlrec;
584             char type[80];
585             
586             if (nativesyntax_to_type(sdb, type, rec))
587                 yaz_log(YLOG_WARN, "Failed to determine record type");
588             xmlrec = ZOOM_record_get(rec, type, NULL);
589             if (!xmlrec)
590                 yaz_log(YLOG_WARN, "ZOOM_record_get failed from %s",
591                         client_get_id(cl));
592             else
593             {
594                 /* OK = 0, -1 = failure, -2 = Filtered */
595                 if (ingest_record(cl, xmlrec, cl->record_offset, nmem) == -1)
596                     yaz_log(YLOG_WARN, "Failed to ingest from %s", client_get_id(cl));
597             }
598             nmem_destroy(nmem);
599         }
600     }
601     else
602     {
603         yaz_log(YLOG_WARN, "Expected record, but got NULL, offset=%d",
604                 offset);
605     }
606 }
607
608 void client_record_response(struct client *cl)
609 {
610     struct connection *co = cl->connection;
611     ZOOM_connection link = connection_get_link(co);
612     ZOOM_resultset resultset = cl->resultset;
613     const char *error, *addinfo;
614
615     if (ZOOM_connection_error(link, &error, &addinfo))
616     {
617         client_set_state(cl, Client_Error);
618         yaz_log(YLOG_WARN, "Search error %s (%s): %s",
619             error, addinfo, client_get_id(cl));
620     }
621     else
622     {
623         if (cl->show_raw && cl->show_raw->active)
624         {
625             ZOOM_record rec = 0;
626             if ((rec = ZOOM_resultset_record(resultset,
627                                              cl->show_raw->position-1)))
628             {
629                 cl->show_raw->active = 0;
630                 ingest_raw_record(cl, rec);
631             }
632             else
633             {
634                 yaz_log(YLOG_WARN, "Expected record, but got NULL, offset=%d",
635                         cl->show_raw->position-1);
636             }
637         }
638         else
639         {
640             client_record_ingest(cl);
641         }
642     }
643 }
644
645 void client_reingest(struct client *cl)
646 {
647     int i = cl->startrecs;
648     int to = cl->record_offset;
649
650     cl->record_offset = i;
651     for (; i < to; i++)
652         client_record_ingest(cl);
653 }
654
655 static void client_set_facets_request(struct client *cl, ZOOM_connection link)
656 {
657     struct session_database *sdb = client_get_database(cl);
658
659     WRBUF w = wrbuf_alloc();
660     
661     struct setting *s;
662
663     for (s = sdb->settings[PZ_FACETMAP]; s; s = s->next)
664     {
665         const char *p = strchr(s->name + 3, ':');
666         if (!p)
667         {
668             yaz_log(YLOG_WARN, "Malformed facetmap name: %s", s->name);
669         }
670         else if (s->value && s->value[0])
671         {
672             wrbuf_puts(w, "@attr 1=");
673             yaz_encode_pqf_term(w, s->value, strlen(s->value));
674             if (s->next)
675                 wrbuf_puts(w, ",");
676         }
677     }
678     yaz_log(YLOG_LOG, "using facets str: %s", wrbuf_cstr(w));
679     ZOOM_connection_option_set(link, "facets",
680                                wrbuf_len(w) ? wrbuf_cstr(w) : 0);
681     wrbuf_destroy(w);
682 }
683
684 int client_has_facet(struct client *cl, const char *name)
685 {
686     struct session_database *sdb = client_get_database(cl);
687     struct setting *s;
688
689     for (s = sdb->settings[PZ_FACETMAP]; s; s = s->next)
690     {
691         const char *p = strchr(s->name + 3, ':');
692         if (p && !strcmp(name, p + 1))
693             return 1;
694     }
695     return 0;
696 }
697
698 void client_start_search(struct client *cl, const char *sort_strategy_and_spec,
699                          int increasing)
700 {
701     struct session_database *sdb = client_get_database(cl);
702     struct connection *co = client_get_connection(cl);
703     ZOOM_connection link = connection_get_link(co);
704     ZOOM_resultset rs;
705     const char *opt_piggyback   = session_setting_oneval(sdb, PZ_PIGGYBACK);
706     const char *opt_queryenc    = session_setting_oneval(sdb, PZ_QUERYENCODING);
707     const char *opt_elements    = session_setting_oneval(sdb, PZ_ELEMENTS);
708     const char *opt_requestsyn  = session_setting_oneval(sdb, PZ_REQUESTSYNTAX);
709     const char *opt_maxrecs     = session_setting_oneval(sdb, PZ_MAXRECS);
710     const char *opt_sru         = session_setting_oneval(sdb, PZ_SRU);
711     const char *opt_sort        = session_setting_oneval(sdb, PZ_SORT);
712     const char *opt_preferred   = session_setting_oneval(sdb, PZ_PREFERRED);
713     const char *extra_args      = session_setting_oneval(sdb, PZ_EXTRA_ARGS);
714     char maxrecs_str[24], startrecs_str[24];
715     ZOOM_query q;
716
717     assert(link);
718
719     cl->hits = 0;
720     cl->record_offset = 0;
721     cl->diagnostic = 0;
722
723     if (extra_args && *extra_args)
724         ZOOM_connection_option_set(link, "extraArgs", extra_args);
725
726     if (opt_preferred) {
727         cl->preferred = atoi(opt_preferred);
728         if (cl->preferred)
729             yaz_log(YLOG_LOG, "Target %s has preferred status: %d",
730                     client_get_id(cl), cl->preferred);
731     }
732     client_set_state(cl, Client_Working);
733
734     if (*opt_piggyback)
735         ZOOM_connection_option_set(link, "piggyback", opt_piggyback);
736     else
737         ZOOM_connection_option_set(link, "piggyback", "1");
738     if (*opt_queryenc)
739         ZOOM_connection_option_set(link, "rpnCharset", opt_queryenc);
740     if (*opt_sru && *opt_elements)
741         ZOOM_connection_option_set(link, "schema", opt_elements);
742     else if (*opt_elements)
743         ZOOM_connection_option_set(link, "elementSetName", opt_elements);
744     if (*opt_requestsyn)
745         ZOOM_connection_option_set(link, "preferredRecordSyntax", opt_requestsyn);
746
747     if (opt_maxrecs && *opt_maxrecs)
748     {
749         cl->maxrecs = atoi(opt_maxrecs);
750     }
751
752     /* convert back to string representation used in ZOOM API */
753     sprintf(maxrecs_str, "%d", cl->maxrecs);
754     ZOOM_connection_option_set(link, "count", maxrecs_str);
755
756     if (cl->maxrecs > 20)
757         ZOOM_connection_option_set(link, "presentChunk", "20");
758     else
759         ZOOM_connection_option_set(link, "presentChunk", maxrecs_str);
760
761     sprintf(startrecs_str, "%d", cl->startrecs);
762     ZOOM_connection_option_set(link, "start", startrecs_str);
763
764     /* TODO Verify does it break something for CQL targets(non-SOLR) ? */
765     /* facets definition is in PQF */
766     client_set_facets_request(cl, link);
767
768     q = ZOOM_query_create();
769     if (cl->cqlquery)
770     {
771         yaz_log(YLOG_LOG, "Search %s CQL: %s", client_get_id(cl),
772                 cl->cqlquery);
773         ZOOM_query_cql(q, cl->cqlquery);
774         if (*opt_sort)
775             ZOOM_query_sortby(q, opt_sort);
776     }
777     else
778     {
779         yaz_log(YLOG_LOG, "Search %s PQF: %s", client_get_id(cl), cl->pquery);
780         
781         ZOOM_query_prefix(q, cl->pquery);
782     }
783     if (sort_strategy_and_spec &&
784         strlen(sort_strategy_and_spec) < 40 /* spec below */)
785     {
786         char spec[50], *p;
787         strcpy(spec, sort_strategy_and_spec);
788         p = strchr(spec, ':');
789         if (p)
790         {
791             *p++ = '\0'; /* cut the string in two */
792             while (*p == ' ')
793                 p++;
794             if (increasing)
795                 strcat(p, " <");
796             else
797                 strcat(p, " >");
798             yaz_log(YLOG_LOG, "applying %s %s", spec, p);
799             ZOOM_query_sortby2(q, spec, p);
800         }
801     }
802     rs = ZOOM_connection_search(link, q);
803     ZOOM_query_destroy(q);
804     ZOOM_resultset_destroy(cl->resultset);
805     cl->resultset = rs;
806     connection_continue(co);
807 }
808
809 struct client *client_create(const char *id)
810 {
811     struct client *cl = xmalloc(sizeof(*cl));
812     cl->maxrecs = 100;
813     cl->startrecs = 0;
814     cl->pquery = 0;
815     cl->cqlquery = 0;
816     cl->database = 0;
817     cl->connection = 0;
818     cl->session = 0;
819     cl->hits = 0;
820     cl->record_offset = 0;
821     cl->diagnostic = 0;
822     cl->state = Client_Disconnected;
823     cl->show_raw = 0;
824     cl->resultset = 0;
825     cl->suggestions = 0;
826     cl->mutex = 0;
827     pazpar2_mutex_create(&cl->mutex, "client");
828     cl->preferred = 0;
829     cl->ref_count = 1;
830     assert(id);
831     cl->id = xstrdup(id);
832     client_use(1);
833     
834     return cl;
835 }
836
837 void client_lock(struct client *c)
838 {
839     yaz_mutex_enter(c->mutex);
840 }
841
842 void client_unlock(struct client *c)
843 {
844     yaz_mutex_leave(c->mutex);
845 }
846
847 void client_incref(struct client *c)
848 {
849     pazpar2_incref(&c->ref_count, c->mutex);
850     yaz_log(YLOG_DEBUG, "client_incref c=%p %s cnt=%d",
851             c, client_get_id(c), c->ref_count);
852 }
853
854 int client_destroy(struct client *c)
855 {
856     if (c)
857     {
858         yaz_log(YLOG_DEBUG, "client_destroy c=%p %s cnt=%d",
859                 c, client_get_id(c), c->ref_count);
860         if (!pazpar2_decref(&c->ref_count, c->mutex))
861         {
862             xfree(c->pquery);
863             c->pquery = 0;
864             xfree(c->cqlquery);
865             c->cqlquery = 0;
866             xfree(c->id);
867             assert(!c->connection);
868
869             if (c->resultset)
870             {
871                 ZOOM_resultset_destroy(c->resultset);
872             }
873             yaz_mutex_destroy(&c->mutex);
874             xfree(c);
875             client_use(-1);
876             return 1;
877         }
878     }
879     return 0;
880 }
881
882 void client_set_connection(struct client *cl, struct connection *con)
883 {
884     if (cl->resultset)
885         ZOOM_resultset_release(cl->resultset);
886     if (con)
887     {
888         assert(cl->connection == 0);
889         cl->connection = con;
890         client_incref(cl);
891     }
892     else
893     {
894         cl->connection = con;
895         client_destroy(cl);
896     }
897 }
898
899 void client_disconnect(struct client *cl)
900 {
901     if (cl->state != Client_Idle)
902         client_set_state(cl, Client_Disconnected);
903     client_set_connection(cl, 0);
904 }
905
906
907 // Initialize CCL map for a target
908 static CCL_bibset prepare_cclmap(struct client *cl)
909 {
910     struct session_database *sdb = client_get_database(cl);
911     struct setting *s;
912     CCL_bibset res;
913
914     if (!sdb->settings)
915         return 0;
916     res = ccl_qual_mk();
917     for (s = sdb->settings[PZ_CCLMAP]; s; s = s->next)
918     {
919         char *p = strchr(s->name + 3, ':');
920         if (!p)
921         {
922             yaz_log(YLOG_WARN, "Malformed cclmap name: %s", s->name);
923             ccl_qual_rm(&res);
924             return 0;
925         }
926         p++;
927         ccl_qual_fitem(res, s->value, p);
928     }
929     return res;
930 }
931
932 // returns a xmalloced CQL query corresponding to the pquery in client
933 static char *make_cqlquery(struct client *cl)
934 {
935     cql_transform_t cqlt = cql_transform_create();
936     Z_RPNQuery *zquery;
937     char *r;
938     WRBUF wrb = wrbuf_alloc();
939     int status;
940     ODR odr_out = odr_createmem(ODR_ENCODE);
941
942     zquery = p_query_rpn(odr_out, cl->pquery);
943     yaz_log(YLOG_LOG, "PQF: %s", cl->pquery);
944     if ((status = cql_transform_rpn2cql_wrbuf(cqlt, wrb, zquery)))
945     {
946         yaz_log(YLOG_WARN, "Failed to generate CQL query, code=%d", status);
947         r = 0;
948     }
949     else
950     {
951         r = xstrdup(wrbuf_cstr(wrb));
952     }     
953     wrbuf_destroy(wrb);
954     odr_destroy(odr_out);
955     cql_transform_close(cqlt);
956     return r;
957 }
958
959 // returns a xmalloced SOLR query corresponding to the pquery in client
960 // TODO Could prob. be merge with the similar make_cqlquery
961 static char *make_solrquery(struct client *cl)
962 {
963     solr_transform_t sqlt = solr_transform_create();
964     Z_RPNQuery *zquery;
965     char *r;
966     WRBUF wrb = wrbuf_alloc();
967     int status;
968     ODR odr_out = odr_createmem(ODR_ENCODE);
969
970     zquery = p_query_rpn(odr_out, cl->pquery);
971     if (zquery == 0) {
972         yaz_log(YLOG_WARN, "Failed to generate RPN from PQF: %s", cl->pquery);
973         return 0;
974     }
975     yaz_log(YLOG_LOG, "PQF: %s", cl->pquery);
976     if ((status = solr_transform_rpn2solr_wrbuf(sqlt, wrb, zquery)))
977     {
978         yaz_log(YLOG_WARN, "Failed to generate SOLR query from PQF %s, code=%d", cl->pquery, status);
979         r = 0;
980     }
981     else
982     {
983         r = xstrdup(wrbuf_cstr(wrb));
984     }
985     wrbuf_destroy(wrb);
986     odr_destroy(odr_out);
987     solr_transform_close(sqlt);
988     return r;
989 }
990
991 static void apply_limit(struct session_database *sdb,
992                         facet_limits_t facet_limits,
993                         WRBUF w_pqf, WRBUF w_ccl)
994 {
995     int i = 0;
996     const char *name;
997     const char *value;
998     NMEM nmem_tmp = nmem_create();
999     for (i = 0; (name = facet_limits_get(facet_limits, i, &value)); i++)
1000     {
1001         struct setting *s = 0;
1002         
1003         for (s = sdb->settings[PZ_LIMITMAP]; s; s = s->next)
1004         {
1005             const char *p = strchr(s->name + 3, ':');
1006             if (p && !strcmp(p + 1, name) && s->value)
1007             {
1008                 char **values = 0;
1009                 int i, num = 0;
1010                 nmem_strsplit_escape2(nmem_tmp, "|", value, &values,
1011                                       &num, 1, '\\', 1);
1012
1013                 if (!strncmp(s->value, "rpn:", 4))
1014                 {
1015                     const char *pqf = s->value + 4;
1016
1017                     wrbuf_puts(w_pqf, "@and ");
1018                     wrbuf_puts(w_pqf, pqf);
1019                     wrbuf_puts(w_pqf, " ");
1020                     for (i = 0; i < num; i++)
1021                     {
1022                         if (i < num - 1)
1023                             wrbuf_puts(w_pqf, "@or ");
1024                         yaz_encode_pqf_term(w_pqf, values[i],
1025                                             strlen(values[i]));
1026                     }
1027                 }
1028                 else if (!strncmp(s->value, "ccl:", 4))
1029                 {
1030                     const char *ccl = s->value + 4;
1031
1032                     wrbuf_puts(w_ccl, " and (");
1033
1034                     for (i = 0; i < num; i++)
1035                     {
1036                         if (i)
1037                             wrbuf_puts(w_ccl, " or ");
1038                         wrbuf_puts(w_ccl, ccl);
1039                         wrbuf_puts(w_ccl, "=\"");
1040                         wrbuf_puts(w_ccl, values[i]);
1041                         wrbuf_puts(w_ccl, "\"");
1042                     }
1043                     wrbuf_puts(w_ccl, ")");
1044
1045                 }
1046                 break;
1047             }
1048         }
1049         nmem_reset(nmem_tmp);
1050         if (!s)
1051         {
1052             yaz_log(YLOG_WARN, "Target %s: limit %s used, but no limitmap defined",
1053                     (sdb->database ? sdb->database->id : "<no id>"), name);
1054         }
1055     }
1056     nmem_destroy(nmem_tmp);
1057 }
1058                         
1059 // Parse the query given the settings specific to this client
1060 int client_parse_query(struct client *cl, const char *query,
1061                        facet_limits_t facet_limits,
1062                        const char *startrecs, const char *maxrecs)
1063 {
1064     struct session *se = client_get_session(cl);
1065     struct session_database *sdb = client_get_database(cl);
1066     struct ccl_rpn_node *cn;
1067     int cerror, cpos;
1068     CCL_bibset ccl_map = prepare_cclmap(cl);
1069     const char *sru = session_setting_oneval(sdb, PZ_SRU);
1070     const char *pqf_prefix = session_setting_oneval(sdb, PZ_PQF_PREFIX);
1071     const char *pqf_strftime = session_setting_oneval(sdb, PZ_PQF_STRFTIME);
1072     const char *query_syntax = session_setting_oneval(sdb, PZ_QUERY_SYNTAX);
1073     WRBUF w_ccl, w_pqf;
1074     int ret_value = 1;
1075
1076     if (!ccl_map)
1077         return -1;
1078
1079
1080     if (maxrecs && atoi(maxrecs) != cl->maxrecs)
1081     {
1082         ret_value = 0;
1083         cl->maxrecs = atoi(maxrecs);
1084     }
1085
1086     if (startrecs && atoi(startrecs) != cl->startrecs)
1087     {
1088         ret_value = 0;
1089         cl->startrecs = atoi(startrecs);
1090     }
1091
1092     w_ccl = wrbuf_alloc();
1093     wrbuf_puts(w_ccl, query);
1094
1095     w_pqf = wrbuf_alloc();
1096     if (*pqf_prefix)
1097     {
1098         wrbuf_puts(w_pqf, pqf_prefix);
1099         wrbuf_puts(w_pqf, " ");
1100     }
1101
1102     apply_limit(sdb, facet_limits, w_pqf, w_ccl);
1103
1104     yaz_log(YLOG_LOG, "CCL query: %s", wrbuf_cstr(w_ccl));
1105     cn = ccl_find_str(ccl_map, wrbuf_cstr(w_ccl), &cerror, &cpos);
1106     ccl_qual_rm(&ccl_map);
1107     if (!cn)
1108     {
1109         client_set_state(cl, Client_Error);
1110         session_log(se, YLOG_WARN, "Failed to parse CCL query '%s' for %s",
1111                     wrbuf_cstr(w_ccl),
1112                     client_get_id(cl));
1113         wrbuf_destroy(w_ccl);
1114         wrbuf_destroy(w_pqf);
1115         return -1;
1116     }
1117     wrbuf_destroy(w_ccl);
1118
1119     if (!pqf_strftime || !*pqf_strftime)
1120         ccl_pquery(w_pqf, cn);
1121     else
1122     {
1123         time_t cur_time = time(0);
1124         struct tm *tm =  localtime(&cur_time);
1125         char tmp_str[300];
1126         const char *cp = tmp_str;
1127
1128         /* see man strftime(3) for things .. In particular %% gets converted
1129          to %.. And That's our original query .. */
1130         strftime(tmp_str, sizeof(tmp_str)-1, pqf_strftime, tm);
1131         for (; *cp; cp++)
1132         {
1133             if (cp[0] == '%')
1134                 ccl_pquery(w_pqf, cn);
1135             else
1136                 wrbuf_putc(w_pqf, cp[0]);
1137         }
1138     }
1139     if (!cl->pquery || strcmp(cl->pquery, wrbuf_cstr(w_pqf)))
1140     {
1141         xfree(cl->pquery);
1142         cl->pquery = xstrdup(wrbuf_cstr(w_pqf));
1143         ret_value = 0;
1144     }
1145     wrbuf_destroy(w_pqf);
1146
1147     yaz_log(YLOG_LOG, "PQF query: %s", cl->pquery);
1148
1149     xfree(cl->cqlquery);
1150
1151     /* Support for PQF on SRU targets. */
1152     /* TODO Refactor */
1153     yaz_log(YLOG_DEBUG, "Query syntax: %s", query_syntax);
1154     if (strcmp(query_syntax, "pqf") != 0 && *sru)
1155     {
1156         if (!strcmp(sru, "solr")) {
1157             if (!(cl->cqlquery = make_solrquery(cl)))
1158                 return -1;
1159         }
1160         else {
1161             if (!(cl->cqlquery = make_cqlquery(cl)))
1162                 return -1;
1163         }
1164     }
1165     else
1166         cl->cqlquery = 0;
1167
1168     /* TODO FIX Not thread safe */
1169     if (!se->relevance)
1170     {
1171         // Initialize relevance structure with query terms
1172         se->relevance = relevance_create_ccl(
1173             se->service->charsets, se->nmem, cn);
1174     }
1175
1176     ccl_rpn_delete(cn);
1177     return ret_value;
1178 }
1179
1180 void client_set_session(struct client *cl, struct session *se)
1181 {
1182     cl->session = se;
1183 }
1184
1185 int client_is_active(struct client *cl)
1186 {
1187     if (cl->connection && (cl->state == Client_Connecting ||
1188                            cl->state == Client_Working))
1189         return 1;
1190     return 0;
1191 }
1192
1193 int client_is_active_preferred(struct client *cl)
1194 {
1195     /* only count if this is a preferred target. */
1196     if (!cl->preferred)
1197         return 0;
1198     /* TODO No sure this the condition that Seb wants */
1199     if (cl->connection && (cl->state == Client_Connecting ||
1200                            cl->state == Client_Working))
1201         return 1;
1202     return 0;
1203 }
1204
1205 Odr_int client_get_hits(struct client *cl)
1206 {
1207     return cl->hits;
1208 }
1209
1210 int client_get_num_records(struct client *cl)
1211 {
1212     return cl->record_offset;
1213 }
1214
1215 void client_set_diagnostic(struct client *cl, int diagnostic)
1216 {
1217     cl->diagnostic = diagnostic;
1218 }
1219
1220 int client_get_diagnostic(struct client *cl)
1221 {
1222     return cl->diagnostic;
1223 }
1224
1225 const char * client_get_suggestions_xml(struct client *cl, WRBUF wrbuf)
1226 {
1227     /* int idx; */
1228     struct suggestions *suggestions = cl->suggestions;
1229
1230     if (!suggestions) {
1231         yaz_log(YLOG_DEBUG, "No suggestions found");
1232         return "";
1233     }
1234     if (suggestions->passthrough) {
1235         yaz_log(YLOG_DEBUG, "Passthrough Suggestions: \n%s\n", suggestions->passthrough);
1236         return suggestions->passthrough;
1237     }
1238     if (suggestions->num == 0) {
1239         return "";
1240     }
1241     /*
1242     for (idx = 0; idx < suggestions->num; idx++) {
1243         wrbuf_printf(wrbuf, "<suggest term=\"%s\"", suggestions->suggest[idx]);
1244         if (suggestions->misspelled[idx] && suggestions->misspelled[idx]) {
1245             wrbuf_puts(wrbuf, suggestions->misspelled[idx]);
1246             wrbuf_puts(wrbuf, "</suggest>\n");
1247         }
1248         else
1249             wrbuf_puts(wrbuf, "/>\n");
1250     }
1251     */
1252     return wrbuf_cstr(wrbuf);
1253 }
1254
1255
1256 void client_set_database(struct client *cl, struct session_database *db)
1257 {
1258     cl->database = db;
1259 }
1260
1261 const char *client_get_id(struct client *cl)
1262 {
1263     return cl->id;
1264 }
1265
1266 int client_get_maxrecs(struct client *cl)
1267 {
1268     return cl->maxrecs;
1269 }
1270
1271 void client_set_preferred(struct client *cl, int v)
1272 {
1273     cl->preferred = v;
1274 }
1275
1276
1277 struct suggestions* client_suggestions_create(const char* suggestions_string)
1278 {
1279     int i;
1280     NMEM nmem;
1281     struct suggestions *suggestions;
1282     if (suggestions_string == 0)
1283         return 0;
1284     nmem = nmem_create();
1285     suggestions = nmem_malloc(nmem, sizeof(*suggestions));
1286     yaz_log(YLOG_DEBUG, "client target suggestions: %s", suggestions_string);
1287
1288     suggestions->nmem = nmem;
1289     suggestions->num = 0;
1290     suggestions->misspelled = 0;
1291     suggestions->suggest = 0;
1292     suggestions->passthrough = nmem_strdup_null(nmem, suggestions_string);
1293
1294     if (suggestions_string)
1295         nmem_strsplit_escape2(suggestions->nmem, "\n", suggestions_string, &suggestions->suggest,
1296                               &suggestions->num, 1, '\\', 0);
1297     /* Set up misspelled array */
1298     suggestions->misspelled = (char **) nmem_malloc(nmem, suggestions->num * sizeof(**suggestions->misspelled));
1299     /* replace = with \0 .. for each item */
1300     for (i = 0; i < suggestions->num; i++)
1301     {
1302         char *cp = strchr(suggestions->suggest[i], '=');
1303         if (cp) {
1304             *cp = '\0';
1305             suggestions->misspelled[i] = cp+1;
1306         }
1307     }
1308     return suggestions;
1309 }
1310
1311 static void client_suggestions_destroy(struct client *cl)
1312 {
1313     NMEM nmem = cl->suggestions->nmem;
1314     cl->suggestions = 0;
1315     nmem_destroy(nmem);
1316 }
1317
1318 /*
1319  * Local variables:
1320  * c-basic-offset: 4
1321  * c-file-style: "Stroustrup"
1322  * indent-tabs-mode: nil
1323  * End:
1324  * vim: shiftwidth=4 tabstop=8 expandtab
1325  */
1326