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